core

package
v0.1.2 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: GPL-2.0, GPL-3.0 Imports: 36 Imported by: 0

Documentation

Overview

ConnectionController is a read-only observability module for connection health. It classifies errors, records read/connection metrics, and exposes health signals.

Reconnect/dial MUST NOT be triggered through ConnectionController. All reconnect behavior goes exclusively through driver.ConnectionManager (EnsureConnected / ScheduleReconnect).

Index

Constants

View Source 
const (
	SLAScanLagP95MsThreshold   = 100.0
	SLAScanDriftAvgMsThreshold = 50.0
	SLAScanMissDeadlineMax     = 0
	SLACircuitBreakerRejectMax = 0
)

SLA 告警阈值（运维对齐用，可通过 diagnostics API 的 sla_warnings 查看）。

View Source 
const (
	SoakSampleInterval         = 15 * time.Second
	SoakMaxTrendSamples        = 480
	SoakBacklogExcessThreshold = 10 // max allowed backlog above registered task count
	SoakPointSuccessRateGate   = 0.99
)
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")
	ErrCircuitOpen           = errors.New("circuit breaker open")
)
View Source 
var (
	ErrDriverNotFound = errors.New("driver not found")
	ErrQueueFull      = errors.New("queue full")
	ErrRateLimited    = errors.New("rate limited")
)
View Source 
var DLT645StandardPointTemplates = []DLT645PointTemplate{

	{ID: "combined_active_energy", Name: "当前组合有功总电能", DataID: "00-00-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "last_settlement_combined_active_energy", Name: "上1结算日组合有功总电能", DataID: "00-00-00-01", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "associated_total_energy", Name: "当前关联总电能", DataID: "00-80-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "last_settlement_a_forward_active_energy", Name: "上1结算日A相正向有功电能", DataID: "00-15-00-01", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "forward_active_energy", Name: "当前正向有功总电能", DataID: "00-01-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "last_settlement_forward_active_energy", Name: "上1结算日正向有功总电能", DataID: "00-01-00-01", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "reverse_active_energy", Name: "当前反向有功总电能", DataID: "00-02-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "combined_reactive1_energy", Name: "当前组合无功1总电能", DataID: "00-03-00-00", DataType: "uint64", Scale: 0.01, Unit: "kvarh"},
	{ID: "combined_reactive2_energy", Name: "当前组合无功2总电能", DataID: "00-04-00-00", DataType: "uint64", Scale: 0.01, Unit: "kvarh"},
	{ID: "forward_reactive_energy", Name: "当前正向无功总电能", DataID: "00-05-00-00", DataType: "uint64", Scale: 0.01, Unit: "kvarh"},
	{ID: "reverse_reactive_energy", Name: "当前反向无功总电能", DataID: "00-06-00-00", DataType: "uint64", Scale: 0.01, Unit: "kvarh"},
	{ID: "a_phase_forward_active_energy", Name: "当前A相正向有功电能", DataID: "00-15-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "a_phase_reverse_active_energy", Name: "当前A相反向有功电能", DataID: "00-16-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "b_phase_forward_active_energy", Name: "当前B相正向有功电能", DataID: "00-29-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "b_phase_reverse_active_energy", Name: "当前B相反向有功电能", DataID: "00-2A-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "c_phase_forward_active_energy", Name: "当前C相正向有功电能", DataID: "00-3D-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},
	{ID: "c_phase_reverse_active_energy", Name: "当前C相反向有功电能", DataID: "00-3E-00-00", DataType: "uint64", Scale: 0.01, Unit: "kWh"},

	{ID: "forward_active_max_demand", Name: "当前正向有功最大需量", DataID: "01-01-00-00", DataType: "uint64", Scale: 0.0001, Unit: "kW"},
	{ID: "forward_active_max_demand_time", Name: "当前正向有功最大需量发生时间", DataID: "01-01-00-00#T", DataType: "string", Scale: 0, Unit: ""},
	{ID: "reverse_active_max_demand", Name: "当前反向有功最大需量", DataID: "01-02-00-00", DataType: "uint64", Scale: 0.0001, Unit: "kW"},
	{ID: "reverse_active_max_demand_time", Name: "当前反向有功最大需量发生时间", DataID: "01-02-00-00#T", DataType: "string", Scale: 0, Unit: ""},

	{ID: "a_phase_voltage", Name: "A相电压", DataID: "02-01-01-00", DataType: "uint16", Scale: 0.1, Unit: "V"},
	{ID: "b_phase_voltage", Name: "B相电压", DataID: "02-01-02-00", DataType: "uint16", Scale: 0.1, Unit: "V"},
	{ID: "c_phase_voltage", Name: "C相电压", DataID: "02-01-03-00", DataType: "uint16", Scale: 0.1, Unit: "V"},
	{ID: "a_phase_current", Name: "A相电流", DataID: "02-02-01-00", DataType: "uint32", Scale: 0.001, Unit: "A"},
	{ID: "b_phase_current", Name: "B相电流", DataID: "02-02-02-00", DataType: "uint32", Scale: 0.001, Unit: "A"},
	{ID: "c_phase_current", Name: "C相电流", DataID: "02-02-03-00", DataType: "uint32", Scale: 0.001, Unit: "A"},
	{ID: "total_active_power", Name: "瞬时总有功功率", DataID: "02-03-00-00", DataType: "int32", Scale: 0.0001, Unit: "kW"},
	{ID: "a_phase_active_power", Name: "A相有功功率", DataID: "02-03-01-00", DataType: "int32", Scale: 0.1, Unit: "kW"},
	{ID: "b_phase_active_power", Name: "B相有功功率", DataID: "02-03-02-00", DataType: "int32", Scale: 0.0001, Unit: "kW"},
	{ID: "c_phase_active_power", Name: "C相有功功率", DataID: "02-03-03-00", DataType: "int32", Scale: 0.0001, Unit: "kW"},
	{ID: "total_reactive_power", Name: "瞬时总无功功率", DataID: "02-04-00-00", DataType: "int32", Scale: 0.0001, Unit: "kvar"},
	{ID: "total_apparent_power", Name: "瞬时总视在功率", DataID: "02-05-00-00", DataType: "int32", Scale: 0.0001, Unit: "kVA"},
	{ID: "total_power_factor", Name: "总功率因数", DataID: "02-06-00-00", DataType: "uint16", Scale: 0.001, Unit: ""},
	{ID: "a_phase_power_factor", Name: "A相功率因数", DataID: "02-06-01-00", DataType: "uint16", Scale: 0.001, Unit: ""},
	{ID: "b_phase_power_factor", Name: "B相功率因数", DataID: "02-06-02-00", DataType: "uint16", Scale: 0.001, Unit: ""},
	{ID: "c_phase_power_factor", Name: "C相功率因数", DataID: "02-06-03-00", DataType: "uint16", Scale: 0.001, Unit: ""},
	{ID: "a_phase_voltage_harmonic_1", Name: "A相电压1次谐波含量", DataID: "02-0A-01-01", DataType: "uint16", Scale: 0.01, Unit: "%"},
	{ID: "b_phase_voltage_harmonic_1", Name: "B相电压1次谐波含量", DataID: "02-0A-02-01", DataType: "uint16", Scale: 0.01, Unit: "%"},
	{ID: "c_phase_voltage_harmonic_1", Name: "C相电压1次谐波含量", DataID: "02-0A-03-01", DataType: "uint16", Scale: 0.01, Unit: "%"},
	{ID: "a_phase_current_harmonic_1", Name: "A相电流1次谐波含量", DataID: "02-0B-01-01", DataType: "uint16", Scale: 0.01, Unit: "%"},
	{ID: "neutral_current", Name: "零线电流", DataID: "02-80-00-01", DataType: "uint16", Scale: 0.01, Unit: "A"},
	{ID: "grid_frequency", Name: "电网频率", DataID: "02-80-00-02", DataType: "uint16", Scale: 0.01, Unit: "Hz"},
	{ID: "current_active_demand", Name: "当前有功需量", DataID: "02-80-00-04", DataType: "int32", Scale: 0.0001, Unit: "kW"},

	{ID: "date_time", Name: "日期及时间", DataID: "04-00-01-01", DataType: "string", Scale: 0, Unit: ""},
	{ID: "max_demand_period", Name: "最大需量周期", DataID: "04-00-01-03", DataType: "uint16", Scale: 0, Unit: "min"},
	{ID: "slip_time", Name: "滑差时间", DataID: "04-00-01-04", DataType: "uint16", Scale: 0, Unit: "min"},
	{ID: "tariff_count", Name: "费率数", DataID: "04-00-02-04", DataType: "uint8", Scale: 0, Unit: ""},
	{ID: "communication_address", Name: "通信地址", DataID: "04-00-04-01", DataType: "string", Scale: 0, Unit: ""},
	{ID: "meter_number", Name: "表号", DataID: "04-00-04-02", DataType: "string", Scale: 0, Unit: ""},
	{ID: "meter_status_word_1", Name: "电表运行状态字1", DataID: "04-00-05-01", DataType: "uint32", Scale: 0, Unit: ""},
	{ID: "meter_status_word_2", Name: "电表运行状态字2", DataID: "04-00-05-02", DataType: "uint32", Scale: 0, Unit: ""},

	{ID: "last_timed_freeze_time", Name: "上1次定时冻结时间", DataID: "05-00-00-01", DataType: "string", Scale: 0, Unit: ""},
	{ID: "last_timed_freeze_forward_active_energy", Name: "上1次定时冻结正向有功总电能", DataID: "05-00-01-01", DataType: "uint32", Scale: 0.01, Unit: "kWh"},
	{ID: "last_timed_freeze_reverse_active_energy", Name: "上1次定时冻结反向有功总电能", DataID: "05-00-02-01", DataType: "uint32", Scale: 0.01, Unit: "kWh"},
	{ID: "last_timed_freeze_forward_active_max_demand", Name: "上1次定时冻结正向有功最大需量", DataID: "05-00-09-01", DataType: "uint32", Scale: 0.0001, Unit: "kW"},
	{ID: "last_timed_freeze_forward_active_max_demand_time", Name: "上1次定时冻结正向有功最大需量发生时间", DataID: "05-00-09-01#T", DataType: "string", Scale: 0, Unit: ""},
	{ID: "last_instant_freeze_time", Name: "上1次瞬时冻结时间", DataID: "05-01-00-01", DataType: "string", Scale: 0, Unit: ""},
	{ID: "last_instant_freeze_forward_active_energy", Name: "上1次瞬时冻结正向有功总电能", DataID: "05-01-01-01", DataType: "uint32", Scale: 0.01, Unit: "kWh"},
	{ID: "last_daily_freeze_time", Name: "上1次日冻结时间", DataID: "05-06-00-01", DataType: "string", Scale: 0, Unit: ""},
	{ID: "last_daily_freeze_forward_active_energy", Name: "上1次日冻结正向有功总电能", DataID: "05-06-01-01", DataType: "uint32", Scale: 0.01, Unit: "kWh"},
}

DLT645StandardPointTemplates lists common standard DIs from DL/T 645-2007 section 4.2.3 and appendix A.2–A.6 (commonly deployed subset).

Functions

func ClassifyEdgeErrorType 

func ClassifyEdgeErrorType(phase, errMsg string) string

ClassifyEdgeErrorType maps phase and message to a stable error category.

func DLT645StationAddress 

func DLT645StationAddress(config map[string]any) string

DLT645StationAddress returns the 12-digit meter address from device config.

func GenerateDLT645StandardPoints 

func GenerateDLT645StandardPoints(stationAddress, deviceID string) []model.Point

GenerateDLT645StandardPoints builds standard DI points for a meter address.

func GenerateModbusRegisterPoints 

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

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

func IsDLT645AutoPointsEnabled 

func IsDLT645AutoPointsEnabled(config map[string]any) bool

IsDLT645AutoPointsEnabled reports whether standard DI points should be auto-created. Defaults to true when the key is absent (new devices and legacy configs).

func IsEdgeErrorMinuteSnapshot 

func IsEdgeErrorMinuteSnapshot(snap model.RuleMinuteSnapshot) bool

IsEdgeErrorMinuteSnapshot reports whether a persisted minute snapshot is an error log row.

func IsVirtualShadowID 

func IsVirtualShadowID(shadowDeviceID string) bool

func MatchesEdgeLogFilter 

func MatchesEdgeLogFilter(snap model.RuleMinuteSnapshot, filter EdgeLogFilter) bool

MatchesEdgeLogFilter reports whether a snapshot satisfies optional edge log filters.

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 AsyncJob 

type AsyncJob struct {
	ID        string         `json:"job_id"`
	Type      AsyncJobType   `json:"type"`
	Status    AsyncJobStatus `json:"status"`
	ChannelID string         `json:"channel_id,omitempty"`
	DeviceID  string         `json:"device_id,omitempty"`
	CreatedAt time.Time      `json:"created_at"`
	UpdatedAt time.Time      `json:"updated_at"`
	Result    any            `json:"result,omitempty"`
	Error     string         `json:"error,omitempty"`
	// contains filtered or unexported fields
}

AsyncJob is a pollable record for long scan/browse work.

type AsyncJobManager 

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

AsyncJobManager stores in-memory scan/browse jobs (same shape as AI task poll).

func NewAsyncJobManager 

func NewAsyncJobManager() *AsyncJobManager

NewAsyncJobManager creates a job store with background TTL eviction.

func (*AsyncJobManager) Cancel 

func (m *AsyncJobManager) Cancel(id string) error

Cancel requests cancellation of a running job.

func (*AsyncJobManager) Get 

func (m *AsyncJobManager) Get(id string) (*AsyncJob, bool)

Get returns a detached snapshot of a job.

func (*AsyncJobManager) Snapshot 

func (m *AsyncJobManager) Snapshot(id string) *AsyncJob

Snapshot copies job fields for JSON responses (excludes cancel).

func (*AsyncJobManager) Stop 

func (m *AsyncJobManager) Stop()

Stop halts the eviction goroutine.

func (*AsyncJobManager) Submit 

func (m *AsyncJobManager) Submit(jobType AsyncJobType, channelID, deviceID string, work func(ctx context.Context) (any, error)) *AsyncJob

Submit creates a job and runs work in the background. Returns a snapshot immediately.

type AsyncJobStatus 

type AsyncJobStatus string

AsyncJobStatus is the lifecycle state of a long-running API job (scan/browse). 

const (
	AsyncJobQueued    AsyncJobStatus = "queued"
	AsyncJobRunning   AsyncJobStatus = "running"
	AsyncJobSucceeded AsyncJobStatus = "succeeded"
	AsyncJobFailed    AsyncJobStatus = "failed"
	AsyncJobCancelled AsyncJobStatus = "cancelled"
)

type AsyncJobType 

type AsyncJobType string

AsyncJobType classifies work submitted via the async job API. 

const (
	AsyncJobScanChannel AsyncJobType = "scan_channel"
	AsyncJobScanDevice  AsyncJobType = "scan_device"
)

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, _ float64) *BackpressureController

