core

package
v0.0.6 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 26, 2026 License: MIT Imports: 32 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultMaxHistoryRecords = 1000
View Source 
const VirtualShadowPrefix = "virtual-"

Variables

View Source 
var (
	ErrActorStopped          = errors.New("actor is stopped")
	ErrMailboxFull           = errors.New("actor mailbox is full")
	ErrInvalidMessage        = errors.New("invalid message type")
	ErrDeviceNotFound        = errors.New("device not found")
	ErrPointNotFound         = errors.New("point not found")
	ErrChannelNotFound       = errors.New("channel not found")
	ErrTimeout               = errors.New("operation timeout")
	ErrInvalidState          = errors.New("invalid state transition")
	ErrConnectionUnavailable = errors.New("connection unavailable")
)
View Source 
var (
	ErrDriverNotFound = errors.New("driver not found")
	ErrQueueFull      = errors.New("queue full")
	ErrRateLimited    = errors.New("rate limited")
)
View Source 
var (
	MaxGlobalReconnectRate = 10
)

Functions

func GenerateModbusRegisterPoints 

func GenerateModbusRegisterPoints(existing []model.Point, opts ModbusRegisterGenOptions, mergeExisting bool) []model.Point

GenerateModbusRegisterPoints 按地址区间生成保持/输入等寄存器点位。 mergeExisting 为 true 时保留同 ID 现有点位配置。

func IsVirtualShadowID 

func IsVirtualShadowID(shadowDeviceID string) bool

func ParseAutoPointsRange 

func ParseAutoPointsRange(rng string) (start, end int, ok bool)

ParseAutoPointsRange 解析 auto_points_range 配置（如 "0-199"）。

func VirtualShadowID 

func VirtualShadowID(virtualDeviceID string) string

Types

type BACnetProtocolAdapter 

type BACnetProtocolAdapter struct{}

BACnetProtocolAdapter BACnet协议适配器

func NewBACnetProtocolAdapter 

func NewBACnetProtocolAdapter() *BACnetProtocolAdapter

NewBACnetProtocolAdapter 创建BACnet协议适配器

func (*BACnetProtocolAdapter) AdjustParameters 

func (a *BACnetProtocolAdapter) AdjustParameters(deviceID string, params map[string]interface{}) map[string]interface{}

AdjustParameters 调整参数

func (*BACnetProtocolAdapter) GetDefaultParameters 

func (a *BACnetProtocolAdapter) GetDefaultParameters() map[string]interface{}

GetDefaultParameters 获取默认参数

func (*BACnetProtocolAdapter) GetProtocolType 

func (a *BACnetProtocolAdapter) GetProtocolType() string

GetProtocolType 获取协议类型

func (*BACnetProtocolAdapter) ValidateParameters 

func (a *BACnetProtocolAdapter) ValidateParameters(params map[string]interface{}) error

ValidateParameters 验证参数

type BackpressureController 

type BackpressureController struct {
	// contains filtered or unexported fields
}

func NewBackpressureController 

func NewBackpressureController(globalLimit int, rate float64) *BackpressureController

func (*BackpressureController) Allow 

func (bc *BackpressureController) Allow(deviceKey string, deviceLimit int) bool

func (*BackpressureController) Release 

func (bc *BackpressureController) Release(deviceKey string)

type BatchAddModbusSlavesResult 

type BatchAddModbusSlavesResult struct {
	Created []model.Device
	Skipped []int // slave_id
	Errors  []string
}

BatchAddModbusSlavesResult 批量添加 Modbus 从站结果。

type ChannelManager 

type ChannelManager struct {
	// contains filtered or unexported fields
}

ChannelManager 管理所有采集通道及其下的设备

func NewChannelManager 

func NewChannelManager(pipeline *DataPipeline, saveFunc func([]model.Channel) error) *ChannelManager

func (*ChannelManager) AddChannel 

func (cm *ChannelManager) AddChannel(ch *model.Channel) error

AddChannel 添加一个采集通道

func (*ChannelManager) AddDevice 

func (cm *ChannelManager) AddDevice(channelID string, dev *model.Device) error

AddDevice 添加设备到通道

func (*ChannelManager) AddPoint 

func (cm *ChannelManager) AddPoint(channelID, deviceID string, point *model.Point) error

AddPoint 添加点位到设备

func (*ChannelManager) AddPoints 

func (cm *ChannelManager) AddPoints(channelID, deviceID string, points []model.Point) error

AddPoints 批量添加点位到设备（单次重启）

func (*ChannelManager) BatchAddModbusSlaves 

func (cm *ChannelManager) BatchAddModbusSlaves(channelID string, slaveStart, slaveEnd, regStart, regEnd int, interval model.Duration, enable bool, datatype, readwrite string, regType model.RegisterType, fc byte) (*BatchAddModbusSlavesResult, error)

BatchAddModbusSlaves 批量添加 Modbus 从站（单次持久化，跳过已存在设备）。

func (*ChannelManager) GenerateDeviceRegisterPoints 

func (cm *ChannelManager) GenerateDeviceRegisterPoints(channelID, deviceID string, opts ModbusRegisterGenOptions, mode string) (*model.Device, error)

GenerateDeviceRegisterPoints 为设备批量生成 Modbus 寄存器点位并持久化。 mode: merge（保留同 ID 现有点）| replace（仅保留新区间点位）。

func (*ChannelManager) GetAllPoints 

func (cm *ChannelManager) GetAllPoints() []map[string]any

func (*ChannelManager) GetChannel 

func (cm *ChannelManager) GetChannel(channelID string) *model.Channel

GetChannel 获取指定通道

func (*ChannelManager) GetChannelDevices 

func (cm *ChannelManager) GetChannelDevices(channelID string) []model.Device

GetChannelDevices 获取指定通道的所有设备

func (*ChannelManager) GetChannelStats 

func (cm *ChannelManager) GetChannelStats() []ChannelStatus

func (*ChannelManager) GetChannels 

func (cm *ChannelManager) GetChannels() []model.Channel

GetChannels 获取所有通道

func (*ChannelManager) GetDevice 

func (cm *ChannelManager) GetDevice(channelID, deviceID string) *model.Device

GetDevice 获取指定通道下的指定设备

func (*ChannelManager) GetDeviceDiagnostics 

func (cm *ChannelManager) GetDeviceDiagnostics(deviceID string) map[string]any

func (*ChannelManager) GetDevicePoints 

func (cm *ChannelManager) GetDevicePoints(channelID, deviceID string) ([]model.PointData, error)

GetDevicePoints 获取指定设备的所有点位数据

func (*ChannelManager) GetDriver 

func (cm *ChannelManager) GetDriver(channelID string) drv.Driver

GetDriver 获取通道的驱动实例

func (*ChannelManager) GetScanEngineMetricsSnapshot 

func (cm *ChannelManager) GetScanEngineMetricsSnapshot() map[string]any

func (*ChannelManager) GetShadowPoint 

func (cm *ChannelManager) GetShadowPoint(channelID, deviceID, pointID string) (*model.ShadowPoint, error)

GetShadowPoint 从影子设备实时快照中读取单个点位数据。 供 OPC UA Server 的 ReadHandler 调用，使第三方客户端能按需获取实时值。

func (*ChannelManager) GetStateManager 

func (cm *ChannelManager) GetStateManager() *CommunicationManageTemplate

GetStateManager 获取状态管理器

func (*ChannelManager) GetTagRegistry 

func (cm *ChannelManager) GetTagRegistry() *TagRegistry

func (*ChannelManager) ReadPoint 

func (cm *ChannelManager) ReadPoint(channelID, deviceID, pointID string) (model.Value, error)

ReadPoint 读取指定通道下设备点位的值

func (*ChannelManager) RemoveChannel 

func (cm *ChannelManager) RemoveChannel(channelID string) error

RemoveChannel 删除采集通道

func (*ChannelManager) RemoveDevice 

func (cm *ChannelManager) RemoveDevice(channelID, deviceID string) error

RemoveDevice 删除设备

func (*ChannelManager) RemoveDevices 

func (cm *ChannelManager) RemoveDevices(channelID string, deviceIDs []string) error

RemoveDevices 批量删除设备

func (*ChannelManager) RemovePoint 

func (cm *ChannelManager) RemovePoint(channelID, deviceID, pointID string) error

RemovePoint 删除设备点位

func (*ChannelManager) RemovePoints 

func (cm *ChannelManager) RemovePoints(channelID, deviceID string, pointIDs []string) error

RemovePoints 批量删除设备点位

func (*ChannelManager) ScanChannel 

func (cm *ChannelManager) ScanChannel(channelID string, params map[string]any) (any, error)

ScanChannel 扫描通道下的设备

func (*ChannelManager) ScanDevice 

func (cm *ChannelManager) ScanDevice(channelID, deviceID string, params map[string]any) (any, error)

ScanDevice 扫描设备下的对象（点位）

func (*ChannelManager) SetShadowCore 

func (cm *ChannelManager) SetShadowCore(sc *ShadowCore)

func (*ChannelManager) SetStatusHandler 

func (cm *ChannelManager) SetStatusHandler(h func(deviceID string, status int))

func (*ChannelManager) SetTopologyChangeHandler 

func (cm *ChannelManager) SetTopologyChangeHandler(h func())

SetTopologyChangeHandler registers a callback invoked when channels/devices/points change. Used to rebuild northbound OPC UA address space.

