简柏林 / thingskit · Commits

Commit 26d4d0de6c461bf95b7bd3f4f0ca4cd3e38dcb8c

Authored by Andrii Shvaika
1 parent f4731905

Merge with master

Inline Side-by-side
Showing 9 changed files with 282 additions and 170 deletions
@@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.TransportPayloadType; @@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.TransportPayloadType;
31 import org.thingsboard.server.common.msg.session.FeatureType; 31 import org.thingsboard.server.common.msg.session.FeatureType;
32 import org.thingsboard.server.transport.coap.AbstractCoapIntegrationTest; 32 import org.thingsboard.server.transport.coap.AbstractCoapIntegrationTest;
33 33
  34 +import java.nio.charset.StandardCharsets;
34 import java.util.concurrent.CountDownLatch; 35 import java.util.concurrent.CountDownLatch;
35 import java.util.concurrent.TimeUnit; 36 import java.util.concurrent.TimeUnit;
36 37
@@ -82,143 +82,8 @@ public abstract class AbstractCoapTransportResource extends CoapResource { @@ -82,143 +82,8 @@ public abstract class AbstractCoapTransportResource extends CoapResource {
82 .setEvent(event).build(); 82 .setEvent(event).build();
83 } 83 }
84 84
85 - @SneakyThrows  
86 - protected int respond(Response response, CoapExchange exchange, TransportProtos.SessionInfoProto sessionInfo) {  
87 - int msgId = ThreadLocalRandom.current().nextInt(NONE, MAX_MID + 1);  
88 - response.setMID(msgId);  
89 - response.addMessageObserver(new MessageObserver() {  
90 - @Override  
91 - public void onRetransmission() {  
92 - }  
93 -  
94 - @Override  
95 - public void onResponse(Response response) {  
96 - }  
97 -  
98 - @Override  
99 - public void onAcknowledgement() {  
100 - TransportProtos.ToDeviceRpcRequestMsg msg = transportContext.getRpcAwaitingAck().remove(msgId);  
101 - if (msg != null) {  
102 - transportService.process(sessionInfo, msg, false, TransportServiceCallback.EMPTY);  
103 - }  
104 - }  
105 -  
106 - @Override  
107 - public void onReject() {  
108 - }  
109 -  
110 - @Override  
111 - public void onTimeout() {  
112 - }  
113 -  
114 - @Override  
115 - public void onCancel() {  
116 - }  
117 -  
118 - @Override  
119 - public void onReadyToSend() {  
120 - }  
121 -  
122 - @Override  
123 - public void onConnecting() {  
124 - }  
125 -  
126 - @Override  
127 - public void onDtlsRetransmission(int flight) {  
128 - }  
129 -  
130 - @Override  
131 - public void onSent(boolean retransmission) {  
132 - }  
133 -  
134 - @Override  
135 - public void onSendError(Throwable error) {  
136 - }  
137 -  
138 - @Override  
139 - public void onContextEstablished(EndpointContext endpointContext) {  
140 - }  
141 -  
142 - @Override  
143 - public void onComplete() {  
144 - }  
145 - });  
146 -  
147 - exchange.respond(response);  
148 - return msgId;  
149 - }  
150 -  
151 - public static class CoapDeviceAuthCallback implements TransportServiceCallback<ValidateDeviceCredentialsResponse> {  
152 - private final TransportContext transportContext;  
153 - private final CoapExchange exchange;  
154 - private final BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess;  
155 -  
156 - public CoapDeviceAuthCallback(TransportContext transportContext, CoapExchange exchange, BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess) {  
157 - this.transportContext = transportContext;  
158 - this.exchange = exchange;  
159 - this.onSuccess = onSuccess;  
160 - }  
161 -  
162 - @Override  
163 - public void onSuccess(ValidateDeviceCredentialsResponse msg) {  
164 - DeviceProfile deviceProfile = msg.getDeviceProfile();  
165 - if (msg.hasDeviceInfo() && deviceProfile != null) {  
166 - TransportProtos.SessionInfoProto sessionInfoProto = SessionInfoCreator.create(msg, transportContext, UUID.randomUUID());  
167 - onSuccess.accept(sessionInfoProto, deviceProfile);  
168 - } else {  
169 - exchange.respond(CoAP.ResponseCode.UNAUTHORIZED);  
170 - }  
171 - }  
172 -  
173 - @Override  
174 - public void onError(Throwable e) {  
175 - log.warn("Failed to process request", e);  
176 - exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);  
177 - }  
178 - }  
179 -  
180 - public static class CoapOkCallback implements TransportServiceCallback<Void> {  
181 - private final CoapExchange exchange;  
182 - private final CoAP.ResponseCode onSuccessResponse;  
183 - private final CoAP.ResponseCode onFailureResponse;  
184 -  
185 - public CoapOkCallback(CoapExchange exchange, CoAP.ResponseCode onSuccessResponse, CoAP.ResponseCode onFailureResponse) {  
186 - this.exchange = exchange;  
187 - this.onSuccessResponse = onSuccessResponse;  
188 - this.onFailureResponse = onFailureResponse;  
189 - }  
190 -  
191 - @Override  
192 - public void onSuccess(Void msg) {  
193 - Response response = new Response(onSuccessResponse);  
194 - response.setAcknowledged(isConRequest());  
195 - exchange.respond(response);  
196 - }  
197 -  
198 - @Override  
199 - public void onError(Throwable e) {  
200 - exchange.respond(onFailureResponse);  
201 - }  
202 -  
203 - private boolean isConRequest() {  
204 - return exchange.advanced().getRequest().isConfirmable();  
205 - } 85 + protected int getNextMsgId() {
  86 + return ThreadLocalRandom.current().nextInt(NONE, MAX_MID + 1);
206 } 87 }
207 88
208 - public static class CoapNoOpCallback implements TransportServiceCallback<Void> {  
209 - private final CoapExchange exchange;  
210 -  
211 - CoapNoOpCallback(CoapExchange exchange) {  
212 - this.exchange = exchange;  
213 - }  
214 -  
215 - @Override  
216 - public void onSuccess(Void msg) {  
217 - }  
218 -  
219 - @Override  
220 - public void onError(Throwable e) {  
221 - exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);  
222 - }  
223 - }  
224 } 89 }
@@ -19,6 +19,7 @@ import com.google.gson.JsonParseException; @@ -19,6 +19,7 @@ import com.google.gson.JsonParseException;
19 import com.google.protobuf.Descriptors; 19 import com.google.protobuf.Descriptors;
20 import com.google.protobuf.DynamicMessage; 20 import com.google.protobuf.DynamicMessage;
21 import lombok.Data; 21 import lombok.Data;
  22 +import lombok.RequiredArgsConstructor;