func (*BackpressureController) Allow 

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

func (*BackpressureController) AllowWithReason 

func (bc *BackpressureController) AllowWithReason(ctx ThrottleContext) (bool, RejectReason)

func (*BackpressureController) LogReject 

func (bc *BackpressureController) LogReject(deviceKey, protocol string, reason RejectReason)

func (*BackpressureController) ReduceTokenRate 

func (bc *BackpressureController) ReduceTokenRate(factor float64)

func (*BackpressureController) RejectByReason 

func (bc *BackpressureController) RejectByReason() map[string]uint64

func (*BackpressureController) RejectTotal 

func (bc *BackpressureController) RejectTotal() uint64

func (*BackpressureController) Release 

func (bc *BackpressureController) Release(deviceKey string)

func (*BackpressureController) TokenRate 

func (bc *BackpressureController) TokenRate() float64

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) AddDevices 

func (cm *ChannelManager) AddDevices(channelID string, devices []model.Device) ([]model.Device, error)

AddDevices 批量添加设备（单次持久化与拓扑通知）。

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) GetChannelScanEngineMetricsSnapshot 

func (cm *ChannelManager) GetChannelScanEngineMetricsSnapshot(channelID string) map[string]any

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 获取指定设备的所有点位数据（优先 Shadow；缺失时返回 Uncertain 元数据，绝不在 API 路径 live 读）。

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) GetSoakMonitorSnapshot 

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