func (*ChannelManager) Shutdown 

func (cm *ChannelManager) Shutdown()

Shutdown 关闭所有通道

func (*ChannelManager) StartChannel 

func (cm *ChannelManager) StartChannel(channelID string) error

StartChannel 启动一个采集通道

func (*ChannelManager) StopChannel 

func (cm *ChannelManager) StopChannel(channelID string) error

StopChannel 停止一个采集通道

func (*ChannelManager) UpdateChannel 

func (cm *ChannelManager) UpdateChannel(ch *model.Channel) error

UpdateChannel 更新采集通道

func (*ChannelManager) UpdateDevice 

func (cm *ChannelManager) UpdateDevice(channelID string, dev *model.Device) error

UpdateDevice 更新通道下的设备

func (*ChannelManager) UpdatePoint 

func (cm *ChannelManager) UpdatePoint(channelID, deviceID string, point *model.Point) (deviceRestarted bool, err error)

UpdatePoint 更新设备点位。返回值 deviceRestarted 表示是否重启了南向采集设备； 北向 OPC UA 地址空间会在保存后异步重建（仅重启 OPC UA 服务，不重启网关主进程）。

func (*ChannelManager) WritePoint 

func (cm *ChannelManager) WritePoint(channelID, deviceID, pointID string, value any) error

WritePoint 写入指定通道下设备点位的值

type ChannelStatus 

type ChannelStatus struct {
	ID              string                `json:"id"`
	Name            string                `json:"name"`
	Protocol        string                `json:"protocol"`
	Status          string                `json:"status"`
	Enable          bool                  `json:"enable"`
	DeviceCount     int                   `json:"device_count"`
	OnlineCount     int                   `json:"online_count"`
	OfflineCount    int                   `json:"offline_count"`
	QualityScore    int                   `json:"qualityScore"`      // 质量评分
	SuccessRate     float64               `json:"successRate"`       // 成功率
	LastCollectTime string                `json:"last_collect_time"` // 最后采集时间
	Metrics         *model.ChannelMetrics `json:"metrics,omitempty"` // 详细指标
}

type CollectContext 

type CollectContext struct {
	TotalCmd   int  // 总命令数：本次采集执行的总命令数量
	SuccessCmd int  // 成功命令数：成功执行的命令数量
	FailCmd    int  // 失败命令数：执行失败的命令数量
	PanicOccur bool // 是否发生panic：标记采集过程中是否出现异常
}

==================== 采集上下文结构 ==================== CollectContext 用于记录单次采集过程中的统计信息

func (*CollectContext) MarkFail 

func (ctx *CollectContext) MarkFail()

MarkFail 记录一次失败命令 调用此方法会增加失败计数，用于统计采集成功率

func (*CollectContext) MarkSuccess 

func (ctx *CollectContext) MarkSuccess()

MarkSuccess 记录一次成功命令 调用此方法会增加成功计数，用于统计采集成功率

type CollectFinalizeFunc 

type CollectFinalizeFunc func(deviceID string, result *ExecuteResult)

CollectFinalizeFunc 采集完成后回写设备通信状态。

type CommunicationManageTemplate 

type CommunicationManageTemplate struct {
	OnStateChange StateChangeCallback // 状态变更回调
	// contains filtered or unexported fields
}

func NewCommunicationManageTemplate 

func NewCommunicationManageTemplate() *CommunicationManageTemplate

NewCommunicationManageTemplate 创建新的通信管理器

func (*CommunicationManageTemplate) FinalizeCollect 

func (c *CommunicationManageTemplate) FinalizeCollect(node *DeviceNodeTemplate, ctx *CollectContext)

FinalizeCollect 最终裁决函数 根据采集上下文统计信息，决定本次采集的整体结果并更新节点状态

参数:

  • node: 设备节点
  • ctx: 采集上下文，包含本次采集的统计信息

裁决逻辑:

  1. 发生panic -> 直接判定为失败
  2. 无任何命令交互 -> 判定为失败
  3. 成功率 ≥ 30% -> 判定为成功
  4. 成功率 < 30% -> 判定为失败

设计原则:

  1. panic具有最高优先级，直接否决
  2. 允许部分失败（30%成功率即可），适应工业现场不稳定性
  3. 无交互视为最严重失败

func (*CommunicationManageTemplate) GetNode 

func (c *CommunicationManageTemplate) GetNode(deviceID string) *DeviceNodeTemplate

GetNode 获取指定的设备节点

func (*CommunicationManageTemplate) RegisterNode 

func (c *CommunicationManageTemplate) RegisterNode(deviceID, name string) *DeviceNodeTemplate

RegisterNode 注册一个新的设备节点

func (*CommunicationManageTemplate) ShouldCollect 

func (c *CommunicationManageTemplate) ShouldCollect(node *DeviceNodeTemplate) bool

ShouldCollect 判断是否允许对指定节点进行本轮采集 根据节点当前状态和退避时间决定是否执行采集

参数:

  • node: 目标设备节点

返回值:

  • bool: true表示允许采集，false表示跳过采集

策略说明:

  • Online/Unstable状态: 始终允许采集
  • Offline/Quarantine状态: 只有在退避时间过后才允许采集

type ConnState 

type ConnState int
const (
	ConnStateDisconnected ConnState = iota
	ConnStateConnecting
	ConnStateConnected
	ConnStateRetrying
	ConnStateDead
	ConnStateHealthy
	ConnStateDegraded
)

func (ConnState) String 

func (s ConnState) String() string

type ConnectionController 

type ConnectionController struct {
	// contains filtered or unexported fields
}

func NewConnectionController 

func NewConnectionController(driverName, deviceID, protocol string) *ConnectionController

func (*ConnectionController) AttemptHalfOpen 

func (cc *ConnectionController) AttemptHalfOpen(success bool)

func (*ConnectionController) CanRetry 

func (cc *ConnectionController) CanRetry() (canRetry bool, waitTime time.Duration)

func (*ConnectionController) GetState 

func (cc *ConnectionController) GetState() ConnState

func (*ConnectionController) GetStatus 

func (cc *ConnectionController) GetStatus() (state ConnState, retryCount int, maxRetries int, coolDownRemaining time.Duration, lastSuccess time.Time, readFailCount int, connectionFailCount int)

func (*ConnectionController) IsConnectionFailure 

func (cc *ConnectionController) IsConnectionFailure(err error) bool

func (*ConnectionController) IsReadFailure 

func (cc *ConnectionController) IsReadFailure(err error) bool

func (*ConnectionController) RecordConnectionFailure 

func (cc *ConnectionController) RecordConnectionFailure() (shouldRetry bool, backoff time.Duration)

func (*ConnectionController) RecordConnectionSuccess 

func (cc *ConnectionController) RecordConnectionSuccess()

func (*ConnectionController) RecordReadFailure 

func (cc *ConnectionController) RecordReadFailure()

func (*ConnectionController) RecordReadSuccess 

func (cc *ConnectionController) RecordReadSuccess()

func (*ConnectionController) Reset 

func (cc *ConnectionController) Reset()

func (*ConnectionController) SetBackoffParams 

func (cc *ConnectionController) SetBackoffParams(base, max time.Duration, factor float64)

func (*ConnectionController) SetMaxFailCount 

func (cc *ConnectionController) SetMaxFailCount(max int)

func (*ConnectionController) SetMaxRetries 

func (cc *ConnectionController) SetMaxRetries(max int)

func (*ConnectionController) SetState 

func (cc *ConnectionController) SetState(state ConnState)

type DataPipeline 

type DataPipeline struct {
	// contains filtered or unexported fields
}

DataPipeline handles the flow of collected data

func NewDataPipeline 

func NewDataPipeline(bufferSize int) *DataPipeline

func (*DataPipeline) AddBatchHandler 

func (dp *DataPipeline) AddBatchHandler(h func([]model.Value))

AddBatchHandler 注册批量处理器，一次 drain 调用一次，减少 handler 开销。

func (*DataPipeline) AddHandler 

func (dp *DataPipeline) AddHandler(h func(model.Value))

func (*DataPipeline) Push 

func (dp *DataPipeline) Push(val model.Value)

func (*DataPipeline) PushBatch 

func (dp *DataPipeline) PushBatch(vals []model.Value)

func (*DataPipeline) SetShadowIngress 

func (dp *DataPipeline) SetShadowIngress(si *ShadowIngress)

func (*DataPipeline) Start 

func (dp *DataPipeline) Start()

type DefaultDeviceAdapter 

type DefaultDeviceAdapter struct {
	// contains filtered or unexported fields
}

DefaultDeviceAdapter 默认设备适配器

func NewDefaultDeviceAdapter 

func NewDefaultDeviceAdapter(rttManager *RTTManager, mtuManager *MTUManager, gapOptimizer *GapOptimizer, manager *DeviceAdapterManager) *DefaultDeviceAdapter

NewDefaultDeviceAdapter 创建默认设备适配器

func (*DefaultDeviceAdapter) CalculateStabilityScore 

func (a *DefaultDeviceAdapter) CalculateStabilityScore(deviceID string) (float64, error)

CalculateStabilityScore 计算稳定性评分

func (*DefaultDeviceAdapter) GetDeviceProfile 

func (a *DefaultDeviceAdapter) GetDeviceProfile(deviceID string) (*model.DeviceCommunicationProfile, error)

GetDeviceProfile 获取设备画像