22 import lombok.extern.slf4j.Slf4j; 23 import lombok.extern.slf4j.Slf4j;
23 import org.eclipse.californium.core.coap.CoAP; 24 import org.eclipse.californium.core.coap.CoAP;
24 import org.eclipse.californium.core.coap.Request; 25 import org.eclipse.californium.core.coap.Request;
@@ -53,6 +54,9 @@ import org.thingsboard.server.common.transport.adaptor.AdaptorException; @@ -53,6 +54,9 @@ import org.thingsboard.server.common.transport.adaptor.AdaptorException;
53 import org.thingsboard.server.common.transport.adaptor.JsonConverter; 54 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
54 import org.thingsboard.server.gen.transport.TransportProtos; 55 import org.thingsboard.server.gen.transport.TransportProtos;
55 import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; 56 import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor;
  57 +import org.thingsboard.server.transport.coap.callback.CoapDeviceAuthCallback;
  58 +import org.thingsboard.server.transport.coap.callback.CoapNoOpCallback;
  59 +import org.thingsboard.server.transport.coap.callback.CoapOkCallback;
56 60
57 import java.util.List; 61 import java.util.List;
58 import java.util.Map; 62 import java.util.Map;
@@ -81,9 +85,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -81,9 +85,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
81 private final Set<UUID> rpcSubscriptions = ConcurrentHashMap.newKeySet(); 85 private final Set<UUID> rpcSubscriptions = ConcurrentHashMap.newKeySet();
82 private final Set<UUID> attributeSubscriptions = ConcurrentHashMap.newKeySet(); 86 private final Set<UUID> attributeSubscriptions = ConcurrentHashMap.newKeySet();
83 87
84 - private ConcurrentMap<String, TbCoapDtlsSessionInfo> dtlsSessionIdMap;  
85 - private long timeout;  
86 - private long sessionReportTimeout; 88 + private final ConcurrentMap<String, TbCoapDtlsSessionInfo> dtlsSessionIdMap;
  89 + private final long timeout;