func (*ChannelManager) GetStateManager 

func (cm *ChannelManager) GetStateManager() *CommunicationManageTemplate

GetStateManager 获取状态管理器

func (*ChannelManager) GetTagRegistry 

func (cm *ChannelManager) GetTagRegistry() *TagRegistry

func (*ChannelManager) Jobs 

func (cm *ChannelManager) Jobs() *AsyncJobManager

Jobs returns the async job manager used for scan/browse APIs.

func (*ChannelManager) OnBACnetAddressDiscovered 

func (cm *ChannelManager) OnBACnetAddressDiscovered(deviceKey, ip string, port int)

OnBACnetAddressDiscovered persists a runtime address change (e.g. UDP port after device reboot).

func (*ChannelManager) ReadAllPoints added in v0.1.2 

func (cm *ChannelManager) ReadAllPoints(channelID, deviceID string) ([]model.Value, error)

ReadAllPoints 立即读取指定设备全部点位一次，并把结果写入 shadow 供前端刷新展示。 与周期采集循环共用 withDriverIO 的通道串行锁，因此不会与该设备/共享链路并发 I/O。

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(ctx context.Context, channelID string, params map[string]any) (any, error)

ScanChannel 扫描通道下的设备。ctx 应由调用方设置超时（API job 或 sync 路径）。

