Skip to content

Commit 9b4fe02

Browse files
d-shapirod-shapiro
authored
Merge pull request #62 from iot-dsa-v2/develop
rebase daniel-certs
2 parents 54066f3 + 65b6c8c commit 9b4fe02

File tree

16 files changed

+274
-193
lines changed

16 files changed

+274
-193
lines changed

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
[![](https://jitpack.io/v/iot-dsa-v2/sdk-dslink-java-v2.svg)](https://jitpack.io/#iot-dsa-v2/sdk-dslink-java-v2)
33

44
* [Developer Guide](https://iot-dsa-v2.github.io/sdk-dslink-java-v2/)
5-
* [Javadoc](https://iot-dsa-v2.github.io/sdk-dslink-java-v2/javadoc/)
5+
* [Javadoc](https://jitpack.io/com/github/iot-dsa-v2/sdk-dslink-java-v2/dslink-v2/master-SNAPSHOT/javadoc/)
66
* JDK 1.6+
77

88

‎build.gradle‎

Lines changed: 2 additions & 2 deletions
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.32.0'
7+
version '0.34.0'
88

99
sourceCompatibility = 1.6
1010
targetCompatibility = 1.6
@@ -26,5 +26,5 @@ subprojects {
2626
}
2727

2828
task wrapper(type: Wrapper) {
29-
gradleVersion = '4.8.1'
29+
gradleVersion = '4.9'
3030
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public DSTransport close() {
6464
return this;
6565
}
6666
open = false;
67-
debug(debug() ? "WsTextTransport.close()" : null, new Exception());
67+
debug(debug() ? "WsTextTransport.close()" : null);
6868
try {
6969
if (session != null) {
7070
session.close();

‎dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/io/DSCharBuffer.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,6 @@ public synchronized void put(char[] msg, int off, int len) {
160160
offset = 0;
161161
}
162162
}
163-
//System.arraycopy(msg, off, buffer, length + offset, len);
164163
System.arraycopy(msg, off, buffer, offset, len);
165164
length += len;
166165
notifyAll();

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

Lines changed: 58 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,7 @@
44
import com.acuity.iot.dsa.dslink.protocol.responder.DSResponder;
55
import com.acuity.iot.dsa.dslink.transport.DSTransport;
66
import java.io.IOException;
7-
import java.util.LinkedList;
8-
import java.util.List;
7+
import java.util.concurrent.ConcurrentLinkedQueue;
98
import org.iot.dsa.conn.DSConnection;
109
import org.iot.dsa.conn.DSIConnected;
1110
import org.iot.dsa.dslink.DSIRequester;
@@ -49,8 +48,8 @@ public abstract class DSSession extends DSNode implements DSIConnected {
4948
private int messageId = 0;
5049
private int nextMessage = 1;
5150
private final Object outgoingMutex = new Object();
52-
private List<OutboundMessage> outgoingRequests = new LinkedList<OutboundMessage>();
53-
private List<OutboundMessage> outgoingResponses = new LinkedList<OutboundMessage>();
51+
private ConcurrentLinkedQueue<OutboundMessage> outgoingRequests = new ConcurrentLinkedQueue<OutboundMessage>();
52+
private ConcurrentLinkedQueue<OutboundMessage> outgoingResponses = new ConcurrentLinkedQueue<OutboundMessage>();
5453
private DSInfo requesterAllowed = getInfo(REQUESTER_ALLOWED);
5554
private ReadThread readThread;
5655
private WriteThread writeThread;
@@ -78,10 +77,8 @@ public void enqueueOutgoingRequest(OutboundMessage arg) {
7877
if (!isRequesterAllowed()) {
7978
throw new IllegalStateException("Requester not allowed");
8079
}
81-
synchronized (outgoingMutex) {
82-
outgoingRequests.add(arg);
83-
outgoingMutex.notify();
84-
}
80+
outgoingRequests.add(arg);
81+
notifyOutgoing();
8582
}
8683
}
8784

@@ -90,10 +87,8 @@ public void enqueueOutgoingRequest(OutboundMessage arg) {
9087
*/
9188
public void enqueueOutgoingResponse(OutboundMessage arg) {
9289
if (connected) {
93-
synchronized (outgoingMutex) {
94-
outgoingResponses.add(arg);
95-
outgoingMutex.notify();
96-
}
90+
outgoingResponses.add(arg);
91+
notifyOutgoing();
9792
}
9893
}
9994

@@ -175,24 +170,14 @@ protected void declareDefaults() {
175170
* Can return null.
176171
*/
177172
protected OutboundMessage dequeueOutgoingRequest() {
178-
synchronized (outgoingMutex) {
179-
if (!outgoingRequests.isEmpty()) {
180-
return outgoingRequests.remove(0);
181-
}
182-
}
183-
return null;
173+
return outgoingRequests.poll();
184174
}
185175

186176
/**
187177
* Can return null.
188178
*/
189179
protected OutboundMessage dequeueOutgoingResponse() {
190-
synchronized (outgoingMutex) {
191-
if (!outgoingResponses.isEmpty()) {
192-
return outgoingResponses.remove(0);
193-
}
194-
}
195-
return null;
180+
return outgoingResponses.poll();
196181
}
197182

198183
/**
@@ -240,20 +225,25 @@ protected boolean hasAckToSend() {
240225
* Override point, this returns the result of hasMessagesToSend.
241226
*/
242227
protected boolean hasSomethingToSend() {
243-
if (ackToSend >= 0) {
244-
return true;
245-
}
246-
if (hasPingToSend()) {
228+
if (hasAckToSend() || hasPingToSend()) {
247229
return true;
248230
}
249231
if (waitingForAcks()) {
250232
return false;
251233
}
252234
if (!outgoingResponses.isEmpty()) {
253-
return true;
235+
for (OutboundMessage msg : outgoingResponses) {
236+
if (msg.canWrite(this)) {
237+
return true;
238+
}
239+
}
254240
}
255241
if (!outgoingRequests.isEmpty()) {
256-
return true;
242+
for (OutboundMessage msg : outgoingRequests) {
243+
if (msg.canWrite(this)) {
244+
return true;
245+
}
246+
}
257247
}
258248
return false;
259249
}
@@ -282,25 +272,19 @@ protected void onConnected() {
282272
connected = true;
283273
lastTimeRecv = lastTimeSend = System.currentTimeMillis();
284274
readThread = new ReadThread(getConnection().getLink().getLinkName() + " Reader");
285-
writeThread = new WriteThread(getConnection().getLink().getLinkName() + " Writer");
286275
readThread.start();
276+
Thread.yield();
277+
writeThread = new WriteThread(getConnection().getLink().getLinkName() + " Writer");
287278
writeThread.start();
288279
}
289280

290281
/**
291282
* Clear the outgoing queues and waits for the the read and write threads to exit.
292283
*/
293284
protected void onDisconnected() {
294-
synchronized (outgoingMutex) {
295-
outgoingRequests.clear();
296-
outgoingResponses.clear();
297-
outgoingMutex.notifyAll();
298-
}
299-
try {
300-
writeThread.join();
301-
} catch (Exception x) {
302-
debug(getPath(), x);
303-
}
285+
outgoingRequests.clear();
286+
outgoingResponses.clear();
287+
notifyOutgoing();
304288
try {
305289
readThread.join();
306290
} catch (Exception x) {
@@ -319,18 +303,21 @@ protected void onDisconnecting() {
319303
}
320304
connected = false;
321305
notifyOutgoing();
306+
try {
307+
writeThread.join();
308+
} catch (Exception x) {
309+
debug(getPath(), x);
310+
}
311+
//Attempt to exit cleanly, try to get acks for sent messages.
312+
waitForAcks(1000);
322313
}
323314

324315
protected void requeueOutgoingRequest(OutboundMessage arg) {
325-
synchronized (outgoingMutex) {
326-
outgoingRequests.add(arg);
327-
}
316+
outgoingRequests.add(arg);
328317
}
329318

330319
protected void requeueOutgoingResponse(OutboundMessage arg) {
331-
synchronized (outgoingMutex) {
332-
outgoingResponses.add(arg);
333-
}
320+
outgoingResponses.add(arg);
334321
}
335322

336323
/**
@@ -385,6 +372,26 @@ private void verifyLastSend() throws IOException {
385372
}
386373
}
387374

375+
/* Try to exit cleanly, wait for all acks for sent messages. */
376+
private void waitForAcks(long timeout) {
377+
long start = System.currentTimeMillis();
378+
synchronized (outgoingMutex) {
379+
while (getMissingAcks() > 0) {
380+
try {
381+
outgoingMutex.wait(500);
382+
} catch (InterruptedException x) {
383+
warn(getPath(), x);
384+
}
385+
if ((System.currentTimeMillis() - start) > timeout) {
386+
debug(debug() ? String
387+
.format("waitForAcks timeout (%s / %s)", ackRcvd, messageId)
388+
: null);
389+
break;
390+
}
391+
}
392+
}
393+
}
394+
388395
///////////////////////////////////////////////////////////////////////////
389396
// Inner Classes
390397
///////////////////////////////////////////////////////////////////////////
@@ -400,6 +407,7 @@ private class ReadThread extends Thread {
400407
}
401408

402409
public void run() {
410+
debug("Enter DSSession.ReadThread");
403411
DSLinkConnection conn = getConnection();
404412
try {
405413
while (connected) {
@@ -415,6 +423,7 @@ public void run() {
415423
conn.connDown(DSException.makeMessage(x));
416424
}
417425
}
426+
debug("Exit DSSession.ReadThread");
418427
}
419428
}
420429

@@ -430,6 +439,7 @@ private class WriteThread extends Thread {
430439

431440
public void run() {
432441
DSLinkConnection conn = getConnection();
442+
debug("Enter DSSession.WriteThread");
433443
try {
434444
while (connected) {
435445
verifyLastRead();
@@ -454,6 +464,7 @@ public void run() {
454464
conn.connDown(DSException.makeMessage(x));
455465
}
456466
}
467+
debug("Exit DSSession.WriteThread");
457468
}
458469
}
459470

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ public void handleUpdate(int sid, String ts, String sts, DSElement value) {
9595
@Override
9696
public void write(DSSession session, MessageWriter writer) {
9797
if (!pendingSubscribe.isEmpty()) {
98+
debug(debug() ? "Sending subscribe requests" : null);
9899
doBeginSubscribe(writer);
99100
Iterator<DSOutboundSubscribeStubs> it = pendingSubscribe.iterator();
100101
while (it.hasNext() && !session.shouldEndMessage()) {
@@ -114,6 +115,7 @@ public void write(DSSession session, MessageWriter writer) {
114115
doEndMessage(writer);
115116
}
116117
if (!pendingUnsubscribe.isEmpty() && !session.shouldEndMessage()) {
118+
debug(debug() ? "Sending unsubscribe requests" : null);
117119
doBeginUnsubscribe(writer);
118120
Iterator<DSOutboundSubscribeStubs> it = pendingUnsubscribe.iterator();
119121
while (it.hasNext() && !session.shouldEndMessage()) {
@@ -129,9 +131,7 @@ public void write(DSSession session, MessageWriter writer) {
129131
}
130132
doEndMessage(writer);
131133
}
132-
synchronized (this) {
133-
enqueued = false;
134-
}
134+
enqueued = false;
135135
if (!pendingSubscribe.isEmpty() || !pendingUnsubscribe.isEmpty()) {
136136
sendMessage();
137137
}
@@ -208,6 +208,7 @@ protected void onDisconnected() {
208208
}
209209
sidMap.clear();
210210
pathMap.clear();
211+
enqueued = false;
211212
}
212213

213214
///////////////////////////////////////////////////////////////////////////
@@ -239,6 +240,7 @@ private void sendMessage() {
239240
* Create or update a subscription.
240241
*/
241242
OutboundSubscribeHandler subscribe(String path, int qos, OutboundSubscribeHandler req) {
243+
trace(trace() ? String.format("Subscribe (qos=%s) %s", qos, path) : null);
242244
DSOutboundSubscribeStub stub = new DSOutboundSubscribeStub(path, qos, req);
243245
DSOutboundSubscribeStubs stubs = null;
244246
synchronized (pathMap) {

‎dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/v1/DS1Session.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -112,19 +112,19 @@ protected void doRecvMessage() throws IOException {
112112
@Override
113113
protected void doSendMessage() {
114114
try {
115+
beginMessage();
115116
if (!waitingForAcks()) {
116117
requestsNext = !requestsNext;
117-
beginMessage();
118118
send(requestsNext);
119119
if (!shouldEndMessage()) {
120120
send(!requestsNext);
121121
}
122-
endMessage();
123122
lastMessageSent = System.currentTimeMillis();
124123
if (requestsBegun || responsesBegun) {
125124
setAckRequired();
126125
}
127126
}
127+
endMessage();
128128
} finally {
129129
requestsBegun = false;
130130
responsesBegun = false;
@@ -250,11 +250,9 @@ private DSIWriter getWriter() {
250250
* Decomposes and processes a complete envelope which can contain multiple requests and
251251
* responses.
252252
*
253-
* @param reader lastRun() will return BEGIN_MAP
253+
* @param reader last() must return BEGIN_MAP
254254
*/
255255
private void processEnvelope(DSIReader reader) {
256-
int msg = -1;
257-
Token next;
258256
switch (reader.next()) {
259257
case END_MAP:
260258
return;
@@ -263,6 +261,8 @@ private void processEnvelope(DSIReader reader) {
263261
default:
264262
throw new IllegalStateException("Poorly formatted request");
265263
}
264+
int msg = -1;
265+
Token next;
266266
boolean sendAck = false;
267267
do {
268268
String key = reader.getString();

0 commit comments

Comments
 (0)