func (*DefaultDeviceAdapter) OptimizeCollectionParams 

func (a *DefaultDeviceAdapter) OptimizeCollectionParams(deviceID string) (map[string]interface{}, error)

OptimizeCollectionParams 优化采集参数

func (*DefaultDeviceAdapter) UpdateDeviceProfile 

func (a *DefaultDeviceAdapter) UpdateDeviceProfile(profile *model.DeviceCommunicationProfile) error

UpdateDeviceProfile 更新设备画像

func (*DefaultDeviceAdapter) UpdateGap 

func (a *DefaultDeviceAdapter) UpdateGap(deviceID string, gap int) error

UpdateGap 更新Gap

func (*DefaultDeviceAdapter) UpdateMTU 

func (a *DefaultDeviceAdapter) UpdateMTU(deviceID string, mtu int) error

UpdateMTU 更新MTU

func (*DefaultDeviceAdapter) UpdateRTT 

func (a *DefaultDeviceAdapter) UpdateRTT(deviceID string, rtt int64) error

UpdateRTT 更新RTT

type DefaultProtocolAdapter 

type DefaultProtocolAdapter struct{}

DefaultProtocolAdapter 默认协议适配器

func NewDefaultProtocolAdapter 

func NewDefaultProtocolAdapter() *DefaultProtocolAdapter

NewDefaultProtocolAdapter 创建默认协议适配器

func (*DefaultProtocolAdapter) AdjustParameters 

func (a *DefaultProtocolAdapter) AdjustParameters(deviceID string, params map[string]interface{}) map[string]interface{}

AdjustParameters 调整参数

func (*DefaultProtocolAdapter) GetDefaultParameters 

func (a *DefaultProtocolAdapter) GetDefaultParameters() map[string]interface{}

GetDefaultParameters 获取默认参数

func (*DefaultProtocolAdapter) GetProtocolType 

func (a *DefaultProtocolAdapter) GetProtocolType() string

GetProtocolType 获取协议类型

func (*DefaultProtocolAdapter) ValidateParameters 

func (a *DefaultProtocolAdapter) ValidateParameters(params map[string]interface{}) error

ValidateParameters 验证参数

type DeviceAdapter 

type DeviceAdapter interface {
	GetDeviceProfile(deviceID string) (*model.DeviceCommunicationProfile, error)
	UpdateDeviceProfile(profile *model.DeviceCommunicationProfile) error
	OptimizeCollectionParams(deviceID string) (map[string]interface{}, error)
	UpdateRTT(deviceID string, rtt int64) error
	UpdateMTU(deviceID string, mtu int) error
	UpdateGap(deviceID string, gap int) error
	CalculateStabilityScore(deviceID string) (float64, error)
}

DeviceAdapter 设备适配器接口

type DeviceAdapterManager 

type DeviceAdapterManager struct {
	// contains filtered or unexported fields
}

DeviceAdapterManager 设备适配器管理器

func NewDeviceAdapterManager 

func NewDeviceAdapterManager() *DeviceAdapterManager

NewDeviceAdapterManager 创建设备适配器管理器

func (*DeviceAdapterManager) GetAdapter 

func (m *DeviceAdapterManager) GetAdapter(deviceID string) DeviceAdapter

GetAdapter 获取设备适配器

func (*DeviceAdapterManager) GetDeviceProfile 

func (m *DeviceAdapterManager) GetDeviceProfile(deviceID string) (*model.DeviceCommunicationProfile, error)

GetDeviceProfile 获取设备画像

func (*DeviceAdapterManager) RegisterAdapter 

func (m *DeviceAdapterManager) RegisterAdapter(deviceID string, adapter DeviceAdapter)

RegisterAdapter 注册设备适配器

func (*DeviceAdapterManager) UpdateDeviceProfile 

func (m *DeviceAdapterManager) UpdateDeviceProfile(profile *model.DeviceCommunicationProfile) error

UpdateDeviceProfile 更新设备画像

type DeviceIO 

type DeviceIO interface {
	WritePoint(channelID, deviceID, pointID string, value any) error
	ReadPoint(channelID, deviceID, pointID string) (model.Value, error)
}

type DeviceIOProfile 

type DeviceIOProfile struct {
	Gap       int
	BatchSize int
}

DeviceIOProfile Modbus 等协议的动态 IO 参数（来自 Shadow 通信画像）。

type DeviceNodeTemplate 

type DeviceNodeTemplate struct {
	DeviceID string            // 设备ID
	Name     string            // 设备名称
	Runtime  *NodeRuntimeState // 运行时状态
}

==================== 设备节点模板 ==================== DeviceNodeTemplate 代表一个设备节点，包含设备信息和运行时状态

type DeviceStorageManager 

type DeviceStorageManager struct {
	// contains filtered or unexported fields
}

DeviceStorageManager 从 ShadowCore 读取影子设备全量点位，按策略写入 device_history_*。

func NewDeviceStorageManager 

func NewDeviceStorageManager(s *storage.Storage, dp *DataPipeline) *DeviceStorageManager

func (*DeviceStorageManager) ClearAllHistory 

func (m *DeviceStorageManager) ClearAllHistory()

func (*DeviceStorageManager) GetHistory 

func (m *DeviceStorageManager) GetHistory(deviceID string, limit int) ([]map[string]any, error)

func (*DeviceStorageManager) GetHistoryByTimeRange 

func (m *DeviceStorageManager) GetHistoryByTimeRange(deviceID string, start, end time.Time, limit int) ([]map[string]any, error)

func (*DeviceStorageManager) RemoveDevice 

func (m *DeviceStorageManager) RemoveDevice(deviceID string)

func (*DeviceStorageManager) SetShadowCore 

func (m *DeviceStorageManager) SetShadowCore(sc *ShadowCore)

SetShadowCore 绑定影子设备核心，快照数据从影子设备全量点位读取。

func (*DeviceStorageManager) SetStorage 

func (m *DeviceStorageManager) SetStorage(s *storage.Storage)

SetStorage 在安装完成后绑定运行时存储（采集值落库）。

func (*DeviceStorageManager) UpdateDeviceConfig 

func (m *DeviceStorageManager) UpdateDeviceConfig(deviceID string, cfg model.DeviceStorage)

UpdateDeviceConfig updates the storage configuration for a device

type DriverTask 

type DriverTask struct {
	Ctx       context.Context
	DeviceKey string
	Points    []model.Point
	ReadFunc  func(context.Context, []model.Point) (map[string]model.Value, error)
	Callback  func(map[string]model.Value, error)
}

type EdgeComputeManager 

type EdgeComputeManager struct {
	// contains filtered or unexported fields
}

func NewEdgeComputeManager 

func NewEdgeComputeManager(pipeline *DataPipeline, store *storage.Storage, saveFunc func([]model.EdgeRule) error) *EdgeComputeManager

func (*EdgeComputeManager) DeleteRule 

func (em *EdgeComputeManager) DeleteRule(id string) error

func (*EdgeComputeManager) GetFailedActions 

func (em *EdgeComputeManager) GetFailedActions() []model.FailedAction

func (*EdgeComputeManager) GetMetrics 

func (em *EdgeComputeManager) GetMetrics() EdgeComputeMetrics

func (*EdgeComputeManager) GetRuleStates 

func (em *EdgeComputeManager) GetRuleStates() map[string]*model.RuleRuntimeState

func (*EdgeComputeManager) GetRules 

func (em *EdgeComputeManager) GetRules() []model.EdgeRule

func (*EdgeComputeManager) GetSharedSources 

func (em *EdgeComputeManager) GetSharedSources() []SharedSourceInfo

func (*EdgeComputeManager) GetWindowData 

func (em *EdgeComputeManager) GetWindowData(ruleID string) []model.Value

func (*EdgeComputeManager) LoadRules 

func (em *EdgeComputeManager) LoadRules(rules []model.EdgeRule)

func (*EdgeComputeManager) QueryLogs 

func (em *EdgeComputeManager) QueryLogs(start, end time.Time, ruleID string) ([]model.RuleMinuteSnapshot, error)

QueryLogs retrieves logs based on time range and optional rule ID

func (*EdgeComputeManager) SetActionHook 

func (em *EdgeComputeManager) SetActionHook(hook func(ruleID string, action model.RuleAction, val model.Value, env map[string]any, err error))

func (*EdgeComputeManager) SetChannelManager 

func (em *EdgeComputeManager) SetChannelManager(cm *ChannelManager)

func (*EdgeComputeManager) SetDeviceWriter 

func (em *EdgeComputeManager) SetDeviceWriter(w DeviceIO)

func (*EdgeComputeManager) SetNorthboundManager 

func (em *EdgeComputeManager) SetNorthboundManager(nbm *NorthboundManager)

func (*EdgeComputeManager) SetStorage 

func (em *EdgeComputeManager) SetStorage(s *storage.Storage)

func (*EdgeComputeManager) Start 

func (em *EdgeComputeManager) Start()

func (*EdgeComputeManager) Stop 

func (em *EdgeComputeManager) Stop()

func (*EdgeComputeManager) UpsertRule 

func (em *EdgeComputeManager) UpsertRule(rule model.EdgeRule) error

type EdgeComputeMetrics 