func (*ChannelManager) ScanDevice 

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

ScanDevice 扫描设备下的对象（点位）。ctx 应由调用方设置超时（OPC UA Browse 可达 180s）。

func (*ChannelManager) SetShadowCore 

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

func (*ChannelManager) SetShadowIngress 

func (cm *ChannelManager) SetShadowIngress(si *ShadowIngress)

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) StartScanChannelJob 

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

StartScanChannelJob submits channel device discovery as an async job.

func (*ChannelManager) StartScanDeviceJob 

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

StartScanDeviceJob submits device object/point browse as an async job.

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 CircuitBreakerEventHandler 

type CircuitBreakerEventHandler func(deviceKey, eventType, message string)

CircuitBreakerEventHandler receives Open/Reject events for channel Event Log wiring.

type CircuitState 

type CircuitState int
const (
	CircuitClosed CircuitState = iota
	CircuitOpen
	CircuitHalfOpen
)

func (CircuitState) String 

func (s CircuitState) String() string

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) MarkOffline 

func (c *CommunicationManageTemplate) MarkOffline(deviceID string)

MarkOffline 将设备标记为离线（例如通道链路断开时批量更新）。

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状态: 只有在退避时间过后才允许采集

func (*CommunicationManageTemplate) UnregisterNode 

func (c *CommunicationManageTemplate) UnregisterNode(nodeID string)

UnregisterNode 注销指定设备/通道节点，释放内存中的运行时状态。 在删除设备或通道时调用，避免残留节点长期积累。

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) 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) HealthScore 

func (cc *ConnectionController) HealthScore() float64

HealthScore returns a 0–1 health signal derived from read/connection failure counts.

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)

RecordConnectionFailure increments observability counters only. It does not authorize reconnect; use driver.ConnectionManager for that.

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) 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 DLT645PointTemplate 

type DLT645PointTemplate struct {
	ID       string
	Name     string
	DataID   string
	DataType string
	Scale    float64
	Unit     string
}

DLT645PointTemplate defines a standard DL/T 645-2007 data identifier point.

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()

