Repository: eclipse/leshan
Pull request #167 (2016-08-18 09:12:31)

Use Client object in ResponseListener instead of clientEndpoint
Added changes to ResponseListener as described in #165 ,


Patch:
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumLwM2mRequestSender.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumLwM2mRequestSender.java
index 9034f749..7092ec71 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumLwM2mRequestSender.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumLwM2mRequestSender.java
@@ -151,8 +151,8 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
     @Override
     public <T extends LwM2mResponse> void send(final Client destination, final String requestTicket,
             final DownlinkRequest<T> request) {
-        send(destination, request, (ResponseCallback<T>) getResponseCallback(destination.getEndpoint(), requestTicket),
-                getErrorCallback(destination.getEndpoint(), requestTicket));
+        send(destination, request, (ResponseCallback<T>) getResponseCallback(destination, requestTicket),
+                getErrorCallback(destination, requestTicket));
     }
 
     /**
@@ -161,13 +161,13 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
      * @parm requestTicket to correlate the response to a request.
      * @return callback which is invoked on getting a response from LWM2M Client.
      */
-    private <T extends LwM2mResponse> ResponseCallback<T> getResponseCallback(final String clientEndpoint,
+    private <T extends LwM2mResponse> ResponseCallback<T> getResponseCallback(final Client client,
             final String requestTicket) {
         return new ResponseCallback<T>() {
             @Override
             public void onResponse(T response) {
-                processingExecutor.execute(
-                        new ResponseProcessingTask(clientEndpoint, requestTicket, responseListeners, response));
+                processingExecutor
+                        .execute(new ResponseProcessingTask(client, requestTicket, responseListeners, response));
             }
         };
     }
@@ -178,12 +178,11 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
      * @param requestTicket to correlate the error to a request.
      * @return
      */
-    private ErrorCallback getErrorCallback(final String clientEndpoint, final String requestTicket) {
+    private ErrorCallback getErrorCallback(final Client client, final String requestTicket) {
         return new ErrorCallback() {
             @Override
             public void onError(Exception e) {
-                processingExecutor
-                        .execute(new ResponseProcessingTask(clientEndpoint, requestTicket, responseListeners, e));
+                processingExecutor.execute(new ResponseProcessingTask(client, requestTicket, responseListeners, e));
             }
         };
     }
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/impl/QueuedRequestSender.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/impl/QueuedRequestSender.java
index 081ee188..d778db8d 100644
--- a/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/impl/QueuedRequestSender.java
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/impl/QueuedRequestSender.java
@@ -163,18 +163,16 @@ public class QueuedRequestSender implements LwM2mRequestSender, Stoppable {
             LOG.debug("Cancelling and removing all pending messages for client {}", client.getEndpoint());
         }
 
-        String clientEndpoint = client.getEndpoint();
-
         // Potentially the first message in the queue would have been sent.
         // So try to cancel it now.
         delegateSender.cancelPendingRequests(client);
 
         // It is better to notify the listener with a RequestCanceledException
         // for each queued message to keep the behavior consistent with CaliforniumLwM2mRequestSender.
-        List<QueuedRequest> removedMessages = messageStore.removeAll(clientEndpoint);
+        List<QueuedRequest> removedMessages = messageStore.removeAll(client.getEndpoint());
         for (QueuedRequest request : removedMessages) {
-            processingExecutor.execute(new ResponseProcessingTask(clientEndpoint, request.getRequestTicket(),
-                    responseListeners, new RequestCanceledException("Queued message cancelled")));
+            processingExecutor.execute(new ResponseProcessingTask(client, request.getRequestTicket(), responseListeners,
+                    new RequestCanceledException("Queued message cancelled")));
         }
     }
 
@@ -187,48 +185,44 @@ public class QueuedRequestSender implements LwM2mRequestSender, Stoppable {
         return new ResponseListener() {
 
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
-                Client registrationInfo = clientRegistry.get(clientEndpoint);
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 // process only if the client has used Queue mode.
-                if (registrationInfo != null && registrationInfo.usesQueueMode()) {
+                if (client != null && client.usesQueueMode()) {
                     LOG.trace("response received in Queue mode successfully: {}", requestTicket);
-                    processResponse(clientEndpoint, requestTicket, response);
+                    processResponse(client, requestTicket, response);
                 }
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
-                Client registrationInfo = clientRegistry.get(clientEndpoint);
+            public void onError(Client client, String requestTicket, Exception exception) {
                 // process only if the client has used Queue mode.
-                if (registrationInfo != null && registrationInfo.usesQueueMode()) {
+                if (client != null && client.usesQueueMode()) {
                     LOG.debug("exception on sending the request: {}", requestTicket, exception);
                     if (exception instanceof TimeoutException) {
-                        timeout(clientEndpoint);
+                        timeout(client.getEndpoint());
                     } else {
-                        processException(clientEndpoint, requestTicket, exception);
+                        processException(client, requestTicket, exception);
                     }
                 }
             }
         };
     }
 
-    private void processResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+    private void processResponse(Client client, String requestTicket, LwM2mResponse response) {
         LOG.debug("Received Response -> {}", requestTicket);
-        messageStore.deleteFirst(clientEndpoint);
-        processingExecutor
-                .execute(new ResponseProcessingTask(clientEndpoint, requestTicket, responseListeners, response));
-        processingExecutor.execute(newRequestSendingTask(clientEndpoint));
+        messageStore.deleteFirst(client.getEndpoint());
+        processingExecutor.execute(new ResponseProcessingTask(client, requestTicket, responseListeners, response));
+        processingExecutor.execute(newRequestSendingTask(client.getEndpoint()));
     }
 
-    private void processException(String clientEndpoint, String requestTicket, Exception exception) {
+    private void processException(Client client, String requestTicket, Exception exception) {
         LOG.debug("Received error response {}", requestTicket);
-        messageStore.deleteFirst(clientEndpoint);
-        processingExecutor
-                .execute(new ResponseProcessingTask(clientEndpoint, requestTicket, responseListeners, exception));
+        messageStore.deleteFirst(client.getEndpoint());
+        processingExecutor.execute(new ResponseProcessingTask(client, requestTicket, responseListeners, exception));
         // If RequestCanceledException is thrown due to cancelPendingMessages call, then there
         // is no use processing next requests which would be removed in the next few moments.
         if (!(exception instanceof RequestCanceledException)) {
-            processingExecutor.execute(newRequestSendingTask(clientEndpoint));
+            processingExecutor.execute(newRequestSendingTask(client.getEndpoint()));
         }
     }
 
@@ -237,14 +231,23 @@ public class QueuedRequestSender implements LwM2mRequestSender, Stoppable {
         clientStatusTracker.setClientUnreachable(clientEndpoint);
     }
 
-    private RequestSendingTask newRequestSendingTask(String endpoint) {
-        return new RequestSendingTask(clientRegistry, delegateSender, clientStatusTracker, messageStore, endpoint);
+    private RequestSendingTask newRequestSendingTask(String clientEndpoint) {
+        return new RequestSendingTask(clientRegistry, delegateSender, clientStatusTracker, messageStore,
+                clientEndpoint);
     }
 
     private final class QueueModeObservationRegistryListener implements ObservationRegistryListener {
         @Override
         public void newValue(Observation observation, LwM2mNode value, List<TimestampedLwM2mNode> timestampedValues) {
             Client client = clientRegistry.findByRegistrationId(observation.getRegistrationId());
+            // if client de-registered already, do nothing.
+            if (client == null) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("No registered client found for registrationId: {}. Request is not sent.",
+                            observation.getRegistrationId());
+                }
+                return;
+            }
             if (client.usesQueueMode() && clientStatusTracker.setClientReachable(client.getEndpoint())
                     && clientStatusTracker.startClientReceiving(client.getEndpoint())) {
                 if (LOG.isDebugEnabled()) {
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseListener.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseListener.java
index 01265c02..d627a644 100644
--- a/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseListener.java
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseListener.java
@@ -16,6 +16,7 @@
 package org.eclipse.leshan.server.response;
 
 import org.eclipse.leshan.core.response.LwM2mResponse;
+import org.eclipse.leshan.server.client.Client;
 
 /**
  * The listener is responsible for handling all the success and error responses for a given requestTicket.
@@ -25,18 +26,18 @@ public interface ResponseListener {
     /**
      * this method is invoked when a response is received from LWM2M Client correlated by the request ticket.
      *
-     * @param clientEndpoint unique identifier of the LWM2M client
+     * @param client the LWM2M client
      * @param requestTicket globally unique identifier used to correlate the response to the orginial request
      * @param response from LWM2M client
      */
-    void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response);
+    void onResponse(Client client, String requestTicket, LwM2mResponse response);
 
     /**
      * this method is invoked when a an error response is received from LWM2M Client correlated by the request ticket.
      *
-     * @param clientEndpoint unique identifier of the LWM2M client
+     * @param client the LWM2M client
      * @param requestTicket globally unique identifier used to correlate the response to the orginial request
      * @param exception error from LWM2M client
      */
-    void onError(String clientEndpoint, String requestTicket, Exception exception);
+    void onError(Client clien, String requestTicket, Exception exception);
 }
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseProcessingTask.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseProcessingTask.java
index 550388a2..7cd73496 100644
--- a/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseProcessingTask.java
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/response/ResponseProcessingTask.java
@@ -19,6 +19,7 @@ package org.eclipse.leshan.server.response;
 import java.util.Collection;
 
 import org.eclipse.leshan.core.response.LwM2mResponse;
+import org.eclipse.leshan.server.client.Client;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,43 +35,43 @@ public class ResponseProcessingTask implements Runnable {
     private final boolean hasException;
     private final LwM2mResponse response;
     private final Collection<ResponseListener> responseListeners;
-    private final String endpoint;
+    private final Client client;
     private final String requestTicket;
 
     /**
      * Creates a new task for processing response on an ordinary response result.
      *
-     * @param clientEndpoint
+     * @param client
      * @param request request being processed
      * @param responseContext response context map for mapping response ID to the callback
      * @param response response to propagate
      */
-    public ResponseProcessingTask(String clientEndpoint, String requestTicket,
-            Collection<ResponseListener> responseListeners, LwM2mResponse response) {
+    public ResponseProcessingTask(Client client, String requestTicket, Collection<ResponseListener> responseListeners,
+            LwM2mResponse response) {
         this.requestTicket = requestTicket;
         this.exception = null;
         this.hasException = false;
         this.responseListeners = responseListeners;
         this.response = response;
-        this.endpoint = clientEndpoint;
+        this.client = client;
     }
 
     /**
      * Creates a new task for processing response on exception.
      *
-     * @param clientEndpoint
+     * @param client
      * @param request request being processed
      * @param responseListeners response context map for mapping response ID to the callback
      * @param exception exception to propagate
      */
-    public ResponseProcessingTask(String clientEndpoint, String requestTicket,
-            Collection<ResponseListener> responseListeners, Exception exception) {
+    public ResponseProcessingTask(Client client, String requestTicket, Collection<ResponseListener> responseListeners,
+            Exception exception) {
         this.requestTicket = requestTicket;
         this.exception = exception;
         this.responseListeners = responseListeners;
         this.hasException = true;
         this.response = null;
-        this.endpoint = clientEndpoint;
+        this.client = client;
     }
 
     @Override
@@ -91,11 +92,11 @@ public class ResponseProcessingTask implements Runnable {
                 if (hasException) {
                     LOG.debug("invoke response listener for requestTicket {} with exception {}", requestTicket,
                             exception);
-                    listener.onError(endpoint, requestTicket, exception);
+                    listener.onError(client, requestTicket, exception);
                 } else {
                     LOG.trace("invoke response listener for requestTicket {} with response {}", requestTicket,
                             response);
-                    listener.onResponse(endpoint, requestTicket, response);
+                    listener.onResponse(client, requestTicket, response);
                 }
             }
         }


Test patch:
diff --git a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeTest.java b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeTest.java
index b1b82ec4..95f1bb97 100644
--- a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeTest.java
+++ b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeTest.java
@@ -43,6 +43,7 @@ import org.eclipse.leshan.core.response.ObserveResponse;
 import org.eclipse.leshan.core.response.WriteResponse;
 import org.eclipse.leshan.integration.tests.util.QueuedModeLeshanClient;
 import org.eclipse.leshan.integration.tests.util.QueuedModeLeshanClient.OnGetCallback;
+import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.observation.ObservationRegistryListener;
 import org.eclipse.leshan.server.queue.MessageStore;
 import org.eclipse.leshan.server.queue.impl.InMemoryMessageStore;
@@ -232,7 +233,7 @@ public class QueueModeTest {
 
         responseListener = new ResponseListener() {
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 if (response instanceof ObserveResponse) {
                     Observation observation = ((ObserveResponse) response).getObservation();
                     assertEquals("/3/0/15", observation.getPath().toString());
@@ -241,7 +242,7 @@ public class QueueModeTest {
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 throw new IllegalStateException("unexpected exception occurred: ", exception);
             }
         };
@@ -289,7 +290,7 @@ public class QueueModeTest {
 
         responseListener = new ResponseListener() {
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 if (response instanceof ObserveResponse) {
                     LOG.trace("Received observe response for ticket {} from LWM2M client {}", requestTicket, response);
                     Observation observation = ((ObserveResponse) response).getObservation();
@@ -299,7 +300,7 @@ public class QueueModeTest {
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 throw new IllegalStateException("unexpected exception occurred: ", exception);
             }
         };
@@ -324,7 +325,7 @@ public class QueueModeTest {
         createAndAddResponseListener(acceptCountDownLatch);
         responseListener = new ResponseListener() {
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 if (response instanceof WriteResponse) {
                     LOG.trace("Received write response for ticket {} from LWM2M client {}", requestTicket, response);
                     acceptCountDownLatch.countDown();
@@ -333,7 +334,7 @@ public class QueueModeTest {
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 throw new IllegalStateException("unexpected exception occurred: ", exception);
             }
         };
@@ -366,7 +367,7 @@ public class QueueModeTest {
         final CountDownLatch serverResponseReceivedCountDownLatch = new CountDownLatch(2);
         responseListener = new ResponseListener() {
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 if (response instanceof ObserveResponse) {
                     serverResponseReceivedCountDownLatch.countDown();
                     LOG.trace("Received observe response for ticket {} from LWM2M client {}", requestTicket, response);
@@ -377,7 +378,7 @@ public class QueueModeTest {
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 throw new IllegalStateException("unexpected exception occurred: ", exception);
             }
         };
@@ -499,12 +500,12 @@ public class QueueModeTest {
         helper.server.getLwM2mRequestSender().addResponseListener(new ResponseListener() {
 
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 responseCountDownLatch.countDown();
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 assertTrue("Expected a RequestCanceledException but received " + exception,
                         exception instanceof RequestCanceledException);
                 exceptionCountDownLatch.countDown();
@@ -539,12 +540,12 @@ public class QueueModeTest {
         helper.server.getLwM2mRequestSender().addResponseListener(new ResponseListener() {
 
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 unexpectedResponseCountDownLatch.countDown();
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 assertTrue("Expected a RequestCanceledException but received " + exception,
                         exception instanceof RequestCanceledException);
                 exceptionCountDownLatch.countDown();
@@ -571,12 +572,12 @@ public class QueueModeTest {
         responseListener = new ResponseListener() {
 
             @Override
-            public void onResponse(String clientEndpoint, String requestTicket, LwM2mResponse response) {
+            public void onResponse(Client client, String requestTicket, LwM2mResponse response) {
                 acceptCountDownLatch.countDown();
             }
 
             @Override
-            public void onError(String clientEndpoint, String requestTicket, Exception exception) {
+            public void onError(Client client, String requestTicket, Exception exception) {
                 throw new IllegalStateException("unexpected exception occurred: ", exception);
             }
         };