type EdgeComputeMetrics struct {
	WorkerPoolSize    int   `json:"worker_pool_size"`
	WorkerPoolUsage   int   `json:"worker_pool_usage"`
	RuleCount         int   `json:"rule_count"`
	SharedSourceCount int   `json:"shared_source_count"`
	CacheSize         int   `json:"cache_size"`
	RulesTriggered    int64 `json:"rules_triggered"`
	RulesExecuted     int64 `json:"rules_executed"`
	RulesDropped      int64 `json:"rules_dropped"`
}

type ExecuteResult 

type ExecuteResult struct {
	Success bool
	Values  map[string]model.Value
	Error   error
}

type ExecutionContext 

type ExecutionContext struct {
	DeviceKey string
	Queue     chan *DriverTask
	Worker    *SerialWorker
	Driver    driver.Driver
	Running   bool
	// contains filtered or unexported fields
}

type ExecutionLayer 

type ExecutionLayer struct {
	// contains filtered or unexported fields
}

func NewExecutionLayer 

func NewExecutionLayer() *ExecutionLayer

func (*ExecutionLayer) Execute 

func (el *ExecutionLayer) Execute(task *ScanTask) *ExecuteResult

func (*ExecutionLayer) GetDriver 

func (el *ExecutionLayer) GetDriver(deviceKey string) driver.Driver

func (*ExecutionLayer) RegisterDriver 

func (el *ExecutionLayer) RegisterDriver(deviceKey string, d driver.Driver)

func (*ExecutionLayer) RegisterProtocol 

func (el *ExecutionLayer) RegisterProtocol(protocol string, pType ProtocolType)

func (*ExecutionLayer) SetIOProfileProvider 

func (el *ExecutionLayer) SetIOProfileProvider(fn IOProfileProvider)

func (*ExecutionLayer) SetPointDegradation 

func (el *ExecutionLayer) SetPointDegradation(m *PointDegradationManager)

func (*ExecutionLayer) Start 

func (el *ExecutionLayer) Start()

func (*ExecutionLayer) Stop 

func (el *ExecutionLayer) Stop()

func (*ExecutionLayer) UnregisterDriver 

func (el *ExecutionLayer) UnregisterDriver(deviceKey string)

type GapOptimizer 

type GapOptimizer struct {
	// contains filtered or unexported fields
}

GapOptimizer Gap优化器

func NewGapOptimizer 

func NewGapOptimizer() *GapOptimizer

NewGapOptimizer 创建Gap优化器

func (*GapOptimizer) ClearGapData 

func (g *GapOptimizer) ClearGapData(deviceID string)

ClearGapData 清除设备的Gap数据

func (*GapOptimizer) GetCurrentGap 

func (g *GapOptimizer) GetCurrentGap(deviceID string) int

GetCurrentGap 获取当前Gap

func (*GapOptimizer) OptimizeGap 

func (g *GapOptimizer) OptimizeGap(deviceID string, mtu int, rtt int64) int

OptimizeGap 优化Gap

func (*GapOptimizer) SetGap 

func (g *GapOptimizer) SetGap(deviceID string, gap int)

SetGap 设置Gap

type IOProfileProvider 

type IOProfileProvider func(deviceID string) DeviceIOProfile

type MTUManager 

type MTUManager struct {
	// contains filtered or unexported fields
}

MTUManager MTU管理器

func NewMTUManager 

func NewMTUManager() *MTUManager

NewMTUManager 创建MTU管理器

func (*MTUManager) ClearMTUData 

func (m *MTUManager) ClearMTUData(deviceID string)

ClearMTUData 清除设备的MTU数据

func (*MTUManager) GetCurrentMTU 

func (m *MTUManager) GetCurrentMTU(deviceID string) int

GetCurrentMTU 获取当前MTU

func (*MTUManager) GetMTUHistory 

func (m *MTUManager) GetMTUHistory(deviceID string) []model.MTUNegotiationRecord

GetMTUHistory 获取MTU历史记录

func (*MTUManager) NegotiateMTU 

func (m *MTUManager) NegotiateMTU(deviceID string, rtt int64) int

NegotiateMTU 协商MTU

func (*MTUManager) SetMTU 

func (m *MTUManager) SetMTU(deviceID string, mtu int)

SetMTU 设置MTU

type ModbusProtocolAdapter 

type ModbusProtocolAdapter struct{}

ModbusProtocolAdapter Modbus协议适配器

func NewModbusProtocolAdapter 

func NewModbusProtocolAdapter() *ModbusProtocolAdapter

NewModbusProtocolAdapter 创建Modbus协议适配器

func (*ModbusProtocolAdapter) AdjustParameters 

func (a *ModbusProtocolAdapter) AdjustParameters(deviceID string, params map[string]interface{}) map[string]interface{}

AdjustParameters 调整参数

func (*ModbusProtocolAdapter) GetDefaultParameters 

func (a *ModbusProtocolAdapter) GetDefaultParameters() map[string]interface{}

GetDefaultParameters 获取默认参数

func (*ModbusProtocolAdapter) GetProtocolType 

func (a *ModbusProtocolAdapter) GetProtocolType() string

GetProtocolType 获取协议类型

func (*ModbusProtocolAdapter) ValidateParameters 

func (a *ModbusProtocolAdapter) ValidateParameters(params map[string]interface{}) error

ValidateParameters 验证参数

type ModbusRegisterGenOptions 

type ModbusRegisterGenOptions struct {
	Start        int
	End          int
	DataType     string
	ReadWrite    string
	RegisterType model.RegisterType
	FunctionCode byte
	DeviceID     string
}

ModbusRegisterGenOptions 批量生成 Modbus 寄存器点位参数。

type NodeRuntimeState 

type NodeRuntimeState struct {
	FailCount     int       // 连续失败次数
	SuccessCount  int       // 连续成功次数
	LastSuccess   time.Time // 最后一次成功时间
	LastFailTime  time.Time // 最后一次失败时间
	NextRetryTime time.Time // 下一次重试时间（用于退避机制）
	State         NodeState // 当前节点状态
}

==================== 节点运行时状态结构 ==================== NodeRuntimeState 存储设备节点的运行时状态信息

type NodeState 

type NodeState int

==================== 节点运行状态定义 ==================== NodeState 定义设备节点的运行状态 

const (
	NodeStateOnline     NodeState = iota // 0 在线状态：设备正常通信
	NodeStateUnstable                    // 1 不稳定状态：设备通信时好时坏
	NodeStateOffline                     // 2 离线状态：设备暂时无法连接
	NodeStateQuarantine                  // 3 隔离状态：设备持续故障，需要延长退避时间
)

设备节点状态枚举

type NorthboundManager 

type NorthboundManager struct {
	// contains filtered or unexported fields
}

func NewNorthboundManager 

func NewNorthboundManager(cfg model.NorthboundConfig, pipeline *DataPipeline, sb model.SouthboundManager, s *storage.Storage, saveFunc func(model.NorthboundConfig) error) *NorthboundManager

func (*NorthboundManager) BatchWriteOPCUA 

func (nm *NorthboundManager) BatchWriteOPCUA(serverID string, requests []opcua.WriteRequest) []opcua.BatchWriteResult

BatchWriteOPCUA 批量写入多个点位

func (*NorthboundManager) DeleteEdgeOSMQTTConfig 

func (nm *NorthboundManager) DeleteEdgeOSMQTTConfig(id string) error

DeleteEdgeOSMQTTConfig 删除 edgeOS(MQTT) 配置

func (*NorthboundManager) DeleteEdgeOSNATSConfig 

func (nm *NorthboundManager) DeleteEdgeOSNATSConfig(id string) error

DeleteEdgeOSNATSConfig 删除 edgeOS(NATS) 配置

func (*NorthboundManager) DeleteHTTPConfig 

func (nm *NorthboundManager) DeleteHTTPConfig(id string) error

func (*NorthboundManager) DeleteMQTTConfig 

func (nm *NorthboundManager) DeleteMQTTConfig(id string) error

func (*NorthboundManager) DeleteOPCUAConfig 

func (nm *NorthboundManager) DeleteOPCUAConfig(id string) error

func (*NorthboundManager) DeleteSparkplugBConfig 

func (nm *NorthboundManager) DeleteSparkplugBConfig(id string) error

func (*NorthboundManager) GetConfig 

func (nm *NorthboundManager) GetConfig() model.NorthboundConfig

func (*NorthboundManager) GetEdgeOSMQTTStats 

func (nm *NorthboundManager) GetEdgeOSMQTTStats(id string) (edgos_mqtt.EdgeOSMQTTStats, error)

GetEdgeOSMQTTStats 获取 edgeOS(MQTT) 统计信息

func (*NorthboundManager) GetEdgeOSNATSStats 

func (nm *NorthboundManager) GetEdgeOSNATSStats(id string) (edgos_nats.EdgeOSNATSStats, error)

GetEdgeOSNATSStats 获取 edgeOS(NATS) 统计信息

func (*NorthboundManager) GetMQTTStats 

func (nm *NorthboundManager) GetMQTTStats(id string) (mqtt.MQTTStats, error)

func (*NorthboundManager) GetNorthboundStats 

func (nm *NorthboundManager) GetNorthboundStats() []NorthboundStatus

func (*NorthboundManager) GetOPCUAStats 

func (nm *NorthboundManager) GetOPCUAStats(id string) (opcua.Stats, error)

func (*NorthboundManager) GetOPCUAWriteHistory 

func (nm *NorthboundManager) GetOPCUAWriteHistory(serverID string, limit int) ([]opcua.WriteHistoryItem, error)

