CoAP Content format improvements

@@ -67,8 +67,6 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
     private static final int REQUEST_ID_POSITION_CERTIFICATE_REQUEST = 4;
     private static final int REQUEST_ID_POSITION_CERTIFICATE_REQUEST = 4;
     private static final String DTLS_SESSION_ID_KEY = "DTLS_SESSION_ID";
     private static final String DTLS_SESSION_ID_KEY = "DTLS_SESSION_ID";
 
 
-    private final ConcurrentMap<TbCoapClientState, ObserveRelation> sessionInfoToObserveRelationMap = new ConcurrentHashMap<>();
-
     private final ConcurrentMap<String, TbCoapDtlsSessionInfo> dtlsSessionIdMap;
     private final ConcurrentMap<String, TbCoapDtlsSessionInfo> dtlsSessionIdMap;
     private final long timeout;
     private final long timeout;
     private final CoapClientContext clients;
     private final CoapClientContext clients;

@@ -50,4 +50,5 @@ public interface CoapTransportAdaptor {
     ProvisionDeviceRequestMsg convertToProvisionRequestMsg(UUID sessionId, Request inbound) throws AdaptorException;
     ProvisionDeviceRequestMsg convertToProvisionRequestMsg(UUID sessionId, Request inbound) throws AdaptorException;
 
 
     int getContentFormat();
     int getContentFormat();
+
 }
 }

@@ -169,4 +169,5 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
     public int getContentFormat() {
     public int getContentFormat() {
         return MediaTypeRegistry.APPLICATION_JSON;
         return MediaTypeRegistry.APPLICATION_JSON;
     }
     }
+
 }
 }

@@ -24,6 +24,7 @@ import org.eclipse.californium.core.server.resources.CoapExchange;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.coap.client.TbCoapClientState;
 import org.thingsboard.server.transport.coap.client.TbCoapClientState;
+import org.thingsboard.server.transport.coap.client.TbCoapContentFormatUtil;
 import org.thingsboard.server.transport.coap.client.TbCoapObservationState;
 import org.thingsboard.server.transport.coap.client.TbCoapObservationState;
 
 
 import java.util.UUID;
 import java.util.UUID;
@@ -74,9 +75,7 @@ public abstract class AbstractSyncSessionCallback implements SessionMsgListener
     }
     }
 
 
     protected void respond(Response response) {
     protected void respond(Response response) {
-        int contentFormat = exchange.getRequestOptions().getContentFormat();
-        contentFormat = contentFormat != MediaTypeRegistry.UNDEFINED ? contentFormat : state.getContentFormat();
-        response.getOptions().setContentFormat(contentFormat);
+        response.getOptions().setContentFormat(TbCoapContentFormatUtil.getContentFormat(exchange.getRequestOptions().getContentFormat(), state.getContentFormat()));
         exchange.respond(response);
         exchange.respond(response);
     }
     }
 
 

