8
8
import org .iot .dsa .node .DSInfo ;
9
9
import org .iot .dsa .node .DSLong ;
10
10
import org .iot .dsa .node .DSMap ;
11
- import org .iot .dsa .node .DSNode ;
12
11
import org .iot .dsa .node .DSString ;
13
12
import org .iot .dsa .node .DSValueType ;
14
13
import org .iot .dsa .node .action .ActionInvocation ;
15
14
import org .iot .dsa .node .action .ActionResult ;
16
15
import org .iot .dsa .node .action .DSAction ;
17
- import org .iot .dsa .node .event .DSIEvent ;
18
- import org .iot .dsa .node .event .DSISubscriber ;
19
- import org .iot .dsa .node .event .DSITopic ;
16
+ import org .iot .dsa .node .event .DSEventFilter ;
20
17
import org .iot .dsa .util .DSException ;
21
18
22
19
/**
23
20
* The root node of this link.
24
21
*/
25
22
public class MainNode extends DSMainNode implements PurgeSettings {
23
+
26
24
private static final Object requesterLock = new Object ();
27
25
private static DSIRequester requester ;
28
26
public static MainNode instance ;
29
-
30
- private DSInfo purgeEnabled = getInfo (Constants .BUFFER_PURGE_ENABLED );
31
27
private DSInfo maxBufferSize = getInfo (Constants .BUFFER_MAX_SIZE );
28
+ private DSInfo purgeEnabled = getInfo (Constants .BUFFER_PURGE_ENABLED );
32
29
33
30
public MainNode () {
34
31
}
35
32
33
+ public long getMaxSizeInBytes () {
34
+ return maxBufferSize .getValue ().toElement ().toLong ();
35
+ }
36
+
36
37
public static DSIRequester getRequester () {
37
38
synchronized (requesterLock ) {
38
39
while (requester == null ) {
@@ -46,6 +47,10 @@ public static DSIRequester getRequester() {
46
47
}
47
48
}
48
49
50
+ public boolean isPurgeEnabled () {
51
+ return purgeEnabled .getValue ().toElement ().toBoolean ();
52
+ }
53
+
49
54
public static void setRequester (DSIRequester requester ) {
50
55
synchronized (requesterLock ) {
51
56
MainNode .requester = requester ;
@@ -63,30 +68,23 @@ protected void declareDefaults() {
63
68
declareDefault (Constants .ACT_ADD_BASIC_CONN , makeAddBasicConnectionAction ());
64
69
declareDefault (Constants .ACT_ADD_OAUTH_CLIENT_CONN , makeAddOauthClientConnectionAction ());
65
70
declareDefault (Constants .ACT_ADD_OAUTH_PASSWORD_CONN , makeAddOauthPassConnectionAction ());
66
- declareDefault (Constants .BUFFER_PURGE_ENABLED , DSBool .FALSE , "Whether old unsent records should automatically be purged from the buffer when the buffer gets too large" );
67
- declareDefault (Constants .BUFFER_MAX_SIZE , DSLong .valueOf (1074000000 ), "Maximum size of buffer in bytes; only applies if auto-purge is enabled" );
68
- declareDefault ("Docs" , DSString .valueOf ("https://github.com/iot-dsa-v2/dslink-java-v2-restadapter/blob/develop/docs/Usage_Guide.md" )).setTransient (true ).setReadOnly (true );
69
- }
70
-
71
- public boolean isPurgeEnabled () {
72
- return purgeEnabled .getValue ().toElement ().toBoolean ();
73
- }
74
-
75
- public long getMaxSizeInBytes () {
76
- return maxBufferSize .getValue ().toElement ().toLong ();
71
+ declareDefault (Constants .BUFFER_PURGE_ENABLED , DSBool .FALSE ,
72
+ "Whether old unsent records should automatically be purged from the buffer when the buffer gets too large" );
73
+ declareDefault (Constants .BUFFER_MAX_SIZE , DSLong .valueOf (1074000000 ),
74
+ "Maximum size of buffer in bytes; only applies if auto-purge is enabled" );
75
+ declareDefault ("Docs" , DSString .valueOf (
76
+ "https://github.com/iot-dsa-v2/dslink-java-v2-restadapter/blob/develop/docs/Usage_Guide.md" ))
77
+ .setTransient (true ).setReadOnly (true );
77
78
}
78
79
79
80
@ Override
80
81
protected void onStarted () {
81
82
instance = this ;
82
- getLink ().getConnection ().subscribe (
83
- DSLinkConnection .CONNECTED , null , null , new DSISubscriber () {
84
- @ Override
85
- public void onEvent (DSNode node , DSInfo child ,
86
- DSIEvent event ) {
87
- MainNode .setRequester (getLink ().getConnection ().getRequester ());
88
- }
89
- });
83
+ getLink ().getUpstream ().subscribe (new DSEventFilter (
84
+ ((event , node , child , data ) -> MainNode .setRequester (
85
+ getLink ().getUpstream ().getRequester ())),
86
+ DSLinkConnection .CONNECTED_EVENT ,
87
+ null ));
90
88
}
91
89
92
90
private void addBasicConnection (DSMap parameters ) {
0 commit comments