GetOPCUAWriteHistory 获取 OPC-UA 写入历史

func (*NorthboundManager) OnDeviceStatusChange 

func (nm *NorthboundManager) OnDeviceStatusChange(deviceID string, status int)

OnDeviceStatusChange handles device status changes and notifies northbound clients It publishes status to all configured endpoints that have this device mapped

func (*NorthboundManager) PublishEdgeOSMQTT 

func (nm *NorthboundManager) PublishEdgeOSMQTT(clientID, topic string, payload []byte) error

PublishEdgeOSMQTT 发布消息到指定的 edgeOS(MQTT) 客户端

func (*NorthboundManager) PublishEdgeOSNATS 

func (nm *NorthboundManager) PublishEdgeOSNATS(clientID, subject string, payload []byte) error

PublishEdgeOSNATS 发布消息到指定的 edgeOS(NATS) 客户端

func (*NorthboundManager) PublishHTTP 

func (nm *NorthboundManager) PublishHTTP(configID string, payload []byte) error

PublishHTTP sends a raw payload via a specific HTTP config

func (*NorthboundManager) PublishMQTT 

func (nm *NorthboundManager) PublishMQTT(clientID string, topic string, payload []byte) error

PublishMQTT publishes a message to a specific MQTT client or all if clientID is empty

func (*NorthboundManager) PublishMQTTClient 

func (nm *NorthboundManager) PublishMQTTClient(clientID string, topic string, payload []byte) error

PublishMQTTClient publishes to a specific client

func (*NorthboundManager) PublishPointsMetadata 

func (nm *NorthboundManager) PublishPointsMetadata()

PublishPointsMetadata publishes all point definitions to all edgeOS clients

func (*NorthboundManager) PublishPointsSync 

func (nm *NorthboundManager) PublishPointsSync(channelID, deviceID string)

PublishPointsSync publishes all point current values for a device to all edgeOS clients

func (*NorthboundManager) RebuildOPCUAServers 

func (nm *NorthboundManager) RebuildOPCUAServers()

RebuildOPCUAServers 重建所有已启用的 OPC UA Server 地址空间（仅重启北向 OPC UA 服务，不影响网关主进程与南向采集）。

func (*NorthboundManager) SetChannelManager 

func (nm *NorthboundManager) SetChannelManager(cm *ChannelManager)

SetChannelManager sets the ChannelManager reference for device lookups

func (*NorthboundManager) Start 

func (nm *NorthboundManager) Start()

func (*NorthboundManager) Stop 

func (nm *NorthboundManager) Stop()

func (*NorthboundManager) SyncOPCUAServer 

func (nm *NorthboundManager) SyncOPCUAServer(id string) error

SyncOPCUAServer 重建指定 OPC UA Server 的地址空间（同步南向点位读写权限等变更）。

func (*NorthboundManager) UpdateConfig 

func (nm *NorthboundManager) UpdateConfig(newConfig model.NorthboundConfig)

UpdateConfig 更新北向配置并重启相关客户端/服务器

func (*NorthboundManager) UpdateOPCUACertificates 

func (nm *NorthboundManager) UpdateOPCUACertificates(id string, certPEM, keyPEM string, trusted []string) (model.OPCUAConfig, error)

UpdateOPCUACertificates updates server/trusted PEM blobs and restarts the OPC UA server.

func (*NorthboundManager) UpsertEdgeOSMQTTConfig 

func (nm *NorthboundManager) UpsertEdgeOSMQTTConfig(cfg model.EdgeOSMQTTConfig) error

UpsertEdgeOSMQTTConfig 更新或插入 edgeOS(MQTT) 配置

func (*NorthboundManager) UpsertEdgeOSNATSConfig 

func (nm *NorthboundManager) UpsertEdgeOSNATSConfig(cfg model.EdgeOSNATSConfig) error

UpsertEdgeOSNATSConfig 更新或插入 edgeOS(NATS) 配置

func (*NorthboundManager) UpsertHTTPConfig 

func (nm *NorthboundManager) UpsertHTTPConfig(cfg model.HTTPConfig) error

func (*NorthboundManager) UpsertMQTTConfig 

func (nm *NorthboundManager) UpsertMQTTConfig(cfg model.MQTTConfig) error

func (*NorthboundManager) UpsertOPCUAConfig 

func (nm *NorthboundManager) UpsertOPCUAConfig(cfg model.OPCUAConfig) (model.OPCUAConfig, error)

func (*NorthboundManager) UpsertSparkplugBConfig 

func (nm *NorthboundManager) UpsertSparkplugBConfig(cfg model.SparkplugBConfig) error

func (*NorthboundManager) WriteOPCUA 

func (nm *NorthboundManager) WriteOPCUA(serverID, channelID, deviceID, pointID string, value any) error

WriteOPCUA 通过 OPC-UA 服务端写入单个点位值

type NorthboundStatus 

type NorthboundStatus struct {
	ID     string `json:"id"`
	Name   string `json:"name"`
	Type   string `json:"type"`
	Status string `json:"status"`
}

type PointDegradationManager 

type PointDegradationManager struct {
	// contains filtered or unexported fields
}

PointDegradationManager 点位级降级：连续失败跳过常规采集，指数探测恢复。

func NewPointDegradationManager 

func NewPointDegradationManager() *PointDegradationManager

func (*PointDegradationManager) FilterForRead 

func (m *PointDegradationManager) FilterForRead(deviceID string, pointIDs []string) (active []string, skipped []string)

FilterForRead 返回应参与本次读取的点位 ID；降级且未到探测时间的点位被跳过。

func (*PointDegradationManager) IsDegraded 

func (m *PointDegradationManager) IsDegraded(deviceID, pointID string) bool

IsDegraded 查询点位是否处于降级状态。

func (*PointDegradationManager) RecordResults 

func (m *PointDegradationManager) RecordResults(deviceID string, results map[string]string)

RecordResults 根据读结果更新点位状态。

func (*PointDegradationManager) SnapshotDevice 

func (m *PointDegradationManager) SnapshotDevice(deviceID string, pointIDs []string) map[string]any

SnapshotDevice 返回设备下各点位的降级快照。

type PriorityQueue 

type PriorityQueue []*ScanTask

func (PriorityQueue) Len 

func (pq PriorityQueue) Len() int

func (PriorityQueue) Less 

func (pq PriorityQueue) Less(i, j int) bool

func (*PriorityQueue) Peek 

func (pq *PriorityQueue) Peek() *ScanTask

func (*PriorityQueue) Pop 

func (pq *PriorityQueue) Pop() interface{}

func (*PriorityQueue) Push 

func (pq *PriorityQueue) Push(x interface{})

func (PriorityQueue) Swap 

func (pq PriorityQueue) Swap(i, j int)

type ProtocolAdapter 

type ProtocolAdapter interface {
	GetProtocolType() string
	AdjustParameters(deviceID string, params map[string]interface{}) map[string]interface{}
	ValidateParameters(params map[string]interface{}) error
	GetDefaultParameters() map[string]interface{}
}

ProtocolAdapter 协议适配器接口

type ProtocolAdapterRegistry 

type ProtocolAdapterRegistry struct {
	// contains filtered or unexported fields
}

ProtocolAdapterRegistry 协议适配器注册表

func NewProtocolAdapterRegistry 

func NewProtocolAdapterRegistry() *ProtocolAdapterRegistry

NewProtocolAdapterRegistry 创建协议适配器注册表

func (*ProtocolAdapterRegistry) GetAdapter 

func (r *ProtocolAdapterRegistry) GetAdapter(protocolType string) ProtocolAdapter

GetAdapter 获取协议适配器

func (*ProtocolAdapterRegistry) RegisterAdapter 

func (r *ProtocolAdapterRegistry) RegisterAdapter(adapter ProtocolAdapter)

RegisterAdapter 注册协议适配器

type ProtocolType 

type ProtocolType int
const (
	ProtocolTypeSerial ProtocolType = iota
	ProtocolTypeParallel
	ProtocolTypeLimited
)

type RTTManager 

type RTTManager struct {
	// contains filtered or unexported fields
}

RTTManager RTT管理器

func NewRTTManager 

func NewRTTManager() *RTTManager

NewRTTManager 创建RTT管理器

func (*RTTManager) ClearRTTData 

func (r *RTTManager) ClearRTTData(deviceID string)

ClearRTTData 清除设备的RTT数据

func (*RTTManager) GetAverageRTT 

func (r *RTTManager) GetAverageRTT(deviceID string) int64

GetAverageRTT 获取平均RTT

func (*RTTManager) GetEWMARTT 

func (r *RTTManager) GetEWMARTT(deviceID string) int64

GetEWMARTT 获取EWMA RTT

func (*RTTManager) GetMaxRTT 

func (r *RTTManager) GetMaxRTT(deviceID string) int64

GetMaxRTT 获取最大RTT

func (*RTTManager) GetRTTSamples 

func (r *RTTManager) GetRTTSamples(deviceID string) []int64

GetRTTSamples 获取RTT采样数据

func (*RTTManager) UpdateRTT 

func (r *RTTManager) UpdateRTT(deviceID string, rtt int64)

UpdateRTT 更新RTT数据

type ResourceController 

type ResourceController struct {
	// contains filtered or unexported fields
}

func NewResourceController 

func NewResourceController(limits ResourceLimits) *ResourceController

func (*ResourceController) Acquire 

func (rc *ResourceController) Acquire()