87 90
88 public CoapTransportResource(CoapTransportContext ctx, CoapServerService coapServerService, String name) { 91 public CoapTransportResource(CoapTransportContext ctx, CoapServerService coapServerService, String name) {
89 super(ctx, name); 92 super(ctx, name);
@@ -91,7 +94,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -91,7 +94,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
91 this.addObserver(new CoapResourceObserver()); 94 this.addObserver(new CoapResourceObserver());
92 this.dtlsSessionIdMap = coapServerService.getDtlsSessionsMap(); 95 this.dtlsSessionIdMap = coapServerService.getDtlsSessionsMap();
93 this.timeout = coapServerService.getTimeout(); 96 this.timeout = coapServerService.getTimeout();
94 - this.sessionReportTimeout = ctx.getSessionReportTimeout(); 97 + long sessionReportTimeout = ctx.getSessionReportTimeout();
95 ctx.getScheduler().scheduleAtFixedRate(() -> { 98 ctx.getScheduler().scheduleAtFixedRate(() -> {
96 Set<CoapObserveSessionInfo> coapObserveSessionInfos = sessionInfoToObserveRelationMap.keySet(); 99 Set<CoapObserveSessionInfo> coapObserveSessionInfos = sessionInfoToObserveRelationMap.keySet();
97 Set<TransportProtos.SessionInfoProto> observeSessions = coapObserveSessionInfos 100 Set<TransportProtos.SessionInfoProto> observeSessions = coapObserveSessionInfos
@@ -110,7 +113,6 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -110,7 +113,6 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
110 return; // because request did not try to establish a relation 113 return; // because request did not try to establish a relation
111 } 114 }
112 if (CoAP.ResponseCode.isSuccess(response.getCode())) { 115 if (CoAP.ResponseCode.isSuccess(response.getCode())) {
113 -  
114 if (!relation.isEstablished()) { 116 if (!relation.isEstablished()) {
115 relation.setEstablished(); 117 relation.setEstablished();
116 addObserveRelation(relation); 118 addObserveRelation(relation);
@@ -280,8 +282,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -280,8 +282,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
280 CoapObserveSessionInfo currentCoapObserveAttrSessionInfo = tokenToCoapSessionInfoMap.get(getTokenFromRequest(request)); 282 CoapObserveSessionInfo currentCoapObserveAttrSessionInfo = tokenToCoapSessionInfoMap.get(getTokenFromRequest(request));
281 if (currentCoapObserveAttrSessionInfo == null) { 283 if (currentCoapObserveAttrSessionInfo == null) {
282 attributeSubscriptions.add(sessionId); 284 attributeSubscriptions.add(sessionId);
283 - registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor,  
284 - transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(), getTokenFromRequest(request)); 285 + registerAsyncCoapSession(exchange, coapTransportAdaptor, transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(),
  286 + sessionInfo, getTokenFromRequest(request));
285 transportService.process(sessionInfo, 287 transportService.process(sessionInfo,
286 TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), new CoapNoOpCallback(exchange)); 288 TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), new CoapNoOpCallback(exchange));
287 transportService.process(sessionInfo, 289 transportService.process(sessionInfo,
@@ -305,11 +307,11 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -305,11 +307,11 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
305 CoapObserveSessionInfo currentCoapObserveRpcSessionInfo = tokenToCoapSessionInfoMap.get(getTokenFromRequest(request)); 307 CoapObserveSessionInfo currentCoapObserveRpcSessionInfo = tokenToCoapSessionInfoMap.get(getTokenFromRequest(request));
306 if (currentCoapObserveRpcSessionInfo == null) { 308 if (currentCoapObserveRpcSessionInfo == null) {
307 rpcSubscriptions.add(sessionId); 309 rpcSubscriptions.add(sessionId);
308 - registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor,  
309 - transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(), getTokenFromRequest(request)); 310 + registerAsyncCoapSession(exchange, coapTransportAdaptor, transportConfigurationContainer.getRpcRequestDynamicMessageBuilder()
  311 + , sessionInfo, getTokenFromRequest(request));
310 transportService.process(sessionInfo, 312 transportService.process(sessionInfo,
311 TransportProtos.SubscribeToRPCMsg.getDefaultInstance(), 313 TransportProtos.SubscribeToRPCMsg.getDefaultInstance(),
312 - new CoapNoOpCallback(exchange) 314 + new CoapOkCallback(exchange, CoAP.ResponseCode.VALID, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)
313 ); 315 );
314 } 316 }
315 break; 317 break;
@@ -359,14 +361,16 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -359,14 +361,16 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
359 return tokenToCoapSessionInfoMap.remove(token); 361 return tokenToCoapSessionInfoMap.remove(token);
360 } 362 }
361 363
362 - private void registerAsyncCoapSession(CoapExchange exchange, TransportProtos.SessionInfoProto sessionInfo, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, String token) { 364 + private void registerAsyncCoapSession(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor,
  365 + DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo, String token) {
363 tokenToCoapSessionInfoMap.putIfAbsent(token, new CoapObserveSessionInfo(sessionInfo)); 366 tokenToCoapSessionInfoMap.putIfAbsent(token, new CoapObserveSessionInfo(sessionInfo));
364 transportService.registerAsyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo)); 367 transportService.registerAsyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo));
365 transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null); 368 transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null);
366 } 369 }
367 370
368 - private CoapSessionListener getCoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo) {  
369 - return new CoapSessionListener(this, exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo); 371 + private CoapSessionListener getCoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor,
  372 + DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo) {
  373 + return new CoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo);
