Repository: eclipse/leshan
Pull request #761 (2019-10-24 12:47:30)

add some executor housekeeping at server side


Patch:
diff --git a/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/request/CaliforniumLwM2mRequestSender.java b/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/request/CaliforniumLwM2mRequestSender.java
index b1208a64..6817cbe4 100644
--- a/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/request/CaliforniumLwM2mRequestSender.java
+++ b/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/request/CaliforniumLwM2mRequestSender.java
@@ -16,6 +16,8 @@
 package org.eclipse.leshan.client.californium.request;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 
 import org.eclipse.californium.core.coap.MessageObserver;
 import org.eclipse.californium.core.coap.Request;
@@ -28,9 +30,12 @@ import org.eclipse.leshan.core.request.UplinkRequest;
 import org.eclipse.leshan.core.response.ErrorCallback;
 import org.eclipse.leshan.core.response.LwM2mResponse;
 import org.eclipse.leshan.core.response.ResponseCallback;
+import org.eclipse.leshan.util.NamedThreadFactory;
 
 public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
 
+    private static final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
+            new NamedThreadFactory("Leshan Async Request timeout"));
     private final CaliforniumEndpointsManager endpointsManager;
 
     public CaliforniumLwM2mRequestSender(CaliforniumEndpointsManager endpointsManager) {
@@ -74,7 +79,8 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
         Request coapRequest = coapClientRequestBuilder.getRequest();
 
         // Add CoAP request callback
-        MessageObserver obs = new AsyncRequestObserver<T>(coapRequest, responseCallback, errorCallback, timeout) {
+        MessageObserver obs = new AsyncRequestObserver<T>(coapRequest, responseCallback, errorCallback, timeout,
+                executor) {
             @Override
             public T buildResponse(Response coapResponse) {
                 // Build LwM2m response
diff --git a/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/AsyncRequestObserver.java b/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/AsyncRequestObserver.java
index 704b59b2..6a3d048d 100644
--- a/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/AsyncRequestObserver.java
+++ b/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/AsyncRequestObserver.java
@@ -15,6 +15,8 @@
  *******************************************************************************/
 package org.eclipse.leshan.core.californium;
 
+import java.util.concurrent.ScheduledExecutorService;
+
 import org.eclipse.californium.core.coap.Request;
 import org.eclipse.californium.core.coap.Response;
 import org.eclipse.leshan.core.response.ErrorCallback;
@@ -44,10 +46,11 @@ public abstract class AsyncRequestObserver<T extends LwM2mResponse> extends Coap
      * @param errorCallback This is called when an error happens. This MUST NOT be null.
      * @param timeoutInMs A response timeout(in millisecond) which is raised if neither a response or error happens (see
      *        https://github.com/eclipse/leshan/wiki/Request-Timeout).
+     * @param executor used to scheduled timeout tasks.
      */
     public AsyncRequestObserver(Request coapRequest, final ResponseCallback<T> responseCallback,
-            final ErrorCallback errorCallback, long timeoutInMs) {
-        super(coapRequest, null, errorCallback, timeoutInMs);
+            final ErrorCallback errorCallback, long timeoutInMs, ScheduledExecutorService executor) {
+        super(coapRequest, null, errorCallback, timeoutInMs, executor);
         this.responseCallback = new CoapResponseCallback() {
 
             @Override
diff --git a/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/CoapAsyncRequestObserver.java b/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/CoapAsyncRequestObserver.java
index f5b583ac..c233e996 100644
--- a/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/CoapAsyncRequestObserver.java
+++ b/leshan-core-cf/src/main/java/org/eclipse/leshan/core/californium/CoapAsyncRequestObserver.java
@@ -18,7 +18,6 @@
  *******************************************************************************/
 package org.eclipse.leshan.core.californium;
 
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -30,7 +29,6 @@ import org.eclipse.leshan.core.request.exception.RequestCanceledException;
 import org.eclipse.leshan.core.request.exception.RequestRejectedException;
 import org.eclipse.leshan.core.request.exception.SendFailedException;
 import org.eclipse.leshan.core.response.ErrorCallback;
-import org.eclipse.leshan.util.NamedThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,13 +44,13 @@ import org.slf4j.LoggerFactory;
 public class CoapAsyncRequestObserver extends AbstractRequestObserver {
 
     private static final Logger LOG = LoggerFactory.getLogger(CoapAsyncRequestObserver.class);
-    private static volatile ScheduledExecutorService executor;
 
     protected CoapResponseCallback responseCallback;
     private final ErrorCallback errorCallback;
     private final long timeoutInMs;
     private ScheduledFuture<?> cleaningTask;
     private boolean cancelled = false;
+    private ScheduledExecutorService executor;
 
     // The Californium API does not ensure that message callback are exclusive
     // meaning that you can get a onReponse call and a onCancel one.
@@ -75,13 +73,15 @@ public class CoapAsyncRequestObserver extends AbstractRequestObserver {
      * @param errorCallback This is called when an error happens. This MUST NOT be null.
      * @param timeoutInMs A response timeout(in millisecond) which is raised if neither a response or error happens (see
      *        https://github.com/eclipse/leshan/wiki/Request-Timeout).
+     * @param executor used to scheduled timeout tasks.
      */
     public CoapAsyncRequestObserver(Request coapRequest, CoapResponseCallback responseCallback,
-            ErrorCallback errorCallback, long timeoutInMs) {
+            ErrorCallback errorCallback, long timeoutInMs, ScheduledExecutorService executor) {
         super(coapRequest);
         this.responseCallback = responseCallback;
         this.errorCallback = errorCallback;
         this.timeoutInMs = timeoutInMs;
+        this.executor = executor;
     }
 
     @Override
@@ -161,7 +161,7 @@ public class CoapAsyncRequestObserver extends AbstractRequestObserver {
         if (!cancelled)
             if (cleaningTask == null) {
                 LOG.trace("Schedule Cleaning Task for {}", coapRequest);
-                cleaningTask = getExecutor().schedule(new Runnable() {
+                cleaningTask = executor.schedule(new Runnable() {
                     @Override
                     public void run() {
                         responseTimedOut.set(true);
@@ -177,16 +177,4 @@ public class CoapAsyncRequestObserver extends AbstractRequestObserver {
         }
         cancelled = true;
     }
-
-    private ScheduledExecutorService getExecutor() {
-        if (executor == null) {
-            synchronized (this.getClass()) {
-                if (executor == null) {
-                    executor = Executors.newScheduledThreadPool(1,
-                            new NamedThreadFactory("Leshan Async Request timeout"));
-                }
-            }
-        }
-        return executor;
-    }
 }
\ No newline at end of file
diff --git a/leshan-server-cf/pom.xml b/leshan-server-cf/pom.xml
index f6704abd..d785c1d5 100644
--- a/leshan-server-cf/pom.xml
+++ b/leshan-server-cf/pom.xml
@@ -73,6 +73,14 @@ Contributors:
                     </instructions>
                 </configuration>
             </plugin>
+            <plugin>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <!-- We need to launch each tests in its own JVM to be able to check number 
+                         of active threads in LeshanBootstrapServerTest and LeshanServerTest -->
+                    <reuseForks>false</reuseForks>
+                </configuration>
+             </plugin>
         </plugins>
     </build>
 </project>
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServer.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServer.java
index 3afc07d2..3bba0a40 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServer.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServer.java
@@ -274,6 +274,9 @@ public class LeshanServer {
         if (securityStore instanceof Startable) {
             ((Startable) securityStore).start();
         }
+        if (requestSender instanceof Startable) {
+            ((Startable) requestSender).start();
+        }
 
         // Start server
         coapServer.start();
@@ -299,6 +302,9 @@ public class LeshanServer {
         if (securityStore instanceof Stoppable) {
             ((Stoppable) securityStore).stop();
         }
+        if (requestSender instanceof Stoppable) {
+            ((Stoppable) requestSender).stop();
+        }
 
         LOG.info("LWM2M server stopped.");
     }
@@ -325,6 +331,14 @@ public class LeshanServer {
             ((Stoppable) securityStore).stop();
         }
 
+        if (requestSender instanceof Destroyable) {
+            ((Destroyable) requestSender).destroy();
+        } else if (requestSender instanceof Stoppable) {
+            ((Stoppable) requestSender).stop();
+        }
+
+        presenceService.destroy();
+
         LOG.info("LWM2M server destroyed.");
     }
 
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/CaliforniumLwM2mBootstrapRequestSender.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/CaliforniumLwM2mBootstrapRequestSender.java
index 09c2ef65..b45c6cdf 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/CaliforniumLwM2mBootstrapRequestSender.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/CaliforniumLwM2mBootstrapRequestSender.java
@@ -28,6 +28,7 @@ import org.eclipse.leshan.core.request.exception.SendFailedException;
 import org.eclipse.leshan.core.response.ErrorCallback;
 import org.eclipse.leshan.core.response.LwM2mResponse;
 import org.eclipse.leshan.core.response.ResponseCallback;
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.bootstrap.BootstrapSession;
 import org.eclipse.leshan.server.bootstrap.LwM2mBootstrapRequestSender;
 import org.eclipse.leshan.server.californium.request.RequestSender;
@@ -37,7 +38,7 @@ import org.slf4j.LoggerFactory;
 /**
  * An implementation of {@link LwM2mBootstrapRequestSender} based on Californium.
  */
-public class CaliforniumLwM2mBootstrapRequestSender implements LwM2mBootstrapRequestSender {
+public class CaliforniumLwM2mBootstrapRequestSender implements LwM2mBootstrapRequestSender, Destroyable {
 
     static final Logger LOG = LoggerFactory.getLogger(CaliforniumLwM2mBootstrapRequestSender.class);
 
@@ -126,4 +127,9 @@ public class CaliforniumLwM2mBootstrapRequestSender implements LwM2mBootstrapReq
     public void cancelOngoingRequests(BootstrapSession destination) {
         sender.cancelRequests(destination.getId());
     }
+
+    @Override
+    public void destroy() {
+        sender.destroy();
+    }
 }
\ No newline at end of file
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServer.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServer.java
index 3f9a43be..306ecd25 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServer.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServer.java
@@ -26,6 +26,9 @@ import org.eclipse.californium.core.server.resources.Resource;
 import org.eclipse.leshan.core.model.LwM2mModel;
 import org.eclipse.leshan.core.node.codec.LwM2mNodeDecoder;
 import org.eclipse.leshan.core.node.codec.LwM2mNodeEncoder;
+import org.eclipse.leshan.server.Destroyable;
+import org.eclipse.leshan.server.Startable;
+import org.eclipse.leshan.server.Stoppable;
 import org.eclipse.leshan.server.bootstrap.BootstrapConfigStore;
 import org.eclipse.leshan.server.bootstrap.BootstrapHandler;
 import org.eclipse.leshan.server.bootstrap.BootstrapHandlerFactory;
@@ -54,6 +57,8 @@ public class LeshanBootstrapServer {
     private final BootstrapConfigStore bsStore;
     private final BootstrapSecurityStore bsSecurityStore;
 
+    private LwM2mBootstrapRequestSender requestSender;
+
     /**
      * /** Initialize a server which will bind to the specified address and port.
      * <p>
@@ -97,8 +102,7 @@ public class LeshanBootstrapServer {
             coapServer.addEndpoint(securedEndpoint);
 
         // create request sender
-        LwM2mBootstrapRequestSender requestSender = createRequestSender(securedEndpoint, unsecuredEndpoint, model,
-                encoder, decoder);
+        requestSender = createRequestSender(securedEndpoint, unsecuredEndpoint, model, encoder, decoder);
 
         // create bootstrap resource
         CoapResource bsResource = createBootstrapResource(
@@ -147,6 +151,9 @@ public class LeshanBootstrapServer {
      * Starts the server and binds it to the specified port.
      */
     public void start() {
+        if (requestSender instanceof Startable) {
+            ((Startable) requestSender).start();
+        }
         coapServer.start();
 
         if (LOG.isInfoEnabled()) {
@@ -161,6 +168,9 @@ public class LeshanBootstrapServer {
      */
     public void stop() {
         coapServer.stop();
+        if (requestSender instanceof Stoppable) {
+            ((Stoppable) requestSender).stop();
+        }
         LOG.info("Bootstrap server stopped.");
     }
 
@@ -171,6 +181,12 @@ public class LeshanBootstrapServer {
      */
     public void destroy() {
         coapServer.destroy();
+
+        if (requestSender instanceof Destroyable) {
+            ((Destroyable) requestSender).destroy();
+        } else if (requestSender instanceof Stoppable) {
+            ((Stoppable) requestSender).stop();
+        }
         LOG.info("Bootstrap server destroyed.");
     }
 
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/registration/InMemoryRegistrationStore.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/registration/InMemoryRegistrationStore.java
index 4be632a8..d468476b 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/registration/InMemoryRegistrationStore.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/registration/InMemoryRegistrationStore.java
@@ -37,6 +37,7 @@ import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -45,6 +46,7 @@ import org.eclipse.californium.core.coap.Token;
 import org.eclipse.californium.core.observe.ObservationUtil;
 import org.eclipse.californium.elements.EndpointContext;
 import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.Startable;
 import org.eclipse.leshan.server.Stoppable;
 import org.eclipse.leshan.server.californium.observation.ObserveUtil;
@@ -60,7 +62,7 @@ import org.slf4j.LoggerFactory;
 /**
  * An in memory store for registration and observation.
  */
-public class InMemoryRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable {
+public class InMemoryRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable, Destroyable {
     private final Logger LOG = LoggerFactory.getLogger(InMemoryRegistrationStore.class);
 
     // Data structure
@@ -76,6 +78,7 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     private ExpirationListener expirationListener;
 
     private final ScheduledExecutorService schedExecutor;
+    private ScheduledFuture<?> cleanerTask;
     private boolean started = false;
     private final long cleanPeriod; // in seconds
 
@@ -447,7 +450,7 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     public synchronized void start() {
         if (!started) {
             started = true;
-            schedExecutor.scheduleAtFixedRate(new Cleaner(), cleanPeriod, cleanPeriod, TimeUnit.SECONDS);
+            cleanerTask = schedExecutor.scheduleAtFixedRate(new Cleaner(), cleanPeriod, cleanPeriod, TimeUnit.SECONDS);
         }
     }
 
@@ -458,15 +461,27 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     public synchronized void stop() {
         if (started) {
             started = false;
-            schedExecutor.shutdownNow();
-            try {
-                schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                LOG.warn("Clean up registration thread was interrupted.", e);
+            if (cleanerTask != null) {
+                cleanerTask.cancel(false);
+                cleanerTask = null;
             }
         }
     }
 
+    /**
+     * Destroy "cleanup" scheduler.
+     */
+    @Override
+    public synchronized void destroy() {
+        started = false;
+        schedExecutor.shutdownNow();
+        try {
+            schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            LOG.warn("Destroying InMemoryRegistrationStore was interrupted.", e);
+        }
+    }
+
     private class Cleaner implements Runnable {
 
         @Override
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumLwM2mRequestSender.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumLwM2mRequestSender.java
index 8c0a1f08..9d5cd937 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumLwM2mRequestSender.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumLwM2mRequestSender.java
@@ -33,6 +33,7 @@ import org.eclipse.leshan.core.response.ErrorCallback;
 import org.eclipse.leshan.core.response.LwM2mResponse;
 import org.eclipse.leshan.core.response.ObserveResponse;
 import org.eclipse.leshan.core.response.ResponseCallback;
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.californium.observation.ObservationServiceImpl;
 import org.eclipse.leshan.server.model.LwM2mModelProvider;
 import org.eclipse.leshan.server.registration.Registration;
@@ -42,7 +43,7 @@ import org.eclipse.leshan.util.Validate;
 /**
  * An implementation of {@link LwM2mRequestSender} and {@link CoapRequestSender} based on Californium.
  */
-public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender, CoapRequestSender {
+public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender, CoapRequestSender, Destroyable {
 
     private final ObservationServiceImpl observationService;
     private final LwM2mModelProvider modelProvider;
@@ -218,4 +219,9 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender, CoapRe
         Validate.notNull(registration);
         sender.cancelRequests(registration.getId());
     }
+
+    @Override
+    public void destroy() {
+        sender.destroy();
+    }
 }
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumQueueModeRequestSender.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumQueueModeRequestSender.java
index b04d0c05..5454839d 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumQueueModeRequestSender.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/CaliforniumQueueModeRequestSender.java
@@ -21,6 +21,7 @@ import org.eclipse.leshan.core.californium.CoapResponseCallback;
 import org.eclipse.leshan.core.request.exception.ClientSleepingException;
 import org.eclipse.leshan.core.request.exception.TimeoutException;
 import org.eclipse.leshan.core.response.ErrorCallback;
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.queue.Presence;
 import org.eclipse.leshan.server.queue.PresenceServiceImpl;
 import org.eclipse.leshan.server.queue.QueueModeLwM2mRequestSender;
@@ -30,7 +31,8 @@ import org.eclipse.leshan.server.request.LwM2mRequestSender;
 /**
  * A {@link LwM2mRequestSender} and {@link CoapRequestSender} which supports LWM2M Queue Mode.
  */
-public class CaliforniumQueueModeRequestSender extends QueueModeLwM2mRequestSender implements CoapRequestSender {
+public class CaliforniumQueueModeRequestSender extends QueueModeLwM2mRequestSender
+        implements CoapRequestSender, Destroyable {
 
     /**
      * @param presenceService the presence service object for setting the client into {@link Presence#SLEEPING} when
@@ -128,4 +130,11 @@ public class CaliforniumQueueModeRequestSender extends QueueModeLwM2mRequestSend
             }
         });
     }
+
+    @Override
+    public void destroy() {
+        if (delegatedSender instanceof Destroyable) {
+            ((Destroyable) delegatedSender).destroy();
+        }
+    }
 }
\ No newline at end of file
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/RequestSender.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/RequestSender.java
index 01e537aa..706c62a9 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/RequestSender.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/request/RequestSender.java
@@ -18,6 +18,9 @@ package org.eclipse.leshan.server.californium.request;
 import java.util.SortedMap;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.eclipse.californium.core.coap.MessageObserver;
@@ -45,7 +48,9 @@ import org.eclipse.leshan.core.request.exception.SendFailedException;
 import org.eclipse.leshan.core.response.ErrorCallback;
 import org.eclipse.leshan.core.response.LwM2mResponse;
 import org.eclipse.leshan.core.response.ResponseCallback;
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.californium.bootstrap.CaliforniumLwM2mBootstrapRequestSender;
+import org.eclipse.leshan.util.NamedThreadFactory;
 import org.eclipse.leshan.util.Validate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,9 +60,13 @@ import org.slf4j.LoggerFactory;
  * <p>
  * It can also link requests to a kind of "session" and cancel all ongoing requests associated to a given "session".
  */
-public class RequestSender {
+public class RequestSender implements Destroyable {
+
     static final Logger LOG = LoggerFactory.getLogger(CaliforniumLwM2mBootstrapRequestSender.class);
 
+    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
+            new NamedThreadFactory("Leshan Async Request timeout"));
+
     private final Endpoint nonSecureEndpoint;
     private final Endpoint secureEndpoint;
     private final LwM2mNodeDecoder decoder;
@@ -190,7 +199,8 @@ public class RequestSender {
         final Request coapRequest = coapClientRequestBuilder.getRequest();
 
         // Add CoAP request callback
-        MessageObserver obs = new AsyncRequestObserver<T>(coapRequest, responseCallback, errorCallback, timeoutInMs) {
+        MessageObserver obs = new AsyncRequestObserver<T>(coapRequest, responseCallback, errorCallback, timeoutInMs,
+                executor) {
             @Override
             public T buildResponse(Response coapResponse) {
                 // Build LwM2m response
@@ -294,7 +304,8 @@ public class RequestSender {
         coapRequest.setDestinationContext(context);
 
         // Add CoAP request callback
-        MessageObserver obs = new CoapAsyncRequestObserver(coapRequest, responseCallback, errorCallback, timeoutInMs);
+        MessageObserver obs = new CoapAsyncRequestObserver(coapRequest, responseCallback, errorCallback, timeoutInMs,
+                executor);
         coapRequest.addMessageObserver(obs);
 
         // Store pending request to be able to cancel it later
@@ -390,4 +401,14 @@ public class RequestSender {
             removeOngoingRequest(requestKey, coapRequest);
         }
     }
+
+    @Override
+    public void destroy() {
+        executor.shutdownNow();
+        try {
+            executor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            LOG.warn("Destroying RequestSender was interrupted.", e);
+        }
+    }
 }
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/PresenceServiceImpl.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/PresenceServiceImpl.java
index 191f29a5..44fa2236 100644
--- a/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/PresenceServiceImpl.java
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/queue/PresenceServiceImpl.java
@@ -25,7 +25,11 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.registration.Registration;
+import org.eclipse.leshan.util.NamedThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Tracks the status of each LWM2M client registered with Queue mode binding. Also ensures that the
@@ -34,12 +38,14 @@ import org.eclipse.leshan.server.registration.Registration;
  * 
  * @see Presence
  */
-public final class PresenceServiceImpl implements PresenceService {
+public final class PresenceServiceImpl implements PresenceService, Destroyable {
+    private final Logger LOG = LoggerFactory.getLogger(PresenceServiceImpl.class);
 
     private final ConcurrentMap<String, PresenceStatus> clientStatusList = new ConcurrentHashMap<>();
     private final List<PresenceListener> listeners = new CopyOnWriteArrayList<>();
     private final ClientAwakeTimeProvider awakeTimeProvider;
-    private final ScheduledExecutorService clientTimersExecutor = Executors.newSingleThreadScheduledExecutor();
+    private final ScheduledExecutorService clientTimersExecutor = Executors
+            .newSingleThreadScheduledExecutor(new NamedThreadFactory("Presence Service"));
 
     public PresenceServiceImpl(ClientAwakeTimeProvider awakeTimeProvider) {
         this.awakeTimeProvider = awakeTimeProvider;
@@ -186,4 +192,14 @@ public final class PresenceServiceImpl implements PresenceService {
             setSleeping(reg);
         }
     }
+
+    @Override
+    public void destroy() {
+        clientTimersExecutor.shutdownNow();
+        try {
+            clientTimersExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            LOG.warn("Destroying presence service was interrupted.", e);
+        }
+    }
 }
diff --git a/leshan-server-redis/src/main/java/org/eclipse/leshan/server/redis/RedisRegistrationStore.java b/leshan-server-redis/src/main/java/org/eclipse/leshan/server/redis/RedisRegistrationStore.java
index a38d7a64..69f8c47b 100644
--- a/leshan-server-redis/src/main/java/org/eclipse/leshan/server/redis/RedisRegistrationStore.java
+++ b/leshan-server-redis/src/main/java/org/eclipse/leshan/server/redis/RedisRegistrationStore.java
@@ -32,11 +32,13 @@ import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.eclipse.californium.core.coap.Token;
 import org.eclipse.californium.elements.EndpointContext;
 import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.Destroyable;
 import org.eclipse.leshan.server.Startable;
 import org.eclipse.leshan.server.Stoppable;
 import org.eclipse.leshan.server.californium.observation.ObserveUtil;
@@ -61,7 +63,7 @@ import redis.clients.util.Pool;
 /**
  * A RegistrationStore which stores registrations and observations in Redis.
  */
-public class RedisRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable {
+public class RedisRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable, Destroyable {
 
     /** Default time in seconds between 2 cleaning tasks (used to remove expired registration). */
     public static final long DEFAULT_CLEAN_PERIOD = 60;
@@ -87,6 +89,7 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
     private ExpirationListener expirationListener;
 
     private final ScheduledExecutorService schedExecutor;
+    private ScheduledFuture<?> cleanerTask;
     private boolean started = false;
 
     private final long cleanPeriod; // in seconds
@@ -696,7 +699,7 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
     public synchronized void start() {
         if (!started) {
             started = true;
-            schedExecutor.scheduleAtFixedRate(new Cleaner(), cleanPeriod, cleanPeriod, TimeUnit.SECONDS);
+            cleanerTask = schedExecutor.scheduleAtFixedRate(new Cleaner(), cleanPeriod, cleanPeriod, TimeUnit.SECONDS);
         }
     }
 
@@ -707,15 +710,27 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
     public synchronized void stop() {
         if (started) {
             started = false;
-            schedExecutor.shutdownNow();
-            try {
-                schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                LOG.warn("Clean up registration thread was interrupted.", e);
+            if (cleanerTask != null) {
+                cleanerTask.cancel(false);
+                cleanerTask = null;
             }
         }
     }
 
+    /**
+     * Destroy "cleanup" scheduler.
+     */
+    @Override
+    public synchronized void destroy() {
+        started = false;
+        schedExecutor.shutdownNow();
+        try {
+            schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            LOG.warn("Destroying RedisRegistrationStore was interrupted.", e);
+        }
+    }
+
     private class Cleaner implements Runnable {
 
         @Override


Test patch:
diff --git a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/LeshanServerTest.java b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/LeshanServerTest.java
new file mode 100644
index 00000000..da4f1473
--- /dev/null
+++ b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/LeshanServerTest.java
@@ -0,0 +1,99 @@
+/*******************************************************************************
+ * Copyright (c) 2019 Sierra Wireless and others.
+ * 
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ * 
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.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
+ *******************************************************************************/
+package org.eclipse.leshan.server.californium;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.InetSocketAddress;
+
+import org.eclipse.leshan.core.request.BindingMode;
+import org.eclipse.leshan.core.request.Identity;
+import org.eclipse.leshan.core.request.ReadRequest;
+import org.eclipse.leshan.core.response.ErrorCallback;
+import org.eclipse.leshan.core.response.ReadResponse;
+import org.eclipse.leshan.core.response.ResponseCallback;
+import org.eclipse.leshan.server.queue.PresenceServiceImpl;
+import org.eclipse.leshan.server.registration.Registration;
+import org.junit.Test;
+
+public class LeshanServerTest {
+
+    @Test
+    public void testStartStopStart() throws InterruptedException {
+        LeshanServer server = new LeshanServerBuilder().build();
+
+        server.start();
+        Thread.sleep(100);
+        server.stop();
+        Thread.sleep(100);
+        server.start();
+    }
+
+    @Test
+    public void testStartDestroy() throws InterruptedException {
+        // look at nb active thread before.
+        int numberOfThreadbefore = Thread.activeCount();
+
+        LeshanServer server = new LeshanServerBuilder().build();
+        server.start();
+        Thread.sleep(100);
+        // HACK force creation thread creation.
+        forceThreadsCreation(server);
+        Thread.sleep(100);
+        server.destroy();
+
+        // ensure all thread are destroyed
+        Thread.sleep(500);
+        assertEquals("All news created threads must be destroyed", numberOfThreadbefore, Thread.activeCount());
+    }
+
+    @Test
+    public void testStartStopDestroy() throws InterruptedException {
+        // look at nb active thread before.
+        int numberOfThreadbefore = Thread.activeCount();
+
+        LeshanServer server = new LeshanServerBuilder().build();
+        server.start();
+        Thread.sleep(100);
+        // HACK force creation thread creation.
+        forceThreadsCreation(server);
+        Thread.sleep(100);
+        server.stop();
+        Thread.sleep(100);
+        server.destroy();
+
+        // ensure all thread are destroyed
+        Thread.sleep(500);
+        assertEquals("All news created threads must be destroyed", numberOfThreadbefore, Thread.activeCount());
+    }
+
+    private void forceThreadsCreation(LeshanServer server) {
+        Registration reg = new Registration.Builder("id", "endpoint", Identity.unsecure(new InetSocketAddress(5555)))
+                .bindingMode(BindingMode.UQ).build();
+        // Force timer thread creation of preference service.
+        ((PresenceServiceImpl) server.getPresenceService()).setAwake(reg);
+        // Force time thread creation of CoapAsyncRequestObserver
+        server.send(reg, new ReadRequest(3), new ResponseCallback<ReadResponse>() {
+            @Override
+            public void onResponse(ReadResponse response) {
+            }
+        }, new ErrorCallback() {
+            @Override
+            public void onError(Exception e) {
+            }
+        });
+    }
+}
diff --git a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServerTest.java b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServerTest.java
new file mode 100644
index 00000000..520913f1
--- /dev/null
+++ b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/bootstrap/LeshanBootstrapServerTest.java
@@ -0,0 +1,117 @@
+/*******************************************************************************
+ * Copyright (c) 2019 Sierra Wireless and others.
+ * 
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ * 
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.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
+ *******************************************************************************/
+package org.eclipse.leshan.server.californium.bootstrap;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.InetSocketAddress;
+
+import org.eclipse.leshan.core.request.BootstrapRequest;
+import org.eclipse.leshan.core.request.Identity;
+import org.eclipse.leshan.core.response.BootstrapResponse;
+import org.eclipse.leshan.core.response.SendableResponse;
+import org.eclipse.leshan.server.bootstrap.BootstrapConfig;
+import org.eclipse.leshan.server.bootstrap.BootstrapConfigStore;
+import org.eclipse.leshan.server.bootstrap.BootstrapHandler;
+import org.eclipse.leshan.server.bootstrap.BootstrapHandlerFactory;
+import org.eclipse.leshan.server.bootstrap.BootstrapSession;
+import org.eclipse.leshan.server.bootstrap.BootstrapSessionManager;
+import org.eclipse.leshan.server.bootstrap.DefaultBootstrapHandler;
+import org.eclipse.leshan.server.bootstrap.LwM2mBootstrapRequestSender;
+import org.junit.Test;
+
+public class LeshanBootstrapServerTest {
+
+    private BootstrapHandler bsHandler;
+
+    private LeshanBootstrapServer createBootstrapServer() {
+        LeshanBootstrapServerBuilder builder = new LeshanBootstrapServerBuilder();
+        builder.setBootstrapHandlerFactory(new BootstrapHandlerFactory() {
+
+            @Override
+            public BootstrapHandler create(BootstrapConfigStore store, LwM2mBootstrapRequestSender sender,
+                    BootstrapSessionManager sessionManager) {
+                bsHandler = new DefaultBootstrapHandler(store, sender, sessionManager);
+                return bsHandler;
+            }
+        });
+        builder.setConfigStore(new BootstrapConfigStore() {
+
+            @Override
+            public BootstrapConfig get(String endpoint, Identity deviceIdentity, BootstrapSession session) {
+                BootstrapConfig config = new BootstrapConfig();
+                config.toDelete.add("/");
+                return config;
+            }
+        });
+        return builder.build();
+    }
+
+    @Test
+    public void testStartStopStart() throws InterruptedException {
+        LeshanBootstrapServer server = createBootstrapServer();
+
+        server.start();
+        Thread.sleep(100);
+        server.stop();
+        Thread.sleep(100);
+        server.start();
+    }
+
+    @Test
+    public void testStartDestroy() throws InterruptedException {
+        // look at nb active thread before.
+        int numberOfThreadbefore = Thread.activeCount();
+
+        LeshanBootstrapServer server = createBootstrapServer();
+        server.start();
+        Thread.sleep(100);
+        // HACK force creation thread creation.
+        forceThreadsCreation();
+        Thread.sleep(100);
+        server.destroy();
+
+        // ensure all thread are destroyed
+        Thread.sleep(500);
+        assertEquals("All news created threads must be destroyed", numberOfThreadbefore, Thread.activeCount());
+    }
+
+    @Test
+    public void testStartStopDestroy() throws InterruptedException {
+        // look at nb active thread before.
+        int numberOfThreadbefore = Thread.activeCount();
+
+        LeshanBootstrapServer server = createBootstrapServer();
+        server.start();
+        Thread.sleep(100);
+        // HACK force creation thread creation.
+        forceThreadsCreation();
+        Thread.sleep(100);
+        server.stop();
+        Thread.sleep(100);
+        server.destroy();
+
+        // ensure all thread are destroyed
+        Thread.sleep(500);
+        assertEquals("All news created threads must be destroyed", numberOfThreadbefore, Thread.activeCount());
+    }
+
+    private void forceThreadsCreation() {
+        SendableResponse<BootstrapResponse> bootstrap = bsHandler
+                .bootstrap(Identity.unsecure(new InetSocketAddress(5683)), new BootstrapRequest("test"));
+        bootstrap.sent();
+    }
+}