lwm2m: comments1

application/src/main/java/org/eclipse/leshan/server/observation/ObservationServiceImpl.java

@@ -12,6 +12,21 @@
  * 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.
+ *
+ * Copyright (c) 2016 Sierra Wireless and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v20.html
+ * and the Eclipse Distribution License is available at
+ *    http://www.eclipse.org/org/documents/edl-v10.html.
+ *
+ * Contributors:
+ *     Sierra Wireless - initial API and implementation
+ *     Michał Wadowski (Orange) - Add Observe-Composite feature.
  */
 package org.eclipse.leshan.server.observation;
 

common/transport/lwm2m/pom.xml

@@ -97,7 +97,7 @@
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>
-            <version>4.2.0</version>
+            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.eclipse.californium</groupId>

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/bootstrap/LwM2MTransportBootstrapService.java

@@ -113,13 +113,8 @@ public class LwM2MTransportBootstrapService {
         serverCoapConfig.setTransient(DtlsConfig.DTLS_RECOMMENDED_CIPHER_SUITES_ONLY);
         serverCoapConfig.set(DtlsConfig.DTLS_RECOMMENDED_CIPHER_SUITES_ONLY, serverConfig.isRecommendedCiphers());
         serverCoapConfig.setTransient(DtlsConfig.DTLS_CONNECTION_ID_LENGTH);
-        int cid = 6;
+        serverCoapConfig.set(DtlsConfig.DTLS_CONNECTION_ID_LENGTH, 6);
-        if (cid > 0) {
-            serverCoapConfig.set(DtlsConfig.DTLS_CONNECTION_ID_LENGTH, cid);
-        }
-
         serverCoapConfig.set(DTLS_RECOMMENDED_CURVES_ONLY, serverConfig.isRecommendedSupportedGroups());
-
         serverCoapConfig.setTransient(DTLS_RETRANSMISSION_TIMEOUT);
         serverCoapConfig.set(DTLS_RETRANSMISSION_TIMEOUT, serverConfig.getDtlsRetransmissionTimeout(), MILLISECONDS);
 

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/secure/LwM2mCredentialsSecurityInfoValidator.java

@@ -67,7 +67,7 @@ public class LwM2mCredentialsSecurityInfoValidator {
                 new TransportServiceCallback<>() {
                     @Override
                     public void onSuccess(ValidateDeviceCredentialsResponse msg) {
-                        log.info("Validated credentials: [{}] [{}]", credentialsId, msg);
+                        log.trace("Validated credentials: [{}] [{}]", credentialsId, msg);
                         resultSecurityStore[0] = createSecurityInfo(credentialsId, msg, keyValue);
                         latch.countDown();
                     }

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2mTransportService.java

@@ -68,7 +68,6 @@ public class DefaultLwM2mTransportService implements LwM2MTransportService {
 
     public static final CipherSuite[] RPK_OR_X509_CIPHER_SUITES = {TLS_PSK_WITH_AES_128_CCM_8, TLS_PSK_WITH_AES_128_CBC_SHA256, TLS_ECDHE_ECDSA_WITH_AES_128_CCM_8, TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256};
     public static final CipherSuite[] PSK_CIPHER_SUITES = {TLS_PSK_WITH_AES_128_CCM_8, TLS_PSK_WITH_AES_128_CBC_SHA256};
-//    public static final CipherSuite[] PSK_CIPHER_SUITES = {TLS_PSK_WITH_AES_128_CCM_8};
 
     private final LwM2mTransportContext context;
     private final LwM2MTransportServerConfig config;
@@ -92,8 +91,7 @@ public class DefaultLwM2mTransportService implements LwM2MTransportService {
          * nameFile = "BC68JAR01A09_TO_BC68JAR01A10.bin"
          * "coap://host:port/{path}/{token}/{nameFile}"
          */
-        LwM2mTransportCoapResource otaCoapResource = new LwM2mTransportCoapResource(otaPackageDataCache, FIRMWARE_UPDATE_COAP_RESOURCE);
+        new LwM2mTransportCoapResource(otaPackageDataCache, FIRMWARE_UPDATE_COAP_RESOURCE);
-//        this.server.coap().getServer().add(otaCoapResource);
         this.context.setServer(server);
         this.startLhServer();
     }
@@ -168,16 +166,19 @@ public class DefaultLwM2mTransportService implements LwM2MTransportService {
         serverCoapConfig.set(DTLS_ROLE, SERVER_ONLY);
         serverCoapConfig.setTransient(DtlsConfig.DTLS_CONNECTION_ID_LENGTH);
         /**
-         * "Control usage of DTLS connection ID.", //
+         * "Control usage of DTLS connection ID.",
-         *                     "- 'on' to activate Connection ID support ", //
+         * If DTLS_CONNECTION_ID_LENGTH enables the use of a connection id, this node id could be used
-         *                     "  (same as -cid 6)", //
+         * to configure the generation of connection ids specific for node in a multi-node deployment (cluster).
-         *                     "- 'off' to deactivate it", //
+         * The value is used as first byte in generated connection ids.
-         *                     "- Positive value define the size in byte of CID generated.", //
+         * DTLS connection ID length. <blank> disabled,
-         *                     "- 0 value means we accept to use CID but will not generated one for foreign peer.", //
+         * 0 enables support without active use of CID.", defaultValue == null, minimumValue == 0);
-         *                     "Default: on"
+         *   "- 'on' to activate Connection ID support ",
+         *   "  (same as -cid 6)",
+         *   "- 'off' to deactivate it",
+         *   "- Positive value define the size in byte of CID generated.",
+         *   "- 0 value means we accept to use CID but will not generated one for foreign peer.",
          */
-        int cid = 6;
+        serverCoapConfig.set(DtlsConfig.DTLS_CONNECTION_ID_LENGTH, 6);
-        serverCoapConfig.set(DtlsConfig.DTLS_CONNECTION_ID_LENGTH, cid);
 
 
         /* Create DTLS Config */

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MNetworkConfig.java

@@ -28,7 +28,6 @@ import static org.eclipse.californium.core.config.CoapConfig.DEFAULT_BLOCKWISE_S
 public class LwM2MNetworkConfig {
 
     public static Configuration getCoapConfig(Configuration coapConfig, Integer serverPortNoSec, Integer serverSecurePort, LwM2MTransportServerConfig config) {
-//        Configuration coapConfig = new Configuration();
         coapConfig.set(CoapConfig.COAP_PORT, serverPortNoSec);
         coapConfig.set(CoapConfig.COAP_SECURE_PORT, serverSecurePort);
         /**

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mServerListener.java

@@ -93,7 +93,7 @@ public class LwM2mServerListener {
 
         @Override
         public void cancelled(Observation observation) {
-            log.info("Canceled Observation [RegistrationId:{}: {}].", observation.getRegistrationId(), observation instanceof SingleObservation ?
+            log.trace("Canceled Observation [RegistrationId:{}: {}].", observation.getRegistrationId(), observation instanceof SingleObservation ?
                     "SingleObservation: " + ((SingleObservation) observation).getPath() :
                     "CompositeObservation: " + ((CompositeObservation) observation).getPaths());
        }

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java

@@ -55,7 +55,7 @@ public class LwM2mSessionMsgListener implements GenericFutureListener<Future<? s
 
     @Override
     public void onAttributeUpdate(UUID sessionId, AttributeUpdateNotificationMsg attributeUpdateNotification) {
-        log.info("[{}] Received attributes update notification to device", sessionId);
+        log.trace("[{}] Received attributes update notification to device", sessionId);
         this.attributesService.onAttributesUpdate(attributeUpdateNotification, this.sessionInfo);
     }
 

pom.xml

@@ -131,7 +131,7 @@
         <antisamy.version>1.7.3</antisamy.version>
         <snmp4j.version>2.8.5</snmp4j.version>
         <!--         TEST SCOPE         -->
-        <awaitility.version>4.1.0</awaitility.version>
+        <awaitility.version>4.2.0</awaitility.version>
         <dbunit.version>2.7.2</dbunit.version>
         <java-websocket.version>1.5.2</java-websocket.version>
         <jupiter.version>5.9.3</jupiter.version> <!-- keep the same version as spring-boot-starter-test depend on jupiter-->

transport/lwm2m/src/main/java/org/eclipse/leshan/server/observation/ObservationServiceImpl.java

@@ -0,0 +1,296 @@
+/**
+ * Copyright © 2016-2024 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.
+ *
+ * Copyright (c) 2016 Sierra Wireless and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v20.html
+ * and the Eclipse Distribution License is available at
+ *    http://www.eclipse.org/org/documents/edl-v10.html.
+ *
+ * Contributors:
+ *     Sierra Wireless - initial API and implementation
+ *     Michał Wadowski (Orange) - Add Observe-Composite feature.
+ */
+package org.eclipse.leshan.server.observation;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.leshan.core.node.LwM2mPath;
+import org.eclipse.leshan.core.observation.CompositeObservation;
+import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.core.observation.SingleObservation;
+import org.eclipse.leshan.core.peer.LwM2mPeer;
+import org.eclipse.leshan.core.response.ObserveCompositeResponse;
+import org.eclipse.leshan.core.response.ObserveResponse;
+import org.eclipse.leshan.server.endpoint.LwM2mServerEndpoint;
+import org.eclipse.leshan.server.endpoint.LwM2mServerEndpointsProvider;
+import org.eclipse.leshan.server.profile.ClientProfile;
+import org.eclipse.leshan.server.registration.Registration;
+import org.eclipse.leshan.server.registration.RegistrationStore;
+import org.eclipse.leshan.server.registration.RegistrationUpdate;
+import org.eclipse.leshan.server.registration.UpdatedRegistration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Implementation of the {@link ObservationService} accessing the persisted observation via the provided
+ * {@link RegistrationStore}.
+ *
+ * When a new observation is added or changed or canceled, the registered listeners are notified.
+ */
+@Slf4j
+public class ObservationServiceImpl implements ObservationService, LwM2mNotificationReceiver {
+
+    private final Logger LOG = LoggerFactory.getLogger(ObservationServiceImpl.class);
+
+    private final RegistrationStore registrationStore;
+    private final LwM2mServerEndpointsProvider endpointProvider;
+    private final boolean updateRegistrationOnNotification;
+
+    private final List<ObservationListener> listeners = new CopyOnWriteArrayList<>();;
+
+    /**
+     * Creates an instance of {@link ObservationServiceImpl}
+     */
+    public ObservationServiceImpl(RegistrationStore store, LwM2mServerEndpointsProvider endpointProvider) {
+        this(store, endpointProvider, false);
+    }
+
+    /**
+     * Creates an instance of {@link ObservationServiceImpl}
+     *
+     * @param updateRegistrationOnNotification will activate registration update on observe notification.
+     *
+     * @since 1.1
+     */
+    public ObservationServiceImpl(RegistrationStore store, LwM2mServerEndpointsProvider endpointProvider,
+                                  boolean updateRegistrationOnNotification) {
+        this.registrationStore = store;
+        this.updateRegistrationOnNotification = updateRegistrationOnNotification;
+        this.endpointProvider = endpointProvider;
+    }
+
+    @Override
+    public int cancelObservations(Registration registration) {
+        // check registration id
+        String registrationId = registration.getId();
+        if (registrationId == null)
+            return 0;
+
+        Collection<Observation> observations = registrationStore.removeObservations(registrationId);
+        if (observations == null)
+            return 0;
+
+        for (Observation observation : observations) {
+            cancel(observation);
+        }
+
+        return observations.size();
+    }
+
+    @Override
+    public int cancelObservations(Registration registration, String nodePath) {
+        if (registration == null || registration.getId() == null || nodePath == null || nodePath.isEmpty())
+            return 0;
+
+        Set<Observation> observations = getObservationsForCancel(registration.getId(), nodePath);
+        for (Observation observation : observations) {
+            cancelObservation(observation);
+        }
+        return observations.size();
+    }
+
+    @Override
+    public int cancelCompositeObservations(Registration registration, String[] nodePaths) {
+        if (registration == null || registration.getId() == null || nodePaths == null || nodePaths.length == 0)
+            return 0;
+
+        Set<Observation> observations = getCompositeObservationsForCancel(registration.getId(), nodePaths);
+        for (Observation observation : observations) {
+            cancelObservation(observation);
+        }
+        return observations.size();
+    }
+
+    @Override
+    public void cancelObservation(Observation observation) {
+        if (observation == null)
+            return;
+
+        registrationStore.removeObservation(observation.getRegistrationId(), observation.getId());
+        cancel(observation);
+    }
+
+    private void cancel(Observation observation) {
+        List<LwM2mServerEndpoint> endpoints = endpointProvider.getEndpoints();
+        for (LwM2mServerEndpoint lwM2mEndpoint : endpoints) {
+            lwM2mEndpoint.cancelObservation(observation);
+        }
+
+        for (ObservationListener listener : listeners) {
+            listener.cancelled(observation);
+        }
+    }
+
+    @Override
+    public Set<Observation> getObservations(Registration registration) {
+        return getObservations(registration.getId());
+    }
+
+    private Set<Observation> getObservations(String registrationId) {
+        if (registrationId == null)
+            return Collections.emptySet();
+
+        return new HashSet<>(registrationStore.getObservations(registrationId));
+    }
+
+    private Set<Observation> getCompositeObservationsForCancel(String registrationId, String[] nodePaths) {
+        if (registrationId == null || nodePaths == null)
+            return Collections.emptySet();
+
+        // array of String to array of LWM2M path
+        List<LwM2mPath> lwPaths = new ArrayList<>(nodePaths.length);
+        for (int i = 0; i < nodePaths.length; i++) {
+            lwPaths.add(new LwM2mPath(nodePaths[i]));
+        }
+
+        // search composite-observation
+        Set<Observation> result = new HashSet<>();
+        for (Observation obs : getObservations(registrationId)) {
+            if (obs instanceof CompositeObservation) {
+                if (lwPaths.equals(((CompositeObservation) obs).getPaths())) {
+                    result.add(obs);
+                }
+            }
+        }
+        return result;
+    }
+
+    private Set<Observation> getObservationsForCancel(String registrationId, String nodePath) {
+        if (registrationId == null || nodePath == null)
+            return Collections.emptySet();
+
+        Set<Observation> result = new HashSet<>();
+        LwM2mPath lwPath = new LwM2mPath(nodePath);
+        for (Observation obs : getObservations(registrationId)) {
+            if (obs instanceof SingleObservation) {
+                LwM2mPath lwPathObs = ((SingleObservation) obs).getPath();
+                if (lwPath.equals(lwPathObs) || lwPathObs.startWith(lwPath)) {                // nodePath = "3",     lwPathObs = "3/0/9": cancel for tne all lwPathObs
+                    result.add(obs);
+                } else if (!lwPath.equals(lwPathObs) && lwPath.startWith(lwPathObs)) {        // nodePath = "3/0/9", lwPathObs = "3":     error...
+                        String errorMsg = String.format(
+                                "Unexpected error <cancelObservation>: There is registration with id [%s] existing observation [%s] includes input observation [%s]!",
+                                registrationId, lwPathObs, lwPath);
+                        throw new IllegalStateException(errorMsg);
+                }
+            }
+        }
+
+        return result;
+    }
+
+    @Override
+    public void addListener(ObservationListener listener) {
+        listeners.add(listener);
+    }
+
+    @Override
+    public void removeListener(ObservationListener listener) {
+        listeners.remove(listener);
+    }
+
+    private Registration updateRegistrationOnRegistration(Observation observation, LwM2mPeer sender,
+            ClientProfile profile) {
+        if (updateRegistrationOnNotification) {
+            RegistrationUpdate regUpdate = new RegistrationUpdate(observation.getRegistrationId(), sender, null, null,
+                    null, null, null, null, null, null, null, null);
+            UpdatedRegistration updatedRegistration = registrationStore.updateRegistration(regUpdate);
+            if (updatedRegistration == null || updatedRegistration.getUpdatedRegistration() == null) {
+                String errorMsg = String.format(
+                        "Unexpected error: There is no registration with id %s for this observation %s",
+                        observation.getRegistrationId(), observation);
+                LOG.error(errorMsg);
+                throw new IllegalStateException(errorMsg);
+            }
+            return updatedRegistration.getUpdatedRegistration();
+        }
+        return profile.getRegistration();
+    }
+
+    // ********** NotificationListener interface **********//
+    @Override
+    public void onNotification(SingleObservation observation, LwM2mPeer sender, ClientProfile profile,
+            ObserveResponse response) {
+        try {
+            Registration updatedRegistration = updateRegistrationOnRegistration(observation, sender, profile);
+            for (ObservationListener listener : listeners) {
+                listener.onResponse(observation, updatedRegistration, response);
+            }
+        } catch (Exception e) {
+            for (ObservationListener listener : listeners) {
+                listener.onError(observation, profile.getRegistration(), e);
+            }
+        }
+    }
+
+    @Override
+    public void onNotification(CompositeObservation observation, LwM2mPeer sender, ClientProfile profile,
+            ObserveCompositeResponse response) {
+        try {
+            Registration updatedRegistration = updateRegistrationOnRegistration(observation, sender, profile);
+            for (ObservationListener listener : listeners) {
+                listener.onResponse(observation, updatedRegistration, response);
+            }
+        } catch (Exception e) {
+            for (ObservationListener listener : listeners) {
+                listener.onError(observation, profile.getRegistration(), e);
+            }
+        }
+    }
+
+    @Override
+    public void onError(Observation observation, LwM2mPeer sender, ClientProfile profile, Exception error) {
+        for (ObservationListener listener : listeners) {
+            listener.onError(observation, profile.getRegistration(), error);
+        }
+    }
+
+    @Override
+    public void newObservation(Observation observation, Registration registration) {
+        for (ObservationListener listener : listeners) {
+            listener.newObservation(observation, registration);
+        }
+    }
+
+    @Override
+    public void cancelled(Observation observation) {
+        for (ObservationListener listener : listeners) {
+            listener.cancelled(observation);
+        }
+
+    }
+}