func (*ResourceController) AcquireConnection 

func (rc *ResourceController) AcquireConnection() bool

func (*ResourceController) CanExecute 

func (rc *ResourceController) CanExecute() bool

func (*ResourceController) GetConnectionCount 

func (rc *ResourceController) GetConnectionCount() int32

func (*ResourceController) GetGoroutineCount 

func (rc *ResourceController) GetGoroutineCount() int32

func (*ResourceController) Monitor 

func (rc *ResourceController) Monitor(wg *sync.WaitGroup)

func (*ResourceController) Release 

func (rc *ResourceController) Release()

func (*ResourceController) ReleaseConnection 

func (rc *ResourceController) ReleaseConnection()

func (*ResourceController) Stop 

func (rc *ResourceController) Stop()

type ResourceLimits 

type ResourceLimits struct {
	GoroutineLimit  int
	FDLimit         int
	ConnectionLimit int
	QueueLimit      int
}

type ScanEngine 

type ScanEngine struct {
	// contains filtered or unexported fields
}

func NewScanEngine 

func NewScanEngine(config ScanEngineConfig) *ScanEngine

func (*ScanEngine) AddTask 

func (se *ScanEngine) AddTask(deviceKey, protocol string, interval time.Duration, priority int, pointIDs []string, params map[string]any) *ScanTask

func (*ScanEngine) AddTaskWithScanClass 

func (se *ScanEngine) AddTaskWithScanClass(deviceKey, protocol, scanClass string, interval time.Duration, priority int, pointIDs []string, params map[string]any) *ScanTask

func (*ScanEngine) GetActiveTaskCount 

func (se *ScanEngine) GetActiveTaskCount() int

func (*ScanEngine) GetMetrics 

func (se *ScanEngine) GetMetrics() *ScanEngineMetrics

func (*ScanEngine) GetPendingTaskCount 

func (se *ScanEngine) GetPendingTaskCount() int

func (*ScanEngine) GetShadowCore 

func (se *ScanEngine) GetShadowCore() *ShadowCore

func (*ScanEngine) GetTask 

func (se *ScanEngine) GetTask(taskID string) *ScanTask

func (*ScanEngine) GetTaskByDeviceKey 

func (se *ScanEngine) GetTaskByDeviceKey(deviceKey string) *ScanTask

func (*ScanEngine) GetTasks 

func (se *ScanEngine) GetTasks() []*ScanTask

func (*ScanEngine) GetTasksByDeviceKey 

func (se *ScanEngine) GetTasksByDeviceKey(deviceKey string) []*ScanTask

func (*ScanEngine) RegisterDriver 

func (se *ScanEngine) RegisterDriver(deviceKey string, d driver.Driver)

func (*ScanEngine) RegisterProtocol 

func (se *ScanEngine) RegisterProtocol(protocol string, pType ProtocolType)

func (*ScanEngine) RemoveTask 

func (se *ScanEngine) RemoveTask(taskID string)

func (*ScanEngine) RemoveTasksByDeviceKey 

func (se *ScanEngine) RemoveTasksByDeviceKey(deviceKey string)

func (*ScanEngine) Run 

func (se *ScanEngine) Run()

func (*ScanEngine) SetCollectFinalize 

func (se *ScanEngine) SetCollectFinalize(fn CollectFinalizeFunc)

func (*ScanEngine) SetIOProfileProvider 

func (se *ScanEngine) SetIOProfileProvider(fn IOProfileProvider)

func (*ScanEngine) SetPointDegradation 

func (se *ScanEngine) SetPointDegradation(m *PointDegradationManager)

func (*ScanEngine) SetShadowCore 

func (se *ScanEngine) SetShadowCore(sc *ShadowCore)

func (*ScanEngine) Stop 

func (se *ScanEngine) Stop()

func (*ScanEngine) UnregisterDriver 

func (se *ScanEngine) UnregisterDriver(deviceKey string)

func (*ScanEngine) UpdateTaskInterval 

func (se *ScanEngine) UpdateTaskInterval(deviceKey string, interval time.Duration)

func (*ScanEngine) UpdateTaskPriority 

func (se *ScanEngine) UpdateTaskPriority(deviceKey string, priority int)

type ScanEngineAdapter 

type ScanEngineAdapter struct {
	// contains filtered or unexported fields
}

func NewScanEngineAdapter 

func NewScanEngineAdapter(scanEngine *ScanEngine) *ScanEngineAdapter

func (*ScanEngineAdapter) GetActiveTaskCount 

func (a *ScanEngineAdapter) GetActiveTaskCount() int

func (*ScanEngineAdapter) GetDriver 

func (a *ScanEngineAdapter) GetDriver(deviceID string) driver.Driver

func (*ScanEngineAdapter) GetPendingTaskCount 

func (a *ScanEngineAdapter) GetPendingTaskCount() int

func (*ScanEngineAdapter) GetTaskStatus 

func (a *ScanEngineAdapter) GetTaskStatus(deviceID string) ScanTaskStatus

func (*ScanEngineAdapter) GetTasks 

func (a *ScanEngineAdapter) GetTasks() []*ScanTask

func (*ScanEngineAdapter) IsStarted 

func (a *ScanEngineAdapter) IsStarted() bool

func (*ScanEngineAdapter) RegisterDevice 

func (a *ScanEngineAdapter) RegisterDevice(
	deviceID string,
	protocol string,
	channelDriver driver.Driver,
	channelMu *sync.Mutex,
	ch *model.Channel,
	dev *model.Device,
	interval time.Duration,
	priority int,
) error

func (*ScanEngineAdapter) Start 

func (a *ScanEngineAdapter) Start()

func (*ScanEngineAdapter) Stop 

func (a *ScanEngineAdapter) Stop()

func (*ScanEngineAdapter) UnregisterDevice 

func (a *ScanEngineAdapter) UnregisterDevice(deviceID string)

func (*ScanEngineAdapter) UpdateDeviceInterval 

func (a *ScanEngineAdapter) UpdateDeviceInterval(deviceID string, interval time.Duration)

func (*ScanEngineAdapter) UpdateDevicePriority 

func (a *ScanEngineAdapter) UpdateDevicePriority(deviceID string, priority int)

type ScanEngineConfig 

type ScanEngineConfig struct {
	TickInterval      time.Duration
	WorkerCount       int
	MaxQueueSize      int
	AntiStarvationSec int
	PriorityLevels    int
	GoroutineLimit    int
	ConnectionLimit   int
}

type ScanEngineMetrics 

type ScanEngineMetrics struct {
	TasksExecuted      atomic.Uint64
	TasksSucceeded     atomic.Uint64
	TasksFailed        atomic.Uint64
	StarvationRescues  atomic.Uint64
	TaskOverdueTotal   atomic.Uint64
	TotalScanLagMicros atomic.Uint64
	ScanLagSamples     atomic.Uint64
}

ScanEngineMetrics 调度引擎可观测指标（对标 Kepware Diagnostics）。

func (*ScanEngineMetrics) RecordExecute 

func (m *ScanEngineMetrics) RecordExecute(success bool, lagMicros int64)

func (*ScanEngineMetrics) RecordOverdue 

func (m *ScanEngineMetrics) RecordOverdue()

func (*ScanEngineMetrics) RecordStarvationRescue 

func (m *ScanEngineMetrics) RecordStarvationRescue()

func (*ScanEngineMetrics) Snapshot 

func (m *ScanEngineMetrics) Snapshot() map[string]any

type ScanTask 

type ScanTask struct {
	ID                  string
	DeviceKey           string
	ScanClass           string
	Protocol            string
	Interval            time.Duration
	BaseInterval        time.Duration
	NextRun             time.Time
	Priority            int
	FailRate            float64
	Status              ScanTaskStatus
	ConsecutiveFailures int
	ConsecutiveSuccess  int
	LastSuccess         time.Time
	LastFailure         time.Time
	PointIDs            []string
	Points              []model.Point
	Params              map[string]any
	// contains filtered or unexported fields
}

func (*ScanTask) GetStatus 

func (t *ScanTask) GetStatus() ScanTaskStatus

func (*ScanTask) SetStatus 

func (t *ScanTask) SetStatus(status ScanTaskStatus)

func (*ScanTask) UpdateNextRun 

func (t *ScanTask) UpdateNextRun(interval time.Duration)

type ScanTaskStatus 

type ScanTaskStatus int
const (
	ScanTaskStatusIdle ScanTaskStatus = iota
	ScanTaskStatusRunning
	ScanTaskStatusDegraded
	ScanTaskStatusStopped
)

func (ScanTaskStatus) String 

func (s ScanTaskStatus) String() string

type SerialQueueManager 

type SerialQueueManager struct {
	// contains filtered or unexported fields
}

func NewSerialQueueManager 

func NewSerialQueueManager() *SerialQueueManager

func (*SerialQueueManager) RegisterDriver 

func (sqm *SerialQueueManager) RegisterDriver(deviceKey string, d driver.Driver)

func (*SerialQueueManager) RemoveContext 

func (sqm *SerialQueueManager) RemoveContext(deviceKey string)

func (*SerialQueueManager) Stop 

func (sqm *SerialQueueManager) Stop()

func (*SerialQueueManager) Submit 

func (sqm *SerialQueueManager) Submit(task *DriverTask) bool

type SerialWorker 

type SerialWorker struct {
	// contains filtered or unexported fields
}