@@ -34,7 +34,7 @@ public class GetAttributesSyncSessionCallback extends AbstractSyncSessionCallbac
     @Override
     @Override
     public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) {
     public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) {
         try {
         try {
-            respond(state.getAdaptor().convertToPublish(isConRequest(state.getAttrs()), msg));
+            respond(state.getAdaptor().convertToPublish(request.isConfirmable(), msg));
         } catch (AdaptorException e) {
         } catch (AdaptorException e) {
             log.trace("[{}] Failed to reply due to error", state.getDeviceId(), e);
             log.trace("[{}] Failed to reply due to error", state.getDeviceId(), e);
             exchange.respond(new Response(CoAP.ResponseCode.INTERNAL_SERVER_ERROR));
             exchange.respond(new Response(CoAP.ResponseCode.INTERNAL_SERVER_ERROR));

@@ -33,7 +33,7 @@ public class ToServerRpcSyncSessionCallback extends AbstractSyncSessionCallback
     @Override
     @Override
     public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg toServerResponse) {
     public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg toServerResponse) {
         try {
         try {
-            respond(state.getAdaptor().convertToPublish(isConRequest(state.getRpc()), toServerResponse));
+            respond(state.getAdaptor().convertToPublish(request.isConfirmable(), toServerResponse));
         } catch (AdaptorException e) {
         } catch (AdaptorException e) {
             log.trace("Failed to reply due to error", e);
             log.trace("Failed to reply due to error", e);
             exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
             exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);

@@ -329,9 +329,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
         state.lock();
         state.lock();
         try {
         try {
             if (state.getConfiguration() == null || state.getAdaptor() == null) {
             if (state.getConfiguration() == null || state.getAdaptor() == null) {
-                state.setConfiguration(getTransportConfigurationContainer(deviceProfile));
-                state.setAdaptor(getCoapTransportAdaptor(state.getConfiguration().isJsonPayload()));
-                state.setContentFormat(state.getAdaptor().getContentFormat());
+                initStateAdaptor(deviceProfile, state);
             }
             }
             if (state.getCredentials() == null) {
             if (state.getCredentials() == null) {
                 state.init(deviceCredentials);
                 state.init(deviceCredentials);
@@ -395,6 +393,12 @@ public class DefaultCoapClientContext implements CoapClientContext {
         }
         }
     }
     }
 
 
+    private void initStateAdaptor(DeviceProfile deviceProfile, TbCoapClientState state) throws AdaptorException {
+        state.setConfiguration(getTransportConfigurationContainer(deviceProfile));
+        state.setAdaptor(getCoapTransportAdaptor(state.getConfiguration().isJsonPayload()));
+        state.setContentFormat(state.getAdaptor().getContentFormat());
+    }
+
     private CoapTransportAdaptor getCoapTransportAdaptor(boolean jsonPayloadType) {
     private CoapTransportAdaptor getCoapTransportAdaptor(boolean jsonPayloadType) {
         return jsonPayloadType ? transportContext.getJsonCoapAdaptor() : transportContext.getProtoCoapAdaptor();
         return jsonPayloadType ? transportContext.getJsonCoapAdaptor() : transportContext.getProtoCoapAdaptor();
     }
     }
@@ -457,7 +461,23 @@ public class DefaultCoapClientContext implements CoapClientContext {
         }
         }
 
 
         @Override
         @Override
+        public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto newSessionInfo, DeviceProfile deviceProfile) {
+            try {
+                initStateAdaptor(deviceProfile, state);
+            } catch (AdaptorException e) {
+                log.warn("[{}] Failed to update device profile: ", deviceProfile.getId(), e);
+            }
+        }
+
+        @Override
         public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
         public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
+            if (deviceProfileOpt.isPresent()) {
+                try {
+                    initStateAdaptor(deviceProfileOpt.get(), state);
+                } catch (AdaptorException e) {
+                    log.warn("[{}] Failed to update device: ", device.getId(), e);
+                }
+            }
             state.onDeviceUpdate(device);
             state.onDeviceUpdate(device);
         }
         }
 
 
@@ -709,9 +729,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
     }
     }
 
 
     private void respond(CoapExchange exchange, Response response, int defContentFormat) {
     private void respond(CoapExchange exchange, Response response, int defContentFormat) {
-        int contentFormat = exchange.getRequestOptions().getContentFormat();
-        contentFormat = contentFormat != MediaTypeRegistry.UNDEFINED ? contentFormat : defContentFormat;
-        response.getOptions().setContentFormat(contentFormat);
+        response.getOptions().setContentFormat(TbCoapContentFormatUtil.getContentFormat(exchange.getRequestOptions().getContentFormat(), defContentFormat));
         exchange.respond(response);
         exchange.respond(response);
     }
     }
 }
 }

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.coap.client;
+
+import org.eclipse.californium.core.coap.MediaTypeRegistry;
+
+public class TbCoapContentFormatUtil {
+
+    public static int getContentFormat(int requestFormat, int adaptorFormat) {
+        if (isStrict(adaptorFormat)) {
+            return adaptorFormat;
+        } else {
+            return requestFormat != MediaTypeRegistry.UNDEFINED ? requestFormat : adaptorFormat;
+        }
+    }
+
+    public static boolean isStrict(int contentFormat) {
+        return contentFormat == MediaTypeRegistry.APPLICATION_OCTET_STREAM;
+    }
+}

@@ -317,7 +317,6 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl
                     this.updateResourcesValue(lwM2MClient, lwM2mResource, path);
                     this.updateResourcesValue(lwM2MClient, lwM2mResource, path);
                 }
                 }
             }
             }
-            clientContext.update(lwM2MClient);
             if (clientContext.awake(lwM2MClient)) {
             if (clientContext.awake(lwM2MClient)) {
                 // clientContext.awake calls clientContext.update
                 // clientContext.awake calls clientContext.update
                 log.debug("[{}] Device is awake", lwM2MClient.getEndpoint());
                 log.debug("[{}] Device is awake", lwM2MClient.getEndpoint());