Skip to content

Commit bc6643b

Browse files
a-hansena-hansen
authored
0.36.0
- Topic refactor. - Connection: "Failure" to "Status Text". - Durable Requester Subscriptions - Backup Service Enhancements - Certification Service Enhancements - Remove index access from DSMap
1 parent b77ff71 commit bc6643b

39 files changed

+1042
-825
lines changed

‎build.gradle

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ subprojects {
44
apply plugin: 'maven'
55

66
group 'org.iot-dsa'
7-
version '0.35.0'
7+
version '0.36.0'
88

99
sourceCompatibility = 1.6
1010
targetCompatibility = 1.6

‎dslink-v2-websocket/src/main/java/org/iot/dsa/dslink/websocket/WsBinaryTransport.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package org.iot.dsa.dslink.websocket;
22

3-
import com.acuity.iot.dsa.dslink.sys.cert.SysCertManager;
3+
import com.acuity.iot.dsa.dslink.sys.cert.SysCertService;
44
import com.acuity.iot.dsa.dslink.transport.BufferedBinaryTransport;
55
import com.acuity.iot.dsa.dslink.transport.DSTransport;
66
import java.io.IOException;
@@ -104,7 +104,7 @@ public DSTransport open() {
104104
URI connUri = new URI(getConnectionUrl());
105105
if ("wss".equalsIgnoreCase(connUri.getScheme())) {
106106
SslEngineConfigurator sslEngineConfigurator = new SslEngineConfigurator(new SslContextConfigurator());
107-
sslEngineConfigurator.setHostnameVerifier(SysCertManager.getInstance().getHostnameVerifier());
107+
sslEngineConfigurator.setHostnameVerifier(SysCertService.getInstance().getHostnameVerifier());
108108
client.getProperties().put(ClientProperties.SSL_ENGINE_CONFIGURATOR, sslEngineConfigurator);
109109
}
110110
client.connectToServer(this, connUri);

‎dslink-v2-websocket/src/main/java/org/iot/dsa/dslink/websocket/WsTextTransport.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
import com.acuity.iot.dsa.dslink.io.DSCharBuffer;
44
import com.acuity.iot.dsa.dslink.io.DSIoException;
5-
import com.acuity.iot.dsa.dslink.sys.cert.SysCertManager;
5+
import com.acuity.iot.dsa.dslink.sys.cert.SysCertService;
66
import com.acuity.iot.dsa.dslink.transport.DSTextTransport;
77
import com.acuity.iot.dsa.dslink.transport.DSTransport;
88
import java.io.IOException;
@@ -156,7 +156,7 @@ public DSTransport open() {
156156
URI connUri = new URI(getConnectionUrl());
157157
if ("wss".equalsIgnoreCase(connUri.getScheme())) {
158158
SslEngineConfigurator sslEngineConfigurator = new SslEngineConfigurator(new SslContextConfigurator());
159-
sslEngineConfigurator.setHostnameVerifier(SysCertManager.getInstance().getHostnameVerifier());
159+
sslEngineConfigurator.setHostnameVerifier(SysCertService.getInstance().getHostnameVerifier());
160160
client.getProperties().put(ClientProperties.SSL_ENGINE_CONFIGURATOR, sslEngineConfigurator);
161161
}
162162
client.connectToServer(this, connUri);

‎dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/requester/DSOutboundSubscribeStub.java

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,18 +13,18 @@
1313
* <p>
1414
* <p>
1515
* There can be multiple subscriptions to a single path. They are all contained in a
16-
* DSOutboundSubscribeStubs object.
16+
* DSOutboundSubscription object.
1717
*
1818
* @author Daniel Shapiro, Aaron Hansen
1919
*/
2020
class DSOutboundSubscribeStub implements OutboundStream {
2121

22-
private DSOutboundSubscribeStub next; //linked list in stubs object
22+
private DSOutboundSubscribeStub next; //linked list in subscription object
2323
private boolean open = true;
2424
private String path;
2525
private int qos = 0;
2626
private OutboundSubscribeHandler request;
27-
private DSOutboundSubscribeStubs stubs;
27+
private DSOutboundSubscription sub;
2828

2929
public DSOutboundSubscribeStub(String path, int qos, OutboundSubscribeHandler request) {
3030
this.path = path;
@@ -40,11 +40,11 @@ public void closeStream() {
4040
}
4141
open = false;
4242
}
43-
stubs.remove(this);
43+
sub.remove(this);
4444
try {
4545
request.onClose();
4646
} catch (Exception x) {
47-
stubs.getSubscriptions().error(path, x);
47+
sub.getSubscriptions().error(path, x);
4848
}
4949
}
5050

@@ -65,14 +65,14 @@ public boolean isStreamOpen() {
6565
return open;
6666
}
6767

68-
public void process(DSDateTime ts, DSElement value, DSStatus status) {
68+
public void update(DSDateTime ts, DSElement value, DSStatus status) {
6969
if (!open) {
7070
return;
7171
}
7272
try {
7373
request.onUpdate(ts, value, status);
7474
} catch (Exception x) {
75-
stubs.getSubscriptions().error(path, x);
75+
sub.getSubscriptions().error(path, x);
7676
}
7777
}
7878

@@ -83,8 +83,8 @@ void setNext(DSOutboundSubscribeStub next) {
8383
this.next = next;
8484
}
8585

86-
void setStubs(DSOutboundSubscribeStubs stubs) {
87-
this.stubs = stubs;
86+
void setSub(DSOutboundSubscription sub) {
87+
this.sub = sub;
8888
}
8989

9090
}

‎dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/requester/DSOutboundSubscribeStubs.java renamed to ‎dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/requester/DSOutboundSubscription.java

Lines changed: 58 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package com.acuity.iot.dsa.dslink.protocol.requester;
22

33
import org.iot.dsa.node.DSElement;
4+
import org.iot.dsa.node.DSNull;
45
import org.iot.dsa.node.DSStatus;
56
import org.iot.dsa.time.DSDateTime;
67

@@ -9,7 +10,7 @@
910
*
1011
* @author Daniel Shapiro, Aaron Hansen
1112
*/
12-
class DSOutboundSubscribeStubs {
13+
class DSOutboundSubscription {
1314

1415
///////////////////////////////////////////////////////////////////////////
1516
// Instance Fields
@@ -21,7 +22,7 @@ class DSOutboundSubscribeStubs {
2122
private DSDateTime lastTs;
2223
private DSElement lastValue;
2324
private String path;
24-
private int qos = 0;
25+
private int qos = -1;
2526
private Integer sid;
2627
private int size;
2728
private State state = State.PENDING_SUBSCRIBE;
@@ -31,7 +32,7 @@ class DSOutboundSubscribeStubs {
3132
// Constructors
3233
///////////////////////////////////////////////////////////////////////////
3334

34-
public DSOutboundSubscribeStubs(
35+
public DSOutboundSubscription(
3536
String path,
3637
DSOutboundSubscriptions subscriptions) {
3738
this.path = path;
@@ -74,7 +75,7 @@ public void onDisconnect() {
7475
}
7576
}
7677

77-
public DSOutboundSubscribeStubs setSid(Integer sid) {
78+
public DSOutboundSubscription setSid(Integer sid) {
7879
this.sid = sid;
7980
return this;
8081
}
@@ -87,30 +88,36 @@ public DSOutboundSubscribeStubs setSid(Integer sid) {
8788
* If already subscribed, will pass the last update to the new subscriber.
8889
*/
8990
void add(DSOutboundSubscribeStub stub) {
91+
int prevQos = qos;
9092
if (stub.getQos() > qos) {
9193
qos = stub.getQos();
9294
}
9395
if (contains(stub)) {
96+
if (qos > prevQos) {
97+
getSubscriptions().sendSubscribe(this);
98+
}
9499
return;
95100
}
96-
stub.setStubs(this);
101+
stub.setSub(this);
97102
if (last == null) {
98103
first = stub;
99104
last = stub;
100105
} else {
101106
last.setNext(stub);
102107
last = stub;
103108
}
104-
//Send the last update to the new subscription
105109
if (++size > 1) {
106110
if (lastValue != null) {
107111
try {
108-
stub.process(lastTs, lastValue, lastStatus);
112+
stub.update(lastTs, lastValue, lastStatus);
109113
} catch (Exception x) {
110114
subscriptions.error(path, x);
111115
}
112116
}
113117
}
118+
if (qos > prevQos) { //need to resubscribe for new qos
119+
getSubscriptions().sendSubscribe(this);
120+
}
114121
}
115122

116123
private boolean contains(DSOutboundSubscribeStub stub) {
@@ -140,21 +147,6 @@ private DSOutboundSubscribeStub predecessor(DSOutboundSubscribeStub stub) {
140147
return cur;
141148
}
142149

143-
void process(DSDateTime ts, DSElement value, DSStatus status) {
144-
DSOutboundSubscribeStub stub = first;
145-
while (stub != null) {
146-
try {
147-
stub.process(ts, value, status);
148-
} catch (Exception x) {
149-
subscriptions.error(path, x);
150-
}
151-
stub = stub.getNext();
152-
}
153-
lastTs = ts;
154-
lastValue = value;
155-
lastStatus = status;
156-
}
157-
158150
void remove(DSOutboundSubscribeStub stub) {
159151
DSOutboundSubscribeStub pred = predecessor(stub);
160152
if (pred == last) { //not contained
@@ -173,6 +165,19 @@ void remove(DSOutboundSubscribeStub stub) {
173165
}
174166
if (--size == 0) {
175167
subscriptions.unsubscribe(this);
168+
} else {
169+
stub = first;
170+
int max = 0;
171+
while (stub != null) {
172+
if (stub.getQos() > max) {
173+
max = stub.getQos();
174+
}
175+
stub = stub.getNext();
176+
}
177+
if (max != qos) {
178+
qos = max;
179+
getSubscriptions().sendSubscribe(this);
180+
}
176181
}
177182
}
178183

@@ -184,6 +189,37 @@ int size() {
184189
return size;
185190
}
186191

192+
void update(DSDateTime ts, DSElement value, DSStatus status) {
193+
DSOutboundSubscribeStub stub = first;
194+
while (stub != null) {
195+
try {
196+
stub.update(ts, value, status);
197+
} catch (Exception x) {
198+
subscriptions.error(path, x);
199+
}
200+
stub = stub.getNext();
201+
}
202+
lastTs = ts;
203+
lastValue = value;
204+
lastStatus = status;
205+
}
206+
207+
void updateDisconnected() {
208+
if (lastStatus == DSStatus.unknown) {
209+
return;
210+
}
211+
lastStatus = DSStatus.unknown;
212+
lastTs = DSDateTime.currentTime();
213+
if (lastValue == null) {
214+
lastValue = DSNull.NULL;
215+
}
216+
DSOutboundSubscribeStub cur = first;
217+
while (cur.getNext() != null) {
218+
cur.update(lastTs, lastValue, lastStatus);
219+
cur = cur.getNext();
220+
}
221+
}
222+
187223
///////////////////////////////////////////////////////////////////////////
188224
// Inner Classes
189225
///////////////////////////////////////////////////////////////////////////

0 commit comments

Comments
 (0)