func (*DataPipeline) Stop 

func (dp *DataPipeline) Stop()

Stop 关闭 pipeline goroutine，确保后续不会再触发 batch handler 写数据库。 signalChan 永不关闭，避免并发 Push 向其发送时因通道已关闭而 panic。

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 // 运行时状态
	// contains filtered or unexported fields
}

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

func (*DeviceNodeTemplate) RuntimeSnapshot added in v0.1.1 

func (n *DeviceNodeTemplate) RuntimeSnapshot() NodeRuntimeState

RuntimeSnapshot 返回节点运行时状态的一致性快照副本。 调用方不得在无锁状态下直接读取 node.Runtime 字段，统一通过本方法取快照， 由 node.mu 保证多字段的原子可见性。

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) HistoryBucketNames 

func (m *DeviceStorageManager) HistoryBucketNames() []string

HistoryBucketNames returns device_history_* bucket names for enabled storage configs.

func (*DeviceStorageManager) IsStorageEnabled 

func (m *DeviceStorageManager) IsStorageEnabled(deviceID string) bool

IsStorageEnabled reports whether history storage is enabled for a device. Used by MCP get_point_history to give actionable feedback when no data exists. IsStorageEnabled 报告设备是否已启用历史存储

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 DriverCircuitBreaker 

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

func NewDriverCircuitBreaker 

func NewDriverCircuitBreaker() *DriverCircuitBreaker

func (*DriverCircuitBreaker) Allow 

func (cb *DriverCircuitBreaker) Allow(key string) bool

func (*DriverCircuitBreaker) DeviceSnapshot 

func (cb *DriverCircuitBreaker) DeviceSnapshot(key string) map[string]any

func (*DriverCircuitBreaker) OpenTotal 

func (cb *DriverCircuitBreaker) OpenTotal() uint64

func (*DriverCircuitBreaker) Record 

func (cb *DriverCircuitBreaker) Record(key string, success bool, timeout bool)

func (*DriverCircuitBreaker) RejectTotal 

func (cb *DriverCircuitBreaker) RejectTotal() uint64

func (*DriverCircuitBreaker) Reset 

func (cb *DriverCircuitBreaker) Reset(key string)

func (*DriverCircuitBreaker) SetEventHandler 

func (cb *DriverCircuitBreaker) SetEventHandler(fn CircuitBreakerEventHandler)

func (*DriverCircuitBreaker) SetOpenedAtForTest 

func (cb *DriverCircuitBreaker) SetOpenedAtForTest(key string, openedAt time.Time)

SetOpenedAtForTest adjusts the open timestamp (integration tests only).

func (*DriverCircuitBreaker) Snapshot 

func (cb *DriverCircuitBreaker) Snapshot() map[string]any

func (*DriverCircuitBreaker) State 

func (cb *DriverCircuitBreaker) State(key string) CircuitState

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) ClearEdgeLogs 

func (em *EdgeComputeManager) ClearEdgeLogs() (EdgeLogsClearResult, error)

ClearEdgeLogs removes historical edge events, failures, and minute-level logs. Rule definitions and runtime state (current_status, execution_phase, windows) are preserved.

func (*EdgeComputeManager) ClearRuntimeState 

func (em *EdgeComputeManager) ClearRuntimeState()

ClearRuntimeState 清空边缘规则运行时状态（内存态，与 runtime.db 清理配合使用）。

func (*EdgeComputeManager) DeleteRule 

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

func (*EdgeComputeManager) GetEvents 

func (em *EdgeComputeManager) GetEvents(ruleID string, limit int) []model.EdgeRuleEvent

func (*EdgeComputeManager) GetFailedActions 

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

func (*EdgeComputeManager) GetFailures 

func (em *EdgeComputeManager) GetFailures(ruleID string, limit int) []model.EdgeFailureRecord

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) 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 error logs based on time range and optional rule ID.

func (*EdgeComputeManager) RuntimeLogStats 

func (em *EdgeComputeManager) RuntimeLogStats() (eventsMem, failuresMem, minuteCache int)

RuntimeLogStats returns in-memory edge log buffer sizes for database stats.

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) SetBatchWindow 

func (em *EdgeComputeManager) SetBatchWindow(d time.Duration)

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"`
	CacheSize             int   `json:"cache_size"`
	WindowBufferTotal     int   `json:"window_buffer_total"`
	PendingSchedulerTasks int   `json:"pending_scheduler_tasks"`
	MinuteCacheSize       int   `json:"minute_cache_size"`
	EventBufferSize       int   `json:"event_buffer_size"`
	FailureBufferSize     int   `json:"failure_buffer_size"`
	BatchWindowMs         int64 `json:"batch_window_ms"`
	RulesTriggered        int64 `json:"rules_triggered"`
	RulesExecuted         int64 `json:"rules_executed"`
	RulesDropped          int64 `json:"rules_dropped"`
	RulesCoalesced        int64 `json:"rules_coalesced"`
	RulesDebounced        int64 `json:"rules_debounced"`
}

type EdgeLogFilter 