370 } 374 }
371 375
372 private String getTokenFromRequest(Request request) { 376 private String getTokenFromRequest(Request request) {
@@ -448,22 +452,14 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -448,22 +452,14 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
448 } 452 }
449 } 453 }
450 454
  455 + @RequiredArgsConstructor
451 private class CoapSessionListener implements SessionMsgListener { 456 private class CoapSessionListener implements SessionMsgListener {
452 457
453 - private final CoapTransportResource coapTransportResource;  
454 private final CoapExchange exchange; 458 private final CoapExchange exchange;
455 private final CoapTransportAdaptor coapTransportAdaptor; 459 private final CoapTransportAdaptor coapTransportAdaptor;
456 private final DynamicMessage.Builder rpcRequestDynamicMessageBuilder; 460 private final DynamicMessage.Builder rpcRequestDynamicMessageBuilder;
457 private final TransportProtos.SessionInfoProto sessionInfo; 461 private final TransportProtos.SessionInfoProto sessionInfo;
458 462
459 - CoapSessionListener(CoapTransportResource coapTransportResource, CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo) {  
460 - this.coapTransportResource = coapTransportResource;  
461 - this.exchange = exchange;  
462 - this.coapTransportAdaptor = coapTransportAdaptor;  
463 - this.rpcRequestDynamicMessageBuilder = rpcRequestDynamicMessageBuilder;  
464 - this.sessionInfo = sessionInfo;  
465 - }  
466 -  
467 @Override 463 @Override
468 public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) { 464 public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) {
469 try { 465 try {
@@ -497,7 +493,9 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -497,7 +493,9 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
497 public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg msg) { 493 public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg msg) {
498 log.trace("[{}] Received RPC command to device", sessionId); 494 log.trace("[{}] Received RPC command to device", sessionId);
499 try { 495 try {
500 - int requestId = respond(coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder), exchange, sessionInfo); 496 + Response response = coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder);
  497 + int requestId = getNextMsgId();
  498 + response.setMID(requestId);
501 if (msg.getPersisted()) { 499 if (msg.getPersisted()) {
502 transportContext.getRpcAwaitingAck().put(requestId, msg); 500 transportContext.getRpcAwaitingAck().put(requestId, msg);
503 transportContext.getScheduler().schedule(() -> { 501 transportContext.getScheduler().schedule(() -> {
@@ -507,6 +505,13 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -507,6 +505,13 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
507 } 505 }
508 }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS); 506 }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
509 } 507 }
  508 + response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> {
  509 + TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
  510 + if (rpcRequestMsg != null) {
  511 + transportService.process(sessionInfo, rpcRequestMsg, false, TransportServiceCallback.EMPTY);
  512 + }
  513 + }));
  514 + exchange.respond(response);
510 } catch (AdaptorException e) { 515 } catch (AdaptorException e) {
511 log.trace("Failed to reply due to error", e); 516 log.trace("Failed to reply due to error", e);
512 closeObserveRelationAndNotify(sessionId, CoAP.ResponseCode.INTERNAL_SERVER_ERROR); 517 closeObserveRelationAndNotify(sessionId, CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
@@ -529,8 +534,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -529,8 +534,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
529 } 534 }
530 535
531 private void closeObserveRelationAndNotify(UUID sessionId, CoAP.ResponseCode responseCode) { 536 private void closeObserveRelationAndNotify(UUID sessionId, CoAP.ResponseCode responseCode) {
532 - Map<CoapObserveSessionInfo, ObserveRelation> sessionToObserveRelationMap = coapTransportResource.getCoapSessionInfoToObserveRelationMap();  
533 - if (coapTransportResource.getObserverCount() > 0 && !CollectionUtils.isEmpty(sessionToObserveRelationMap)) { 537 + Map<CoapObserveSessionInfo, ObserveRelation> sessionToObserveRelationMap = CoapTransportResource.this.getCoapSessionInfoToObserveRelationMap();
  538 + if (CoapTransportResource.this.getObserverCount() > 0 && !CollectionUtils.isEmpty(sessionToObserveRelationMap)) {
534 Optional<CoapObserveSessionInfo> observeSessionToClose = sessionToObserveRelationMap.keySet().stream().filter(coapObserveSessionInfo -> { 539 Optional<CoapObserveSessionInfo> observeSessionToClose = sessionToObserveRelationMap.keySet().stream().filter(coapObserveSessionInfo -> {
535 TransportProtos.SessionInfoProto sessionToDelete = coapObserveSessionInfo.getSessionInfoProto(); 540 TransportProtos.SessionInfoProto sessionToDelete = coapObserveSessionInfo.getSessionInfoProto();
536 UUID observeSessionId = new UUID(sessionToDelete.getSessionIdMSB(), sessionToDelete.getSessionIdLSB()); 541 UUID observeSessionId = new UUID(sessionToDelete.getSessionIdMSB(), sessionToDelete.getSessionIdLSB());
@@ -539,16 +544,16 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @@ -539,16 +544,16 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
539 if (observeSessionToClose.isPresent()) { 544 if (observeSessionToClose.isPresent()) {
540 CoapObserveSessionInfo coapObserveSessionInfo = observeSessionToClose.get(); 545 CoapObserveSessionInfo coapObserveSessionInfo = observeSessionToClose.get();
541 ObserveRelation observeRelation = sessionToObserveRelationMap.get(coapObserveSessionInfo); 546 ObserveRelation observeRelation = sessionToObserveRelationMap.get(coapObserveSessionInfo);
542 - coapTransportResource.clearAndNotifyObserveRelation(observeRelation, responseCode); 547 + CoapTransportResource.this.clearAndNotifyObserveRelation(observeRelation, responseCode);
543 } 548 }
544 } 549 }
545 } 550 }
546 551
547 private void closeAndDeregister() { 552 private void closeAndDeregister() {
548 Request request = exchange.advanced().getRequest(); 553 Request request = exchange.advanced().getRequest();
549 - String token = coapTransportResource.getTokenFromRequest(request);  
550 - CoapObserveSessionInfo deleted = coapTransportResource.lookupAsyncSessionInfo(token);  
551 - coapTransportResource.closeAndDeregister(deleted.getSessionInfoProto()); 554 + String token = CoapTransportResource.this.getTokenFromRequest(request);
  555 + CoapObserveSessionInfo deleted = CoapTransportResource.this.lookupAsyncSessionInfo(token);
  556 + CoapTransportResource.this.closeAndDeregister(deleted.getSessionInfoProto());
552 } 557 }
553 558
554 } 559 }
@@ -20,22 +20,19 @@ import org.eclipse.californium.core.coap.CoAP; @@ -20,22 +20,19 @@ import org.eclipse.californium.core.coap.CoAP;
20 import org.eclipse.californium.core.coap.Request; 20 import org.eclipse.californium.core.coap.Request;
21 import org.eclipse.californium.core.coap.Response; 21 import org.eclipse.californium.core.coap.Response;
22 import org.eclipse.californium.core.network.Exchange; 22 import org.eclipse.californium.core.network.Exchange;
23 -import org.eclipse.californium.core.observe.ObserveRelation;  
24 import org.eclipse.californium.core.server.resources.CoapExchange; 23 import org.eclipse.californium.core.server.resources.CoapExchange;
25 import org.eclipse.californium.core.server.resources.Resource; 24 import org.eclipse.californium.core.server.resources.Resource;
26 -import org.eclipse.californium.core.server.resources.ResourceObserver;  
27 -import org.thingsboard.common.util.ThingsBoardExecutors;  
28 import org.thingsboard.server.common.data.DeviceTransportType; 25 import org.thingsboard.server.common.data.DeviceTransportType;
29 import org.thingsboard.server.common.data.StringUtils; 26 import org.thingsboard.server.common.data.StringUtils;
30 import org.thingsboard.server.common.data.ota.OtaPackageType; 27 import org.thingsboard.server.common.data.ota.OtaPackageType;
31 import org.thingsboard.server.common.data.security.DeviceTokenCredentials; 28 import org.thingsboard.server.common.data.security.DeviceTokenCredentials;
32 import org.thingsboard.server.common.transport.TransportServiceCallback; 29 import org.thingsboard.server.common.transport.TransportServiceCallback;
33 import org.thingsboard.server.gen.transport.TransportProtos; 30 import org.thingsboard.server.gen.transport.TransportProtos;
  31 +import org.thingsboard.server.transport.coap.callback.CoapDeviceAuthCallback;
34 32
35 import java.util.List; 33 import java.util.List;
36 import java.util.Optional; 34 import java.util.Optional;
37 import java.util.UUID; 35 import java.util.UUID;
38 -import java.util.concurrent.ExecutorService;  
39 36
40 @Slf4j 37 @Slf4j
41 public class OtaPackageTransportResource extends AbstractCoapTransportResource { 38 public class OtaPackageTransportResource extends AbstractCoapTransportResource {
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.transport.coap;
  17 +
  18 +import lombok.RequiredArgsConstructor;
  19 +import org.eclipse.californium.core.coap.MessageObserver;
  20 +import org.eclipse.californium.core.coap.Response;
  21 +import org.eclipse.californium.elements.EndpointContext;
  22 +
  23 +import java.util.function.Consumer;
  24 +
  25 +@RequiredArgsConstructor
  26 +public class TbCoapMessageObserver implements MessageObserver {
  27 +
  28 + private final int msgId;
  29 + private final Consumer<Integer> onAcknowledge;
  30 +
  31 + @Override
  32 + public void onRetransmission() {
  33 +
  34 + }
  35 +
  36 + @Override
  37 + public void onResponse(Response response) {
  38 +
  39 + }
  40 +
  41 + @Override
  42 + public void onAcknowledgement() {
  43 + onAcknowledge.accept(msgId);
  44 + }
  45 +
  46 + @Override
  47 + public void onReject() {
  48 +
  49 + }
  50 +
  51 + @Override
  52 + public void onTimeout() {
  53 +
  54 + }
  55 +
  56 + @Override
  57 + public void onCancel() {
  58 +
  59 + }
  60 +
  61 + @Override
  62 + public void onReadyToSend() {
  63 +
  64 + }
  65 +
  66 + @Override
  67 + public void onConnecting() {
  68 +
  69 + }
  70 +
  71 + @Override
  72 + public void onDtlsRetransmission(int flight) {
  73 +
  74 + }
  75 +
  76 + @Override
  77 + public void onSent(boolean retransmission) {
  78 +
  79 + }
  80 +
  81 + @Override
  82 + public void onSendError(Throwable error) {
  83 +
  84 + }
  85 +
  86 + @Override
  87 + public void onContextEstablished(EndpointContext endpointContext) {
  88 +
  89 + }
  90 +
  91 + @Override
  92 + public void onComplete() {
  93 +
  94 + }
  95 +}
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.transport.coap.callback;
  17 +
  18 +import lombok.extern.slf4j.Slf4j;
  19 +import org.eclipse.californium.core.coap.CoAP;
  20 +import org.eclipse.californium.core.server.resources.CoapExchange;
  21 +import org.thingsboard.server.common.data.DeviceProfile;
  22 +import org.thingsboard.server.common.transport.TransportContext;
  23 +import org.thingsboard.server.common.transport.TransportServiceCallback;
  24 +import org.thingsboard.server.common.transport.auth.SessionInfoCreator;
  25 +import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
  26 +import org.thingsboard.server.gen.transport.TransportProtos;
  27 +import org.thingsboard.server.transport.coap.AbstractCoapTransportResource;
  28 +
  29 +import java.util.UUID;
  30 +import java.util.function.BiConsumer;
  31 +
  32 +@Slf4j
  33 +public class CoapDeviceAuthCallback implements TransportServiceCallback<ValidateDeviceCredentialsResponse> {
  34 + private final TransportContext transportContext;
  35 + private final CoapExchange exchange;
  36 + private final BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess;
  37 +
  38 + public CoapDeviceAuthCallback(TransportContext transportContext, CoapExchange exchange, BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess) {
  39 + this.transportContext = transportContext;
  40 + this.exchange = exchange;
  41 + this.onSuccess = onSuccess;
  42 + }
  43 +
  44 + @Override
  45 + public void onSuccess(ValidateDeviceCredentialsResponse msg) {
  46 + DeviceProfile deviceProfile = msg.getDeviceProfile();
  47 + if (msg.hasDeviceInfo() && deviceProfile != null) {
  48 + TransportProtos.SessionInfoProto sessionInfoProto = SessionInfoCreator.create(msg, transportContext, UUID.randomUUID());
  49 + onSuccess.accept(sessionInfoProto, deviceProfile);
  50 + } else {
  51 + exchange.respond(CoAP.ResponseCode.UNAUTHORIZED);
  52 + }
  53 + }
  54 +
  55 + @Override
  56 + public void onError(Throwable e) {
  57 + log.warn("Failed to process request", e);
  58 + exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
  59 + }
  60 +}
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.transport.coap.callback;
  17 +
  18 +import org.eclipse.californium.core.coap.CoAP;
  19 +import org.eclipse.californium.core.server.resources.CoapExchange;
  20 +import org.thingsboard.server.common.transport.TransportServiceCallback;
  21 +
  22 +public class CoapNoOpCallback implements TransportServiceCallback<Void> {
  23 + private final CoapExchange exchange;
  24 +
  25 + public CoapNoOpCallback(CoapExchange exchange) {
  26 + this.exchange = exchange;
  27 + }
  28 +
  29 + @Override
  30 + public void onSuccess(Void msg) {
  31 + }
  32 +
  33 + @Override
  34 + public void onError(Throwable e) {
  35 + exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
  36 + }
  37 +}
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.transport.coap.callback;
  17 +
  18 +import org.eclipse.californium.core.coap.CoAP;
  19 +import org.eclipse.californium.core.coap.Response;
  20 +import org.eclipse.californium.core.server.resources.CoapExchange;
  21 +import org.thingsboard.server.common.transport.TransportServiceCallback;
  22 +
  23 +public class CoapOkCallback implements TransportServiceCallback<Void> {
  24 +
  25 + protected final CoapExchange exchange;
  26 + protected final CoAP.ResponseCode onSuccessResponse;
  27 + protected final CoAP.ResponseCode onFailureResponse;
  28 +
  29 + public CoapOkCallback(CoapExchange exchange, CoAP.ResponseCode onSuccessResponse, CoAP.ResponseCode onFailureResponse) {
  30 + this.exchange = exchange;
  31 + this.onSuccessResponse = onSuccessResponse;
  32 + this.onFailureResponse = onFailureResponse;
  33 + }
  34 +
  35 + @Override
  36 + public void onSuccess(Void msg) {
  37 + Response response = new Response(onSuccessResponse);
  38 + response.setConfirmable(isConRequest());
  39 + exchange.respond(response);
  40 + }
  41 +
  42 + @Override
  43 + public void onError(Throwable e) {
  44 + exchange.respond(onFailureResponse);
  45 + }
  46 +
  47 + protected boolean isConRequest() {
  48 + return exchange.advanced().getRequest().isConfirmable();
  49 + }
  50 +}
@@ -35,6 +35,8 @@ import org.thingsboard.server.gen.transport.TransportProtos; @@ -35,6 +35,8 @@ import org.thingsboard.server.gen.transport.TransportProtos;
35 import org.thingsboard.server.gen.transport.coap.MeasurementTypeProtos; 35 import org.thingsboard.server.gen.transport.coap.MeasurementTypeProtos;
36 import org.thingsboard.server.gen.transport.coap.MeasurementsProtos; 36 import org.thingsboard.server.gen.transport.coap.MeasurementsProtos;
37 import org.thingsboard.server.transport.coap.AbstractCoapTransportResource; 37 import org.thingsboard.server.transport.coap.AbstractCoapTransportResource;
  38 +import org.thingsboard.server.transport.coap.callback.CoapDeviceAuthCallback;
  39 +import org.thingsboard.server.transport.coap.callback.CoapOkCallback;
38 import org.thingsboard.server.transport.coap.CoapTransportContext; 40 import org.thingsboard.server.transport.coap.CoapTransportContext;
39 import org.thingsboard.server.transport.coap.efento.utils.CoapEfentoUtils; 41 import org.thingsboard.server.transport.coap.efento.utils.CoapEfentoUtils;
40 42