type ShadowBridge 

type ShadowBridge struct {
	// contains filtered or unexported fields
}

ShadowBridge 将 ShadowCore 快照变更扇出到 DataPipeline， 使边缘规则、北向推送与历史落库与 UI 影子数据源一致。

func NewShadowBridge 

func NewShadowBridge(pipeline *DataPipeline) *ShadowBridge

func (*ShadowBridge) Attach 

func (sb *ShadowBridge) Attach(sc *ShadowCore)

Attach 订阅 ShadowCore 变更并推送到 DataPipeline。

type ShadowCore 

type ShadowCore struct {
	// contains filtered or unexported fields
}

ShadowCore 维护每物理设备唯一的内存态影子设备（全量点位 + 通信画像），不落盘。

func NewShadowCore 

func NewShadowCore() *ShadowCore

func (*ShadowCore) CheckConsistency 

func (sc *ShadowCore) CheckConsistency(deviceID string, t time.Time) (*model.ConsistencyCheckResult, error)

func (*ShadowCore) CompareAndSwap 

func (sc *ShadowCore) CompareAndSwap(deviceID string, expectedVersion uint64, updates map[string]any) (*model.ShadowWriteResponse, error)

func (*ShadowCore) DeleteShadowDevice 

func (sc *ShadowCore) DeleteShadowDevice(deviceID string) error

func (*ShadowCore) GetAllShadowDevices 

func (sc *ShadowCore) GetAllShadowDevices() []*model.ShadowDevice

func (*ShadowCore) GetDeviceOptimization 

func (sc *ShadowCore) GetDeviceOptimization(deviceID string) map[string]interface{}

GetDeviceOptimization 返回设备 RTT/MTU/Gap 通信画像（微秒 RTT）。

func (*ShadowCore) GetMetrics 

func (sc *ShadowCore) GetMetrics() map[string]interface{}

func (*ShadowCore) GetShadowDevice 

func (sc *ShadowCore) GetShadowDevice(deviceID string) (*model.ShadowDevice, error)

func (*ShadowCore) GetShadowPoint 

func (sc *ShadowCore) GetShadowPoint(deviceID, pointID string) (*model.ShadowPoint, error)

func (*ShadowCore) GetVirtualShadowDevice 

func (sc *ShadowCore) GetVirtualShadowDevice(virtualDeviceID string) (*model.VirtualDevice, error)

GetVirtualShadowDevice 返回虚拟影子设备快照。

func (*ShadowCore) GetVirtualShadowPoint 

func (sc *ShadowCore) GetVirtualShadowPoint(virtualDeviceID, pointID string) (*model.ShadowPoint, error)

GetVirtualShadowPoint 读取虚拟影子设备单个点位。

func (*ShadowCore) ResolvePublishTarget 

func (sc *ShadowCore) ResolvePublishTarget(shadowDeviceID string) (channelID, deviceID string, err error)

ResolvePublishTarget 解析订阅通知中的 channel / device，供 ShadowBridge 与 WebSocket 使用。

func (*ShadowCore) Start 

func (sc *ShadowCore) Start()

func (*ShadowCore) Stop 

func (sc *ShadowCore) Stop()

func (*ShadowCore) Subscribe 

func (sc *ShadowCore) Subscribe(sub ShadowSubscriber)

func (*ShadowCore) UpdateDeviceRTT 

func (sc *ShadowCore) UpdateDeviceRTT(deviceID string, rtt int64)

UpdateDeviceRTT 更新设备的RTT数据

func (*ShadowCore) WriteShadowDevice 

func (sc *ShadowCore) WriteShadowDevice(msg model.ShadowIngressMessage) (*model.ShadowWriteResponse, error)

func (*ShadowCore) WriteShadowPoint 

func (sc *ShadowCore) WriteShadowPoint(req model.ShadowWriteRequest) (*model.ShadowWriteResponse, error)

func (*ShadowCore) WriteVirtualShadowDevice 

func (sc *ShadowCore) WriteVirtualShadowDevice(channelID, virtualDeviceID string, points map[string]model.ShadowPoint)

WriteVirtualShadowDevice 将虚拟影子计算结果写入 ShadowCore 并通知订阅者（Pipeline / UI）。

type ShadowDeviceOptimizer 

type ShadowDeviceOptimizer struct {
	// contains filtered or unexported fields
}

ShadowDeviceOptimizer 影子设备优化器，集成RTT、MTU、Gap优化

func NewShadowDeviceOptimizer 

func NewShadowDeviceOptimizer() *ShadowDeviceOptimizer

NewShadowDeviceOptimizer 创建影子设备优化器

func (*ShadowDeviceOptimizer) ClearDeviceData 

func (sdo *ShadowDeviceOptimizer) ClearDeviceData(deviceID string)

ClearDeviceData 清除设备数据

func (*ShadowDeviceOptimizer) GetAllDevices 

func (sdo *ShadowDeviceOptimizer) GetAllDevices() []string

GetAllDevices 获取所有设备的优化数据

func (*ShadowDeviceOptimizer) GetDeviceOptimization 

func (sdo *ShadowDeviceOptimizer) GetDeviceOptimization(deviceID string) map[string]interface{}

GetDeviceOptimization 获取设备优化参数

func (*ShadowDeviceOptimizer) GetGapOptimizer 

func (sdo *ShadowDeviceOptimizer) GetGapOptimizer() *GapOptimizer

GetGapOptimizer 获取Gap优化器

func (*ShadowDeviceOptimizer) GetMTUManager 

func (sdo *ShadowDeviceOptimizer) GetMTUManager() *MTUManager

GetMTUManager 获取MTU管理器

func (*ShadowDeviceOptimizer) GetRTTManager 

func (sdo *ShadowDeviceOptimizer) GetRTTManager() *RTTManager

GetRTTManager 获取RTT管理器

func (*ShadowDeviceOptimizer) LogDeviceOptimization 

func (sdo *ShadowDeviceOptimizer) LogDeviceOptimization(deviceID string)

LogDeviceOptimization 记录设备优化数据

func (*ShadowDeviceOptimizer) UpdateDeviceRTT 

func (sdo *ShadowDeviceOptimizer) UpdateDeviceRTT(deviceID string, rtt int64)

UpdateDeviceRTT 更新设备RTT并触发相关优化

func (*ShadowDeviceOptimizer) UpdateShadowDeviceProfile 

func (sdo *ShadowDeviceOptimizer) UpdateShadowDeviceProfile(shadowDevice *model.ShadowDevice)

UpdateShadowDeviceProfile 更新影子设备的通信画像

type ShadowIngress 

type ShadowIngress struct {
	// contains filtered or unexported fields
}

func NewShadowIngress 

func NewShadowIngress(sc *ShadowCore, bufferSize int, flushInterval time.Duration) *ShadowIngress

func (*ShadowIngress) GetBufferSize 

func (si *ShadowIngress) GetBufferSize() int

func (*ShadowIngress) GetMetrics 

func (si *ShadowIngress) GetMetrics() ShadowIngressMetrics

func (*ShadowIngress) Ingest 

func (si *ShadowIngress) Ingest(val model.Value) error

func (*ShadowIngress) IngestBatch 

func (si *ShadowIngress) IngestBatch(values []model.Value) error

func (*ShadowIngress) IngestDirect 

func (si *ShadowIngress) IngestDirect(msg model.ShadowIngressMessage) error

func (*ShadowIngress) Start 

func (si *ShadowIngress) Start()

func (*ShadowIngress) Stop 

func (si *ShadowIngress) Stop()

type ShadowIngressMetrics 

type ShadowIngressMetrics struct {
	TotalMessages   uint64
	TotalPoints     uint64
	FailedMessages  uint64
	LastProcessTime time.Time
}

type ShadowSubscriber 

type ShadowSubscriber func(deviceID string, points map[string]model.ShadowPoint)

type SharedSourceInfo 

type SharedSourceInfo struct {
	SourceID        string   `json:"source_id"`
	Subscribers     []string `json:"subscribers"`
	SubscriberCount int      `json:"subscriber_count"`
}

type StateChangeCallback 

type StateChangeCallback func(deviceID string, oldState, newState NodeState)

==================== 通信管理模板 ==================== CommunicationManageTemplate 管理设备采集的状态机和重试策略

type StoreForwardManager 

type StoreForwardManager struct {
	// contains filtered or unexported fields
}

StoreForwardManager 统一 Store & Forward：南向 values 与北向 NorthboundCache。

func NewStoreForwardManager 

func NewStoreForwardManager(store *storage.Storage, policy StoreForwardPolicy) *StoreForwardManager

func (*StoreForwardManager) CacheNorthbound 

func (m *StoreForwardManager) CacheNorthbound(configID string, data []byte) error

CacheNorthbound 缓存北向离线消息（复用 NorthboundCache bucket）。

func (*StoreForwardManager) HandleBatch 

func (m *StoreForwardManager) HandleBatch(batch []model.Value)

HandleBatch 作为 Pipeline 批量处理器缓存南向采集值。

func (*StoreForwardManager) ReplaySouthbound 

func (m *StoreForwardManager) ReplaySouthbound(limit int) []model.Value

ReplaySouthbound 回放最近缓存的南向值。

func (*StoreForwardManager) SetStorage 

func (m *StoreForwardManager) SetStorage(store *storage.Storage)

type StoreForwardPolicy 