type EdgeLogFilter struct {
	RuleID    string
	Category  string
	ChannelID string
	DeviceID  string
}

EdgeLogFilter selects persisted edge error logs.

type EdgeLogsClearResult 

type EdgeLogsClearResult struct {
	EventsMemory   int      `json:"events_memory"`
	FailuresMemory int      `json:"failures_memory"`
	MinuteCache    int      `json:"minute_cache"`
	Buckets        []string `json:"buckets"`
}

EdgeLogsClearResult summarizes what was cleared by ClearEdgeLogs.

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) GetBackpressure 

func (el *ExecutionLayer) GetBackpressure() *BackpressureController

func (*ExecutionLayer) GetCircuitBreaker 

func (el *ExecutionLayer) GetCircuitBreaker() *DriverCircuitBreaker

func (*ExecutionLayer) GetDriver 

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

func (*ExecutionLayer) GetSerialQueueDepths 

func (el *ExecutionLayer) GetSerialQueueDepths() map[string]int

func (*ExecutionLayer) ReduceBackpressureRate 

func (el *ExecutionLayer) ReduceBackpressureRate(factor float64)

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) SetCircuitBreakerEventHandler 

func (el *ExecutionLayer) SetCircuitBreakerEventHandler(fn CircuitBreakerEventHandler)

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) BindShadowCore 

func (nm *NorthboundManager) BindShadowCore(sc *ShadowCore)

BindShadowCore attaches ShadowCore → EAN Event publishing on the hot path. Safe to call multiple times; subscription is registered once.

func (*NorthboundManager) DeleteBACnetServerConfig 

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

DeleteBACnetServerConfig 删除 BACnet Server 配置

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) GetBACnetServerStats 

func (nm *NorthboundManager) GetBACnetServerStats(id string) (bacnet.Stats, error)

GetBACnetServerStats 获取 BACnet Server 统计信息

func (*NorthboundManager) GetBACnetServerWriteHistory 

func (nm *NorthboundManager) GetBACnetServerWriteHistory(id string, limit int) ([]bacnet.WriteHistoryItem, error)

GetBACnetServerWriteHistory 获取 BACnet Server 写入历史

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. Virtual shadow devices do not emit southbound status events; only physical devices are handled here.

func (*NorthboundManager) PublishDeviceReport 

func (nm *NorthboundManager) PublishDeviceReport()

PublishDeviceReport republishes the device inventory (edgeCore/devices/report) to all edgeOS clients. Called when channels/devices are added/updated or the northbound Devices mapping changes, so EdgeOS's device list stays in sync without re-connecting.

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) RebuildBACnetServers 

func (nm *NorthboundManager) RebuildBACnetServers()

RebuildBACnetServers 同步所有已启用 BACnet Server 的地址空间

func (*NorthboundManager) RebuildOPCUAServers 

func (nm *NorthboundManager) RebuildOPCUAServers()

RebuildOPCUAServers 同步所有已启用 OPC UA Server 的地址空间（默认不停服； 结构性配置变更才 Stop+Start）。不影响网关主进程与南向采集。

func (*NorthboundManager) RefreshEANEventPublishers 

func (nm *NorthboundManager) RefreshEANEventPublishers()

RefreshEANEventPublishers re-collects EventPublishers from connected edgeOS clients.

func (*NorthboundManager) SetChannelManager 

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

SetChannelManager sets the ChannelManager reference for device lookups

func (*NorthboundManager) SetVirtualShadowManager 

func (nm *NorthboundManager) SetVirtualShadowManager(vsm *VirtualShadowManager)

func (*NorthboundManager) Start 

func (nm *NorthboundManager) Start()

func (*NorthboundManager) Stop 

func (nm *NorthboundManager) Stop()

func (*NorthboundManager) SyncBACnetServer 

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

SyncBACnetServer 同步指定 BACnet Server 的地址空间

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) UpsertBACnetServerConfig 

func (nm *NorthboundManager) UpsertBACnetServerConfig(cfg model.BACnetServerConfig) (model.BACnetServerConfig, string, error)

UpsertBACnetServerConfig 更新或插入 BACnet Server 配置

func (*NorthboundManager) UpsertEdgeOSMQTTConfig 

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

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

func (*NorthboundManager) UpsertEdgeOSNATSConfig 

func (nm *NorthboundManager) UpsertEdgeOSNATSConfig(cfg model.EdgeOSNATSConfig) (string, 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) (string, error)

func (*NorthboundManager) UpsertOPCUAConfig 

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

func (*NorthboundManager) UpsertSparkplugBConfig 

