@ -39,6 +39,8 @@ import org.eclipse.leshan.core.request.SimpleDownlinkRequest;
import org.eclipse.leshan.core.request.WriteAttributesRequest ;
import org.eclipse.leshan.core.request.WriteCompositeRequest ;
import org.eclipse.leshan.core.request.WriteRequest ;
import org.eclipse.leshan.core.request.exception.ClientSleepingException ;
import org.eclipse.leshan.core.request.exception.TimeoutException ;
import org.eclipse.leshan.core.response.DeleteResponse ;
import org.eclipse.leshan.core.response.DiscoverResponse ;
import org.eclipse.leshan.core.response.ExecuteResponse ;
@ -59,6 +61,7 @@ import org.thingsboard.server.queue.util.TbLwM2mTransportComponent;
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig ;
import org.thingsboard.server.transport.lwm2m.server.LwM2mTransportContext ;
import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient ;
import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClientContext ;
import org.thingsboard.server.transport.lwm2m.server.common.LwM2MExecutorAwareService ;
import org.thingsboard.server.transport.lwm2m.server.downlink.composite.TbLwM2MReadCompositeRequest ;
import org.thingsboard.server.transport.lwm2m.server.log.LwM2MTelemetryLogService ;
@ -95,6 +98,7 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
private final LwM2mTransportContext context ;
private final LwM2MTransportServerConfig config ;
private final LwM2MTelemetryLogService logService ;
private final LwM2mClientContext clientContext ;
@PostConstruct
public void init ( ) {
@ -236,10 +240,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
path . getObjectId ( ) , path . getObjectInstanceId ( ) , path . getResourceId ( ) , request . getValue ( ) ) ;
sendSimpleRequest ( client , downlink , request . getTimeout ( ) , callback ) ;
} catch ( Exception e ) {
callback . onError ( JacksonUtil . toString ( request ) , e ) ;
callback . onError ( toString ( request ) , e ) ;
}
} else {
callback . onValidationError ( JacksonUtil . toString ( request ) , "Resource " + request . getVersionedId ( ) + " is not configured in the device profile!" ) ;
callback . onValidationError ( toString ( request ) , "Resource " + request . getVersionedId ( ) + " is not configured in the device profile!" ) ;
}
}
@ -251,7 +255,7 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
//TODO: replace config.getTimeout();
sendWriteCompositeRequest ( client , downlink , config . getTimeout ( ) , callback ) ;
} catch ( Exception e ) {
callback . onError ( JacksonUtil . toString ( rpcWriteCompositeRequest ) , e ) ;
callback . onError ( toString ( rpcWriteCompositeRequest ) , e ) ;
}
}
@ -281,10 +285,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
WriteRequest downlink = new WriteRequest ( WriteRequest . Mode . UPDATE , contentFormat , resultIds . getObjectId ( ) , resultIds . getObjectInstanceId ( ) , resources ) ;
sendSimpleRequest ( client , downlink , request . getTimeout ( ) , callback ) ;
} else {
callback . onValidationError ( JacksonUtil . toString ( request ) , "No resources to update!" ) ;
callback . onValidationError ( toString ( request ) , "No resources to update!" ) ;
}
} else {
callback . onValidationError ( JacksonUtil . toString ( request ) , "Update of the root level object is not supported yet!" ) ;
callback . onValidationError ( toString ( request ) , "Update of the root level object is not supported yet!" ) ;
}
}
@ -298,6 +302,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
}
private < R extends DownlinkRequest < T > , T extends LwM2mResponse > void sendRequest ( LwM2mClient client , R request , long timeoutInMs , DownlinkRequestCallback < R , T > callback , Function < R , String > pathToStringFunction ) {
if ( ! clientContext . isDownlinkAllowed ( client ) ) {
log . trace ( "[{}] ignore downlink request cause client is sleeping." , client . getEndpoint ( ) ) ;
return ;
}
Registration registration = client . getRegistration ( ) ;
try {
logService . log ( client , String . format ( "[%s][%s] Sending request: %s to %s" , registration . getId ( ) , registration . getSocketAddress ( ) , request . getClass ( ) . getSimpleName ( ) , pathToStringFunction . apply ( request ) ) ) ;
@ -307,20 +315,21 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
callback . onSuccess ( request , response ) ;
} catch ( Exception e ) {
log . error ( "[{}] failed to process successful response [{}] " , registration . getEndpoint ( ) , response , e ) ;
} finally {
clientContext . awake ( client ) ;
}
} ) ;
} , e - > {
executor . submit ( ( ) - > {
callback . onError ( JacksonUtil . toString ( request ) , e ) ;
} ) ;
} ) ;
} , e - > handleDownlinkError ( client , request , callback , e ) ) ;
} catch ( Exception e ) {
callback . onError ( JacksonUtil . toString ( request ) , e ) ;
handleDownlinkError ( client , request , callback , e ) ;
}
}
private < R extends SimpleDownlinkRequest < T > , T extends LwM2mResponse > void sendWriteCompositeRequest ( LwM2mClient client , WriteCompositeRequest request , long timeoutInMs ,
DownlinkRequestCallback < WriteCompositeRequest , WriteCompositeResponse > callback ) {
private < R extends SimpleDownlinkRequest < T > , T extends LwM2mResponse > void sendWriteCompositeRequest ( LwM2mClient client , WriteCompositeRequest request , long timeoutInMs , DownlinkRequestCallback < WriteCompositeRequest , WriteCompositeResponse > callback ) {
if ( ! clientContext . isDownlinkAllowed ( client ) ) {
log . trace ( "[{}] ignore downlink request cause client is sleeping." , client . getEndpoint ( ) ) ;
return ;
}
Registration registration = client . getRegistration ( ) ;
try {
logService . log ( client , String . format ( "[%s][%s] Sending request: %s to %s" , registration . getId ( ) , registration . getSocketAddress ( ) , request . getClass ( ) . getSimpleName ( ) , request . getPaths ( ) ) ) ;
@ -334,63 +343,28 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
}
} catch ( Exception e ) {
log . error ( "[{}] failed to process successful response [{}] " , registration . getEndpoint ( ) , response , e ) ;
} finally {
clientContext . awake ( client ) ;
}
} ) ;
} , e - > {
executor . submit ( ( ) - > {
callback . onError ( JacksonUtil . toString ( request ) , e ) ;
} ) ;
} ) ;
} , e - > handleDownlinkError ( client , request , callback , e ) ) ;
} catch ( Exception e ) {
callback . onError ( JacksonUtil . toString ( request ) , e ) ;
handleDownlinkError ( client , request , callback , e ) ;
}
}
// private <R extends DownlinkRequest<T>, T extends LwM2mResponse> void sendReadRequestComposite(LwM2mClient client, ReadCompositeRequest request, long timeoutInMs,
// DownlinkRequestCallback<ReadCompositeRequest, ReadCompositeResponse> callback) {
// Registration registration = client.getRegistration();
// try {
// logService.log(client, String.format("[%s][%s] Sending request: %s to %s", registration.getId(), registration.getSocketAddress(), request.getClass().getSimpleName(), request.getPaths()));
// context.getServer().send(registration, request, timeoutInMs, response -> {
// executor.submit(() -> {
// try {
// /**
// * [{"bn":"/3/0/","n":"0","vs":"Thingsboard Test Device"},
// * {"n":"1","vs":"Model 500"},
// * {"n":"2","vs":"TH-500-000-0001"},
// * {"n":"3","vs":"TestThingsboard@TestMore1024_2.04"},
// * {"n":"6","v":1},{"n":"7","v":56},
// * {"n":"8","v":42},{"n":"9","v":16},
// * {"n":"10","v":127619},{"n":"13","v":1624520988},
// * {"n":"14","vs":"+03"},{"n":"15","vs":"Europe/Kiev"},
// * {"n":"16","vs":"U"},{"n":"17","vs":"smart meters"},
// * {"n":"18","vs":"1.01"},{"n":"19","vs":"1.02"},
// * {"n":"20","v":3},{"n":"21","v":256000},
// * {"bn":"/5/0/","n":"1","vs":""},
// * {"n":"3","v":0},{"n":"5","v":0},
// * {"n":"6","vs":""},{"n":"7","vs":""},
// * {"n":"8/0","v":0},{"n":"8/1","v":1},
// * {"n":"9","v":2},
// * {"bn":"/1/0/","n":"0","v":123},
// * {"n":"1","v":300},
// * {"n":"6","vb":false},
// * {"n":"22","vs":"U"},
// * {"n":"7","vs":"U"}]
// */
// callback.onSuccess(request, response);
// } catch (Exception e) {
// log.error("[{}] failed to process successful response [{}] ", registration.getEndpoint(), response, e);
// }
// });
// }, e -> {
// executor.submit(() -> {
// callback.onError(JacksonUtil.toString(request), e);
// });
// });
// } catch (Exception e) {
// callback.onError(JacksonUtil.toString(request), e);
// }
// }
private < R extends DownlinkRequest < T > , T extends LwM2mResponse > void handleDownlinkError ( LwM2mClient client , R request , DownlinkRequestCallback < R , T > callback , Exception e ) {
log . trace ( "[{}] Received downlink error: {}." , client . getEndpoint ( ) , e ) ;
executor . submit ( ( ) - > {
if ( e instanceof TimeoutException | | e instanceof ClientSleepingException ) {
log . trace ( "[{}] Received {}, client is probably sleeping" , client . getEndpoint ( ) , e . getClass ( ) . getSimpleName ( ) ) ;
clientContext . asleep ( client ) ;
} else {
log . trace ( "[{}] Received {}" , client . getEndpoint ( ) , e . getClass ( ) . getSimpleName ( ) ) ;
}
callback . onError ( toString ( request ) , e ) ;
} ) ;
}
private WriteRequest getWriteRequestSingleResource ( ResourceModel . Type type , ContentFormat contentFormat , int objectId , int instanceId , int resourceId , Object value ) {
switch ( type ) {
@ -483,10 +457,22 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
String id = fromVersionedIdToObjectId ( versionedId ) ;
if ( id ! = null & & new LwM2mPath ( id ) . isResource ( ) & & ! client . isResourceMultiInstances ( versionedId , modelProvider ) ) {
return client . getDefaultContentFormat ( ) ;
}
else {
} else {
return ContentFormat . DEFAULT ;
}
}
}
private < R > String toString ( R request ) {
try {
try {
return JacksonUtil . toString ( request ) ;
} catch ( Exception e ) {
return request . toString ( ) ;
}
} catch ( Exception e ) {
log . warn ( "Failed to convert request to string" , e ) ;
return request ! = null ? request . getClass ( ) . getSimpleName ( ) : "" ;
}
}
}
Andrii Shvaika
5 years ago
Andrew Shvayka