type StoreForwardPolicy struct {
	MaxSouthboundRecords int
	MaxNorthboundPerID   int
}

StoreForwardPolicy 统一南向历史与北向离线缓存策略。

type SystemManager 

type SystemManager struct {
	// contains filtered or unexported fields
}

func NewSystemManager 

func NewSystemManager(cfg *config.Config) *SystemManager

func (*SystemManager) GetConfig 

func (sm *SystemManager) GetConfig() model.SystemConfig

func (*SystemManager) GetNetworkInterfaces 

func (sm *SystemManager) GetNetworkInterfaces() ([]model.NetworkInterface, error)

func (*SystemManager) GetRoutes 

func (sm *SystemManager) GetRoutes() ([]model.StaticRoute, error)

func (*SystemManager) GetUser 

func (sm *SystemManager) GetUser(username string) (*model.UserConfig, bool)

func (*SystemManager) SetConfigManager 

func (sm *SystemManager) SetConfigManager(cm *config.ConfigManager)

SetConfigManager 注入配置管理器，使系统/用户配置持久化到数据库。

func (*SystemManager) UpdateConfig 

func (sm *SystemManager) UpdateConfig(newConfig model.SystemConfig) error

func (*SystemManager) UpdateUserPassword 

func (sm *SystemManager) UpdateUserPassword(username, newPassword string) error

type TCPProtocolAdapter 

type TCPProtocolAdapter struct{}

TCPProtocolAdapter TCP协议适配器

func NewTCPProtocolAdapter 

func NewTCPProtocolAdapter() *TCPProtocolAdapter

NewTCPProtocolAdapter 创建TCP协议适配器

func (*TCPProtocolAdapter) AdjustParameters 

func (a *TCPProtocolAdapter) AdjustParameters(deviceID string, params map[string]interface{}) map[string]interface{}

AdjustParameters 调整参数

func (*TCPProtocolAdapter) GetDefaultParameters 

func (a *TCPProtocolAdapter) GetDefaultParameters() map[string]interface{}

GetDefaultParameters 获取默认参数

func (*TCPProtocolAdapter) GetProtocolType 

func (a *TCPProtocolAdapter) GetProtocolType() string

GetProtocolType 获取协议类型

func (*TCPProtocolAdapter) ValidateParameters 

func (a *TCPProtocolAdapter) ValidateParameters(params map[string]interface{}) error

ValidateParameters 验证参数

type TagEntry 

type TagEntry struct {
	ChannelID string
	DeviceID  string
	PointID   string
	Name      string
	Alias     string
	EU        string
	Scale     float64
	Offset    float64
	ScanClass string
}

TagEntry 表示 Tag 数据库中的标准化点位元数据。

type TagRegistry 

type TagRegistry struct {
	// contains filtered or unexported fields
}

TagRegistry 统一点位 ID、别名、工程单位与缩放因子。

func NewTagRegistry 

func NewTagRegistry() *TagRegistry

func (*TagRegistry) ApplyScaling 

func (tr *TagRegistry) ApplyScaling(v model.Value) model.Value

func (*TagRegistry) Count 

func (tr *TagRegistry) Count() int

func (*TagRegistry) Get 

func (tr *TagRegistry) Get(channelID, deviceID, pointID string) (TagEntry, bool)

func (*TagRegistry) RegisterFromDevice 

func (tr *TagRegistry) RegisterFromDevice(channelID string, dev *model.Device)

func (*TagRegistry) Resolve 

func (tr *TagRegistry) Resolve(alias string) (TagEntry, error)

func (*TagRegistry) UnregisterDevice 

func (tr *TagRegistry) UnregisterDevice(channelID, deviceID string)

type TokenBucket 

type TokenBucket struct {
	// contains filtered or unexported fields
}

func NewTokenBucket 

func NewTokenBucket(rate float64, capacity float64) *TokenBucket

func (*TokenBucket) Allow 

func (tb *TokenBucket) Allow() bool

type VirtualShadowEngine 

type VirtualShadowEngine struct {
	// contains filtered or unexported fields
}

func NewVirtualShadowEngine 

func NewVirtualShadowEngine(sc *ShadowCore) *VirtualShadowEngine

func (*VirtualShadowEngine) CreateVirtualDevice 

func (vse *VirtualShadowEngine) CreateVirtualDevice(deviceID, channelID string, formulaPoints map[string]string) error

func (*VirtualShadowEngine) DeleteVirtualDevice 

func (vse *VirtualShadowEngine) DeleteVirtualDevice(deviceID string) error

func (*VirtualShadowEngine) GetAllVirtualDevices 

func (vse *VirtualShadowEngine) GetAllVirtualDevices() []*model.VirtualDevice

func (*VirtualShadowEngine) GetDependencyGraph 

func (vse *VirtualShadowEngine) GetDependencyGraph() map[string][]string

func (*VirtualShadowEngine) GetMetrics 

func (vse *VirtualShadowEngine) GetMetrics() map[string]interface{}

func (*VirtualShadowEngine) GetVirtualDevice 

func (vse *VirtualShadowEngine) GetVirtualDevice(deviceID string) (*model.VirtualDevice, error)

func (*VirtualShadowEngine) RecomputeDevice 

func (vse *VirtualShadowEngine) RecomputeDevice(deviceID string)

RecomputeDevice 同步重算虚拟设备点位（供 API 刷新当前值）。

func (*VirtualShadowEngine) ReplaceVirtualDevice 

func (vse *VirtualShadowEngine) ReplaceVirtualDevice(deviceID, channelID string, formulaPoints map[string]string) error

ReplaceVirtualDevice 全量替换虚拟设备公式并触发重算。

func (*VirtualShadowEngine) UpdateFormula 

func (vse *VirtualShadowEngine) UpdateFormula(deviceID, pointID, newFormula string) error

type VirtualShadowManager 

type VirtualShadowManager struct {
	// contains filtered or unexported fields
}

VirtualShadowManager 管理虚拟影子设备配置与运行时引擎。

func NewVirtualShadowManager 

func NewVirtualShadowManager(vse *VirtualShadowEngine, cm *ChannelManager, shadow *ShadowCore, saveFn func([]model.VirtualShadowDeviceConfig) error) *VirtualShadowManager

func (*VirtualShadowManager) Create 

func (m *VirtualShadowManager) Create(cfg model.VirtualShadowDeviceConfig) error

func (*VirtualShadowManager) Delete 

func (m *VirtualShadowManager) Delete(id string) error

func (*VirtualShadowManager) Engine 

func (m *VirtualShadowManager) Engine() *VirtualShadowEngine

func (*VirtualShadowManager) Get 

func (m *VirtualShadowManager) Get(id string) (*model.VirtualShadowDeviceConfig, error)

func (*VirtualShadowManager) GetRuntime 

func (m *VirtualShadowManager) GetRuntime(id string) (*model.VirtualDevice, map[string]model.ShadowPoint, error)

func (*VirtualShadowManager) List 

func (m *VirtualShadowManager) List() []model.VirtualShadowDeviceConfig

func (*VirtualShadowManager) ListDevicePointSources 

func (m *VirtualShadowManager) ListDevicePointSources(channelID, deviceID, query string) ([]model.PointSourceRef, error)

ListDevicePointSources 返回指定设备的可选点位（支持点位级过滤）。

func (*VirtualShadowManager) ListPointSources 

func (m *VirtualShadowManager) ListPointSources() []model.PointSourceRef

func (*VirtualShadowManager) Load 

func (m *VirtualShadowManager) Load(configs []model.VirtualShadowDeviceConfig)

func (*VirtualShadowManager) RefreshRuntime 

func (m *VirtualShadowManager) RefreshRuntime(id string) (*model.VirtualDevice, map[string]model.ShadowPoint, error)

RefreshRuntime 触发重算后返回最新运行时快照。

func (*VirtualShadowManager) ReloadAll 

func (m *VirtualShadowManager) ReloadAll()

ReloadAll 在通道/设备/点位拓扑变更后，重新应用所有已启用的虚拟影子设备。

func (*VirtualShadowManager) SearchSourceDevices 

func (m *VirtualShadowManager) SearchSourceDevices(query, channelID string, limit int) []model.SourceDeviceSummary

SearchSourceDevices 按设备名称/ID/通道模糊检索源设备（不含点位明细）。

func (*VirtualShadowManager) Update 

func (m *VirtualShadowManager) Update(id string, cfg model.VirtualShadowDeviceConfig) error

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

func NewWorker 

func NewWorker(id int, taskQueue chan func(), stopCh chan struct{}, wg *sync.WaitGroup) *Worker

type WorkerPool 

type WorkerPool struct {
	// contains filtered or unexported fields
}

func NewWorkerPool 

func NewWorkerPool(workerCount int) *WorkerPool

func (*WorkerPool) ActiveCount 

func (wp *WorkerPool) ActiveCount() int

func (*WorkerPool) PendingCount 

func (wp *WorkerPool) PendingCount() int

func (*WorkerPool) SetWorkerCount 

func (wp *WorkerPool) SetWorkerCount(count int)

func (*WorkerPool) Start 

func (wp *WorkerPool) Start()

func (*WorkerPool) Stop 

func (wp *WorkerPool) Stop()

func (*WorkerPool) Submit 

func (wp *WorkerPool) Submit(task func()) bool

func (*WorkerPool) WaitForIdle 

func (wp *WorkerPool) WaitForIdle(timeout time.Duration) bool

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.