func (nm *NorthboundManager) UpsertSparkplugBConfig(cfg model.SparkplugBConfig) (string, 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 ProtocolCongestionController 

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

ProtocolCongestionController is deprecated; protocol rate limiting lives in BackpressureController.AllowWithReason (layer 3). Kept for test compatibility.

func NewProtocolCongestionController 

func NewProtocolCongestionController() *ProtocolCongestionController

func (*ProtocolCongestionController) Allow 

func (pc *ProtocolCongestionController) Allow(protocol string) bool

func (*ProtocolCongestionController) RejectTotal 

func (pc *ProtocolCongestionController) RejectTotal() uint64

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 RejectReason 

type RejectReason string
const (
	RejectNone            RejectReason = ""
	RejectGlobalSemaphore RejectReason = "global_semaphore"
	RejectDeviceSemaphore RejectReason = "device_semaphore"
	RejectProtocolRate    RejectReason = "protocol_rate"
)

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) ExecuteTask 

func (se *ScanEngine) ExecuteTask(task *ScanTask) *ExecuteResult

func (*ScanEngine) GetActiveTaskCount 

func (se *ScanEngine) GetActiveTaskCount() int

func (*ScanEngine) GetCircuitBreaker 

func (se *ScanEngine) GetCircuitBreaker() *DriverCircuitBreaker

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) IsRunning 

func (se *ScanEngine) IsRunning() bool

func (*ScanEngine) OperationalSnapshot 

func (se *ScanEngine) OperationalSnapshot() map[string]any

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) SetCircuitBreakerEventHandler 

func (se *ScanEngine) SetCircuitBreakerEventHandler(fn CircuitBreakerEventHandler)

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) SetShadowIngress 

func (se *ScanEngine) SetShadowIngress(si *ShadowIngress)

func (*ScanEngine) Stop 

func (se *ScanEngine) Stop()

func (*ScanEngine) UnregisterDriver 

func (se *ScanEngine) UnregisterDriver(deviceKey string)

func (*ScanEngine) UpdateTaskDriverConfig 

func (se *ScanEngine) UpdateTaskDriverConfig(deviceKey string, updates map[string]any)

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) UpdateDeviceDriverConfig 

func (a *ScanEngineAdapter) UpdateDeviceDriverConfig(deviceID string, updates map[string]any)

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           // 并发采集 worker 数
	MaxQueueSize    int           // 就绪队列最大长度（超限时资源排队）
	GoroutineLimit  int           // 协程上限
	ConnectionLimit int           // 并发连接上限
}

ScanEngineConfig 采集调度引擎配置。参照 Kepware：仅保留间隔节奏与并发限流， 去除截止期/抖动/防饿死等复杂调度参数，靠冷却降级与自愈保障稳定性。

type ScanEngineMetrics 

type ScanEngineMetrics struct {
	TasksExecuted         atomic.Uint64
	TasksSucceeded        atomic.Uint64
	TasksFailed           atomic.Uint64
	StarvationRescues     atomic.Uint64
	TaskOverdueTotal      atomic.Uint64
	ScanMissDeadlineTotal atomic.Uint64
	ScanDriftMicrosTotal  atomic.Uint64
	ScanDriftSamples      atomic.Uint64
	TotalScanLagMicros    atomic.Uint64
	ScanLagSamples        atomic.Uint64
	MaxScanLagMicros      atomic.Uint64
	IntervalAdjustedTotal atomic.Uint64

	// 恢复能力指标（对标 Kepware 诊断）：采集过程中被 recover 的 panic 次数
	// 与成功重新入队的次数。 nonzero TaskPanicsTotal 是必须告警的严重信号。
	TaskPanicsTotal     atomic.Uint64
	TaskRecoveriesTotal atomic.Uint64
	// contains filtered or unexported fields
}

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

func (*ScanEngineMetrics) AdaptiveSlowdownFactor 

func (m *ScanEngineMetrics) AdaptiveSlowdownFactor() float64

func (*ScanEngineMetrics) AvgLagMs 

func (m *ScanEngineMetrics) AvgLagMs() float64

func (*ScanEngineMetrics) ChannelSLAWarnings 

func (m *ScanEngineMetrics) ChannelSLAWarnings(channelID string, cb *DriverCircuitBreaker, deviceKeys []string) []map[string]any

func (*ScanEngineMetrics) ChannelSnapshot 

func (m *ScanEngineMetrics) ChannelSnapshot(channelID string) map[string]any

func (*ScanEngineMetrics) GlobalFailRate 

func (m *ScanEngineMetrics) GlobalFailRate() float64

func (*ScanEngineMetrics) RecordDrift 

func (m *ScanEngineMetrics) RecordDrift(driftMicros int64)

func (*ScanEngineMetrics) RecordDriftForChannel 

func (m *ScanEngineMetrics) RecordDriftForChannel(channelID string, driftMicros int64)

func (*ScanEngineMetrics) RecordExecute 

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

func (*ScanEngineMetrics) RecordExecuteForChannel 

func (m *ScanEngineMetrics) RecordExecuteForChannel(channelID string, success bool, lagMicros int64)

func (*ScanEngineMetrics) RecordIntervalAdjusted 

func (m *ScanEngineMetrics) RecordIntervalAdjusted()

func (*ScanEngineMetrics) RecordMissDeadline 

func (m *ScanEngineMetrics) RecordMissDeadline()

func (*ScanEngineMetrics) RecordMissDeadlineForChannel 

func (m *ScanEngineMetrics) RecordMissDeadlineForChannel(channelID string)

