跳到主要内容

DataChange 与状态回调 (C)

C SDK 把数据变化和订阅状态分别暴露为两个回调签名:

回调注册函数触发时机
DarraUa_OnDataChange_AddNode / _AddNodes / _MonitoredItem_SetCallbackMI 数据变化
DarraUa_OnSubscriptionState_Subscription_SetStateCallback订阅级状态变化 (LATE / EXPIRED / ...)
关联

DataChange 回调​

typedef void (*DarraUa_OnDataChange)(
    DarraUa_SubscriptionHandle  sub,
    DarraUa_MonitoredItemHandle mi,
    const DarraUa_DataValue*    value,
    void*                       user_context);

参数说明​

参数说明
sub触发的订阅句柄
mi触发的 MonitoredItem 本地句柄
valueDataValue, 生命周期 = 回调函数内, 不要持久保存指针
user_context注册时传入的上下文指针

用法​

static void on_change(DarraUa_SubscriptionHandle sub,
                      DarraUa_MonitoredItemHandle mi,
                      const DarraUa_DataValue* dv,
                      void* ctx)
{
    if (!dv) return;

    uint32_t status = DarraUa_DataValue_GetStatus(dv);
    DarraUa_DateTime ts = DarraUa_DataValue_GetSourceTimestamp(dv);
    const DarraUa_Variant* v = DarraUa_DataValue_GetValue(dv);
    DarraUa_BuiltinType type = v ? DarraUa_Variant_GetType(v) : DARRA_UA_TYPE_NULL;

    if (DARRA_UA_STATUS_IS_BAD(status)) {
        printf("[mi=%u] BAD: %s\n", (unsigned)mi, DarraUa_StatusName(status));
        return;
    }

    if (type == DARRA_UA_TYPE_DOUBLE) {
        double d = 0.0;
        DarraUa_Variant_GetDouble(v, &d);
        printf("[mi=%u] %.3f @ ft=%lld\n", (unsigned)mi, d, (long long)ts);
    }
}

user_context 用法​

把每个 MI 的元数据 (NodeId 字符串 / 业务对象指针) 通过 user_context 传入, 回调里按 mi 反查:

typedef struct { const char* node_id; int channel_index; } MiContext;

static MiContext g_ctx[100];

for (int i = 0; i < 100; ++i) {
    g_ctx[i].node_id = "ns=2;s=...";
    g_ctx[i].channel_index = i;
    DarraUa_Subscription_AddNode(sub, g_ctx[i].node_id, DARRA_UA_ATTR_VALUE,
        -1.0, on_change, &g_ctx[i], NULL);
}

static void on_change(DarraUa_SubscriptionHandle sub,
                      DarraUa_MonitoredItemHandle mi,
                      const DarraUa_DataValue* dv, void* ctx)
{
    MiContext* mc = (MiContext*)ctx;
    /* 按 mc->channel_index 找业务通道 */
}

订阅级状态回调​

typedef enum {
    DARRA_UA_SUB_STATE_GOOD              = 0,
    DARRA_UA_SUB_STATE_LATE              = 1,
    DARRA_UA_SUB_STATE_KEEP_ALIVE        = 2,
    DARRA_UA_SUB_STATE_LIFETIME_EXPIRED  = 3,
    DARRA_UA_SUB_STATE_RECREATING        = 4
} DarraUa_SubscriptionState;

typedef void (*DarraUa_OnSubscriptionState)(
    DarraUa_SubscriptionHandle sub,
    DarraUa_SubscriptionState  new_state,
    DarraUa_Status             status,
    void*                      user_context);

DARRA_OPCUA_API DarraUa_Status DARRA_OPCUA_CALL
DarraUa_Subscription_SetStateCallback(
    DarraUa_SubscriptionHandle  sub,
    DarraUa_OnSubscriptionState cb,
    void*                       user_context);

状态含义​

状态含义
GOOD正常, 数据持续推送
LATE服务端发现 PublishRequest 跟不上 publishingInterval, 数据可能延迟
KEEP_ALIVE周期 KeepAlive (无数据变化)
LIFETIME_EXPIREDlifetime_count 内未收到 PublishRequest, 服务端清理订阅
RECREATINGSDK 检测到订阅丢失, 自动重建中

用法​

static void on_sub_state(DarraUa_SubscriptionHandle sub,
                         DarraUa_SubscriptionState s,
                         DarraUa_Status code, void* ctx)
{
    (void)sub; (void)ctx;
    if (s == DARRA_UA_SUB_STATE_LIFETIME_EXPIRED)
        fprintf(stderr, "Subscription expired (%s)\n", DarraUa_StatusName(code));
}

DarraUa_Subscription_SetStateCallback(sub, on_sub_state, NULL);

线程模型​

必须切回 UI 线程

回调在 Stack 内部 Publish 线程 上调用, 不在调用方线程. UI 操作必须 marshal 到 UI 线程, 否则跨线程访问控件会崩.

  • 同一订阅的 DataChange 回调串行 (不并发)
  • 跨订阅的回调可能并发, 共享数据加锁
  • 回调里不要做长操作 (>100 ms), 否则阻塞 Publish 队列
  • 回调里若想再次 Read 同节点, 用 DarraUa_Session_ReadNode 同步重读 (不要拷贝 value 指针出去, 它马上失效)

跨平台原子计数:

#if defined(_WIN32)
#  include <intrin.h>
#  define ATOMIC_INC(p)  _InterlockedIncrement((volatile long*)(p))
#else
#  define ATOMIC_INC(p)  __sync_add_and_fetch((p), 1)
#endif

static volatile long g_callback_count = 0;
static void on_change(...) { ATOMIC_INC(&g_callback_count); }

与 ua.Events.* (C# 概念) 的对比​

C# 提供 ua.Events.DataChange 全局聚合通道. C SDK 没有此设施, 业务自己维护一个全局回调 (即 single_cb 在 _AddNodes 里共用) 起到等价作用. 也可在每个 MI 的回调里用 user_context 做"统一 Sink + 业务分发".

下一步​