core

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: GPL-2.0, GPL-3.0 Imports: 37 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 AdaptiveThrottle 

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

func NewAdaptiveThrottle 

func NewAdaptiveThrottle(metrics *ScanEngineMetrics) *AdaptiveThrottle

func (*AdaptiveThrottle) ApplyInterval 

func (at *AdaptiveThrottle) ApplyInterval(task *ScanTask) bool

func (*AdaptiveThrottle) DeviceFactor 

func (at *AdaptiveThrottle) DeviceFactor(deviceKey string) float64

func (*AdaptiveThrottle) EffectiveInterval 

func (at *AdaptiveThrottle) EffectiveInterval(base time.Duration) time.Duration

func (*AdaptiveThrottle) Factor 

func (at *AdaptiveThrottle) Factor() float64

func (*AdaptiveThrottle) Refresh 

func (at *AdaptiveThrottle) Refresh(queueDepth, queueLimit int, failRate, avgRTTMs float64) float64

func (*AdaptiveThrottle) UpdateDeviceRTT 

func (at *AdaptiveThrottle) UpdateDeviceRTT(deviceKey string, rttMs float64)

UpdateDeviceRTT records per-device RTT and raises interval factor when RTT > 2× baseline.

type AggregatedStats 

type AggregatedStats struct {
	SuccessCount int
	FailCount    int
	FailRate     float64
	LastError    error
}

AggregatedStats summarizes feedback within a flush window.

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

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

FeedbackAggregator batches scheduling feedback before updateTaskState (v5.2 sketch).

func NewFeedbackAggregator 

func NewFeedbackAggregator(window time.Duration, onFlush func(string, AggregatedStats)) *FeedbackAggregator

func (*FeedbackAggregator) Start 

func (fa *FeedbackAggregator) Start()

func (*FeedbackAggregator) Stop 

func (fa *FeedbackAggregator) Stop()

func (*FeedbackAggregator) Submit 

func (fa *FeedbackAggregator) Submit(ev FeedbackEvent)

func (*FeedbackAggregator) Window 

func (fa *FeedbackAggregator) Window() time.Duration

type FeedbackEvent 

type FeedbackEvent struct {
	DeviceKey string
	TaskID    string
	Success   bool
	Err       error
	At        time.Time
	LagMicros int64
}

FeedbackEvent is a single task execution outcome for aggregation.

type GCMonitor 

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

func NewGCMonitor 

func NewGCMonitor(onHighPause func(pauseMaxMs float64)) *GCMonitor

func (*GCMonitor) Metrics 

func (m *GCMonitor) Metrics() *GCMonitorMetrics

func (*GCMonitor) Start 

func (m *GCMonitor) Start()

func (*GCMonitor) Stop 

func (m *GCMonitor) Stop()

type GCMonitorMetrics 

type GCMonitorMetrics struct {
	PauseMaxMs        atomic.Uint64
	AllocRateBytesSec atomic.Uint64
}

func (*GCMonitorMetrics) Snapshot 

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

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

func (se *ScanEngine) GetGCMonitor() *GCMonitor

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
	MaxQueueSize      int
	AntiStarvationSec int
	PriorityLevels    int
	GoroutineLimit    int
	ConnectionLimit   int
	JitterBound       time.Duration
}

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
	DeadlineAt          time.Time
	PhaseOffset         time.Duration
	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) 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.