func (*ScanEngineMetrics) RecordOverdue 

func (m *ScanEngineMetrics) RecordOverdue()

func (*ScanEngineMetrics) RecordStarvationRescue 

func (m *ScanEngineMetrics) RecordStarvationRescue()

func (*ScanEngineMetrics) ResetWindow 

func (m *ScanEngineMetrics) ResetWindow()

ResetWindow clears cumulative counters and lag samples for steady-state measurement windows (benchmarks/soak).

func (*ScanEngineMetrics) SLAWarnings 

func (m *ScanEngineMetrics) SLAWarnings(cb *DriverCircuitBreaker) []map[string]any

func (*ScanEngineMetrics) SetAdaptiveSlowdownFactor 

func (m *ScanEngineMetrics) SetAdaptiveSlowdownFactor(factor float64)

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
	LastScheduledAt     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) LastSuccessTime added in v0.1.2 

func (t *ScanTask) LastSuccessTime() time.Time

LastSuccessTime 返回最近一次成功采集时间（并发安全）。

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) QueueDepths 

func (sqm *SerialQueueManager) QueueDepths() map[string]int

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 NewShadowCoreWithNotifyWorkers 

func NewShadowCoreWithNotifyWorkers(workers int) *ShadowCore

func (*ShadowCore) ApplyShadowWrites 

func (sc *ShadowCore) ApplyShadowWrites(msgs []model.ShadowIngressMessage) error

ApplyShadowWrites applies multiple ingress messages under a single lock, emitting one delta notify per device touched.

func (*ShadowCore) CheckConsistency 

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

func (*ShadowCore) ClearAllShadowDevices 

func (sc *ShadowCore) ClearAllShadowDevices()

ClearAllShadowDevices 清空全部内存态影子设备（含虚拟影子与优化画像）。

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) NotifyWorkerCount 

func (sc *ShadowCore) NotifyWorkerCount() int

func (*ShadowCore) ResolvePublishTarget 

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

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

func (*ShadowCore) RuntimePointStats 

func (sc *ShadowCore) RuntimePointStats() (deviceCount, pointCount int)

RuntimePointStats returns in-memory shadow device and point counts for database stats.

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 更新影子设备的通信画像（无条件，供测试与显式刷新）。

func (*ShadowDeviceOptimizer) UpdateShadowDeviceProfileIfNeeded 

func (sdo *ShadowDeviceOptimizer) UpdateShadowDeviceProfileIfNeeded(deviceID, channelID string, profilePtr **model.DeviceCommunicationProfile) bool

UpdateShadowDeviceProfileIfNeeded 仅在 RTT 变化超过阈值时更新通信画像。 profilePtr 指向当前快照上的 profile 指针；更新时会替换为新克隆。

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
	BatchFlushes    uint64
	LastProcessTime time.Time
}

type ShadowSubscriber 

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

type ShadowWriteRingBuffer 

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

ShadowWriteRingBuffer is a fixed-capacity queue for shadow ingress messages. Producers push message pointers (zero-copy payload); Flush drains and returns the batch for batch apply to ShadowCore.

func NewShadowWriteRingBuffer 

func NewShadowWriteRingBuffer(capacity int) *ShadowWriteRingBuffer

func (*ShadowWriteRingBuffer) Flush 

func (rb *ShadowWriteRingBuffer) Flush() []*model.ShadowIngressMessage

func (*ShadowWriteRingBuffer) Len 

func (rb *ShadowWriteRingBuffer) Len() int

func (*ShadowWriteRingBuffer) Push 

func (rb *ShadowWriteRingBuffer) Push(msg *model.ShadowIngressMessage) bool

type SoakMonitor 

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

SoakMonitor tracks in-process ScanEngine SLA samples for dashboard soak views.

func NewSoakMonitor 

func NewSoakMonitor(cm *ChannelManager) *SoakMonitor

func (*SoakMonitor) Snapshot 

func (sm *SoakMonitor) Snapshot() map[string]any

func (*SoakMonitor) Start 

func (sm *SoakMonitor) Start()

func (*SoakMonitor) Stop 

func (sm *SoakMonitor) Stop()

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) AddRoute 

func (sm *SystemManager) AddRoute(route model.StaticRoute) error

func (*SystemManager) DeleteRoute 

func (sm *SystemManager) DeleteRoute(route model.StaticRoute) error

func (*SystemManager) GetConfig 

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

func (*SystemManager) GetHostnameAccessStatus 

func (sm *SystemManager) GetHostnameAccessStatus() network.HostnameAccessStatus

func (*SystemManager) GetNetworkBackendInfo 

func (sm *SystemManager) GetNetworkBackendInfo() network.BackendInfo

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

func (*SystemManager) ValidateConnectivity 

func (sm *SystemManager) ValidateConnectivity(targets []model.ConnectivityTarget) (model.ConnectivityReport, 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 ThrottleContext 

type ThrottleContext struct {
	DeviceKey   string
	Protocol    string
	DeviceLimit int
}

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.