tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Fix for HTTP and CoAP device activity

pull/2723/head
Andrii Shvaika 6 years ago
parent
2e148f6b7d
commit
830d02ef6a
3 changed files with 46 additions and 3 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 23
      common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
  2. 13
      common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java
  3. 13
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportServiceCallback.java

23
common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
View File

@ -37,6 +37,7 @@ import org.thingsboard.server.gen.transport.TransportProtos;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -55,6 +56,8 @@ public class CoapTransportResource extends CoapResource {
private final Field observerField;
private final long timeout;
private final ConcurrentMap<String, TransportProtos.SessionInfoProto> tokenToSessionIdMap = new ConcurrentHashMap<>();
private final Set<UUID> rpcSubscriptions = ConcurrentHashMap.newKeySet();
private final Set<UUID> attributeSubscriptions = ConcurrentHashMap.newKeySet();
public CoapTransportResource(CoapTransportContext context, String name) {
super(name);
@ -149,11 +152,13 @@ public class CoapTransportResource extends CoapResource {
transportService.process(sessionInfo,
transportContext.getAdaptor().convertToPostAttributes(sessionId, request),
new CoapOkCallback(exchange));
reportActivity(sessionId, sessionInfo);
break;
case POST_TELEMETRY_REQUEST:
transportService.process(sessionInfo,
transportContext.getAdaptor().convertToPostTelemetry(sessionId, request),
new CoapOkCallback(exchange));
reportActivity(sessionId, sessionInfo);
break;
case CLAIM_REQUEST:
transportService.process(sessionInfo,
@ -161,6 +166,7 @@ public class CoapTransportResource extends CoapResource {
new CoapOkCallback(exchange));
break;
case SUBSCRIBE_ATTRIBUTES_REQUEST:
attributeSubscriptions.add(sessionId);
advanced.setObserver(new CoapExchangeObserverProxy((ExchangeObserver) observerField.get(advanced),
registerAsyncCoapSession(exchange, request, sessionInfo, sessionId)));
transportService.process(sessionInfo,
@ -168,6 +174,7 @@ public class CoapTransportResource extends CoapResource {
new CoapNoOpCallback(exchange));
break;
case UNSUBSCRIBE_ATTRIBUTES_REQUEST:
attributeSubscriptions.remove(sessionId);
TransportProtos.SessionInfoProto attrSession = lookupAsyncSessionInfo(request);
if (attrSession != null) {
transportService.process(attrSession,
@ -177,6 +184,7 @@ public class CoapTransportResource extends CoapResource {
}
break;
case SUBSCRIBE_RPC_COMMANDS_REQUEST:
rpcSubscriptions.add(sessionId);
advanced.setObserver(new CoapExchangeObserverProxy((ExchangeObserver) observerField.get(advanced),
registerAsyncCoapSession(exchange, request, sessionInfo, sessionId)));
transportService.process(sessionInfo,
@ -184,13 +192,13 @@ public class CoapTransportResource extends CoapResource {
new CoapNoOpCallback(exchange));
break;
case UNSUBSCRIBE_RPC_COMMANDS_REQUEST:
rpcSubscriptions.remove(sessionId);
TransportProtos.SessionInfoProto rpcSession = lookupAsyncSessionInfo(request);
if (rpcSession != null) {
transportService.process(rpcSession,
TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(),
new CoapOkCallback(exchange));
transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null);
transportService.deregisterSession(rpcSession);
closeAndDeregister(sessionInfo);
}
break;
case TO_DEVICE_RPC_RESPONSE:
@ -221,6 +229,14 @@ public class CoapTransportResource extends CoapResource {
}));
}
private void reportActivity(UUID sessionId, TransportProtos.SessionInfoProto sessionInfo) {
transportContext.getTransportService().process(sessionInfo, TransportProtos.SubscriptionInfoProto.newBuilder()
.setAttributeSubscription(attributeSubscriptions.contains(sessionId))
.setRpcSubscription(rpcSubscriptions.contains(sessionId))
.setLastActivityTime(System.currentTimeMillis())
.build(), TransportServiceCallback.EMPTY);
}
private TransportProtos.SessionInfoProto lookupAsyncSessionInfo(Request request) {
String token = request.getSource().getHostAddress() + ":" + request.getSourcePort() + ":" + request.getTokenString();
return tokenToSessionIdMap.remove(token);
@ -438,6 +454,9 @@ public class CoapTransportResource extends CoapResource {
private void closeAndDeregister(TransportProtos.SessionInfoProto session) {
transportService.process(session, getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null);
transportService.deregisterSession(session);
UUID sessionId = new UUID(session.getSessionIdMSB(), session.getSessionIdLSB());
rpcSubscriptions.remove(sessionId);
attributeSubscriptions.remove(sessionId);
}
}

13
common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java
View File

@ -36,6 +36,7 @@ import org.thingsboard.server.common.transport.TransportContext;
import org.thingsboard.server.common.transport.TransportService;
import org.thingsboard.server.common.transport.TransportServiceCallback;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto;
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
@ -102,6 +103,7 @@ public class DeviceApiController {
TransportService transportService = transportContext.getTransportService();
transportService.process(sessionInfo, JsonConverter.convertToAttributesProto(new JsonParser().parse(json)),
new HttpOkCallback(responseWriter));
reportActivity(sessionInfo);
}));
return responseWriter;
}
@ -115,6 +117,7 @@ public class DeviceApiController {
TransportService transportService = transportContext.getTransportService();
transportService.process(sessionInfo, JsonConverter.convertToTelemetryProto(new JsonParser().parse(json)),
new HttpOkCallback(responseWriter));
reportActivity(sessionInfo);
}));
return responseWriter;
}
@ -274,7 +277,6 @@ public class DeviceApiController {
}
}
private static class HttpSessionListener implements SessionMsgListener {
private final DeferredResult<ResponseEntity> responseWriter;
@ -308,4 +310,13 @@ public class DeviceApiController {
responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg).toString(), HttpStatus.OK));
}
}
private void reportActivity(SessionInfoProto sessionInfo) {
transportContext.getTransportService().process(sessionInfo, TransportProtos.SubscriptionInfoProto.newBuilder()
.setAttributeSubscription(false)
.setRpcSubscription(false)
.setLastActivityTime(System.currentTimeMillis())
.build(), TransportServiceCallback.EMPTY);
}
}

13
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportServiceCallback.java
View File

@ -20,7 +20,20 @@ package org.thingsboard.server.common.transport;
*/
public interface TransportServiceCallback<T> {
TransportServiceCallback<Void> EMPTY = new TransportServiceCallback<Void>() {
@Override
public void onSuccess(Void msg) {
}
@Override
public void onError(Throwable e) {
}
};
void onSuccess(T msg);
void onError(Throwable e);
}

Write Preview
Loading…
Cancel
Save