4747import org .eclipse .milo .opcua .sdk .server .nodes .UaFolderNode ;
4848import org .eclipse .milo .opcua .sdk .server .nodes .UaNode ;
4949import org .eclipse .milo .opcua .sdk .server .nodes .UaVariableNode ;
50- import org .eclipse .milo .opcua .sdk . server . util . SubscriptionModel ;
50+ import org .eclipse .milo .opcua .stack . core . AttributeId ;
5151import org .eclipse .milo .opcua .stack .core .Identifiers ;
5252import org .eclipse .milo .opcua .stack .core .UaException ;
5353import org .eclipse .milo .opcua .stack .core .security .SecurityPolicy ;
5757import org .eclipse .milo .opcua .stack .core .types .builtin .NodeId ;
5858import org .eclipse .milo .opcua .stack .core .types .builtin .StatusCode ;
5959import org .eclipse .milo .opcua .stack .core .types .builtin .Variant ;
60+ import org .eclipse .milo .opcua .stack .core .types .structured .ReadValueId ;
6061import org .slf4j .Logger ;
6162import org .slf4j .LoggerFactory ;
6263
7172import java .util .Objects ;
7273import java .util .Set ;
7374import java .util .UUID ;
75+ import java .util .concurrent .ConcurrentHashMap ;
76+ import java .util .concurrent .ConcurrentMap ;
77+ import java .util .concurrent .CopyOnWriteArrayList ;
78+ import java .util .concurrent .ScheduledFuture ;
79+ import java .util .concurrent .TimeUnit ;
7480import java .util .stream .Collectors ;
7581
7682public class OpcUaNameSpace extends ManagedNamespaceWithLifecycle {
7783 private static final Logger LOGGER = LoggerFactory .getLogger (OpcUaNameSpace .class );
7884 public static final String NAMESPACE_URI = "urn:apache:iotdb:opc-server" ;
79- private final SubscriptionModel subscriptionModel ;
8085 private final OpcUaServerBuilder builder ;
8186
87+ // Do not use subscription model because the original subscription model has some bugs
88+ private final ConcurrentMap <NodeId , List <DataItem >> nodeSubscriptions = new ConcurrentHashMap <>();
89+
90+ // Debounce task cache: used to merge updates within a short period of time, avoiding unnecessary
91+ // duplicate pushes
92+ private final ConcurrentMap <NodeId , ScheduledFuture <?>> debounceTasks = new ConcurrentHashMap <>();
93+ // Debounce interval: within 10ms, the same node is updated multiple times, and only the last one
94+ // will be pushed (can be adjusted according to your site delay requirements, the minimum can be
95+ // set to 1ms)
96+ private final long debounceIntervalMs ;
97+
8298 public OpcUaNameSpace (final OpcUaServer server , final OpcUaServerBuilder builder ) {
8399 super (server , NAMESPACE_URI );
84100 this .builder = builder ;
101+ debounceIntervalMs = builder .getDebounceTimeMs ();
85102
86- subscriptionModel = new SubscriptionModel (server , this );
87- getLifecycleManager ().addLifecycle (subscriptionModel );
88103 getLifecycleManager ()
89104 .addLifecycle (
90105 new Lifecycle () {
@@ -291,7 +306,7 @@ private void transferTabletRowForClientServerModel(
291306 if (Objects .isNull (measurementNode .getValue ())
292307 || Objects .isNull (measurementNode .getValue ().getSourceTime ())
293308 || measurementNode .getValue ().getSourceTime ().getUtcTime () < utcTimestamp ) {
294- measurementNode .setValue ( dataValue );
309+ notifyNodeValueChange ( measurementNode .getNodeId (), dataValue , measurementNode );
295310 }
296311 } else {
297312 value = values .get (i );
@@ -311,9 +326,11 @@ private void transferTabletRowForClientServerModel(
311326 if (Objects .isNull (valueNode .getValue ())
312327 || Objects .isNull (valueNode .getValue ().getSourceTime ())
313328 || valueNode .getValue ().getSourceTime ().getUtcTime () < timestamp ) {
314- valueNode .setValue (
329+ notifyNodeValueChange (
330+ valueNode .getNodeId (),
315331 new DataValue (
316- new Variant (value ), currentQuality , new DateTime (timestamp ), new DateTime ()));
332+ new Variant (value ), currentQuality , new DateTime (timestamp ), new DateTime ()),
333+ valueNode );
317334 }
318335 }
319336 }
@@ -546,24 +563,131 @@ public static NodeId convertToOpcDataType(final TSDataType type) {
546563 }
547564 }
548565
566+ /**
567+ * On point value changing, notify all subscribed clients proactively
568+ *
569+ * @param nodeId NodeId of the changing node
570+ * @param newValue New value of the node (DataValue object containing value, status code, and
571+ * timestamp)
572+ * @param variableNode Corresponding UaVariableNode instance, used to update the local cached
573+ * value of the node
574+ */
575+ public void notifyNodeValueChange (
576+ NodeId nodeId , DataValue newValue , UaVariableNode variableNode ) {
577+ // 1. Update the local cached value of the node
578+ variableNode .setValue (newValue );
579+
580+ // 2. If there are no subscribers, return directly without doing any extra operations
581+ List <DataItem > subscribedItems = nodeSubscriptions .get (nodeId );
582+ if (subscribedItems == null || subscribedItems .isEmpty ()) {
583+ return ;
584+ }
585+
586+ // 2. Debounce+Async Push: Asynchronously push the expensive push operation, while merging
587+ // high-frequency repeated updates
588+ debounceTasks .compute (
589+ nodeId ,
590+ (k , oldTask ) -> {
591+ // If there is already a pending push task, cancel it, we only need the latest value
592+ if (oldTask != null && !oldTask .isDone ()) {
593+ oldTask .cancel (false );
594+ }
595+
596+ // Submit the push task to the Milo's scheduled thread pool, delay DEBOUNCE_INTERVAL_MS
597+ // execution
598+ return getServer ()
599+ .getScheduledExecutorService ()
600+ .schedule (
601+ () -> {
602+ try {
603+ // Batch push changes to all subscribers, this time-consuming operation is put
604+ // into the thread pool, not blocking your data update thread
605+ for (DataItem item : subscribedItems ) {
606+ try {
607+ item .setValue (newValue );
608+ } catch (Exception e ) {
609+ // Single client push failure does not affect other clients
610+ LOGGER .warn (
611+ "Failed to push value change to client, nodeId={}" , nodeId , e );
612+ }
613+ }
614+ } finally {
615+ // Task execution completed, clean up the debounce cache
616+ debounceTasks .remove (nodeId );
617+ }
618+ },
619+ debounceIntervalMs ,
620+ TimeUnit .MILLISECONDS );
621+ });
622+ }
623+
549624 @ Override
550625 public void onDataItemsCreated (final List <DataItem > dataItems ) {
551- subscriptionModel .onDataItemsCreated (dataItems );
626+ for (DataItem item : dataItems ) {
627+ final ReadValueId readValueId = item .getReadValueId ();
628+ // Only handle Value attribute subscription (align with the original SubscriptionModel logic,
629+ // ignore other attribute subscriptions)
630+ if (!AttributeId .Value .isEqual (readValueId .getAttributeId ())) {
631+ continue ;
632+ }
633+ final NodeId nodeId = readValueId .getNodeId ();
634+
635+ // 1. Add the new subscription item to the subscription mapping
636+ nodeSubscriptions .compute (
637+ nodeId ,
638+ (k , existingList ) -> {
639+ List <DataItem > list =
640+ existingList != null ? existingList : new CopyOnWriteArrayList <>();
641+ list .add (item );
642+ return list ;
643+ });
644+
645+ // 2. 【Key Optimization】Proactively push the current node's initial value when the new
646+ // subscription item is created
647+ // Eliminate Bad_WaitingForInitialData, no need to wait for any polling
648+ try {
649+ UaVariableNode node = (UaVariableNode ) getNodeManager ().getNode (nodeId ).orElse (null );
650+ if (node != null && node .getValue () != null ) {
651+ // Immediately push the current value to the new subscriber, the client will instantly be
652+ // able to get the initial data
653+ item .setValue (node .getValue ());
654+ }
655+ } catch (Exception e ) {
656+ LOGGER .warn ("Failed to send initial value to new subscription, nodeId={}" , nodeId , e );
657+ }
658+ }
552659 }
553660
554661 @ Override
555662 public void onDataItemsModified (final List <DataItem > dataItems ) {
556- subscriptionModel .onDataItemsModified (dataItems );
663+ // Push mode, client modifies subscription parameters (e.g. sampling interval) has no effect on
664+ // our active push, no additional processing is needed
557665 }
558666
559667 @ Override
560668 public void onDataItemsDeleted (final List <DataItem > dataItems ) {
561- subscriptionModel .onDataItemsDeleted (dataItems );
669+ for (DataItem item : dataItems ) {
670+ final ReadValueId readValueId = item .getReadValueId ();
671+ if (!AttributeId .Value .isEqual (readValueId .getAttributeId ())) {
672+ continue ;
673+ }
674+ final NodeId nodeId = readValueId .getNodeId ();
675+
676+ // When the client cancels the subscription, remove this subscription item from the mapping
677+ nodeSubscriptions .computeIfPresent (
678+ nodeId ,
679+ (k , existingList ) -> {
680+ existingList .remove (item );
681+ // Automatically clean up the key when there are no subscribers, save memory
682+ return existingList .isEmpty () ? null : existingList ;
683+ });
684+ }
562685 }
563686
564687 @ Override
565688 public void onMonitoringModeChanged (final List <MonitoredItem > monitoredItems ) {
566- subscriptionModel .onMonitoringModeChanged (monitoredItems );
689+ // Push mode, monitoring mode change has no effect on active push, no additional processing is
690+ // needed
567691 }
568692
569693 /////////////////////////////// Conflict detection ///////////////////////////////
@@ -573,8 +697,14 @@ public void checkEquals(
573697 final String password ,
574698 final String securityDir ,
575699 final boolean enableAnonymousAccess ,
576- final Set <SecurityPolicy > securityPolicies ) {
700+ final Set <SecurityPolicy > securityPolicies ,
701+ final long debounceTimeMs ) {
577702 builder .checkEquals (
578- user , password , Paths .get (securityDir ), enableAnonymousAccess , securityPolicies );
703+ user ,
704+ password ,
705+ Paths .get (securityDir ),
706+ enableAnonymousAccess ,
707+ securityPolicies ,
708+ debounceTimeMs );
579709 }
580710}
0 commit comments