Repository: eclipse/leshan
Pull request #210 (2016-11-29 15:06:52)

Create a new store which handles registrations and observations.
I refactor the code to create a new store which handles registrations and observations.

This aims to fix several issues about registrations and observations lifecycle. Merging the store allow to easily handle atomicity on registration/observations operations. (E.g. deregister will removed current observations)
more details : #192

This PR is focused on introducing the store without too many impact on current user API. (ClientRegistry and ObservationRegistry).

But this reveal some issues in the current API (e.g. `ClientRegistry` expose methods like `registerClient()` or `updateClient()` which is internal API). So the next step will probably to clean the API (with some renaming :-/ ).


Patch:
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/CaliforniumRegistrationStore.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/CaliforniumRegistrationStore.java
new file mode 100644
index 00000000..4054fa86
--- /dev/null
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/CaliforniumRegistrationStore.java
@@ -0,0 +1,27 @@
+/*******************************************************************************
+ * 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 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 org.eclipse.californium.core.observe.ObservationStore;
+import org.eclipse.leshan.server.registration.RegistrationStore;
+
+/**
+ * A registration store which is able to store Californium observation.
+ */
+public interface CaliforniumRegistrationStore extends RegistrationStore, ObservationStore
+{
+
+}
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServerBuilder.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServerBuilder.java
index 132f6e94..9b1be00a 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServerBuilder.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanServerBuilder.java
@@ -24,9 +24,8 @@ import org.eclipse.leshan.core.node.codec.LwM2mNodeDecoder;
 import org.eclipse.leshan.core.node.codec.LwM2mNodeEncoder;
 import org.eclipse.leshan.server.LwM2mServer;
 import org.eclipse.leshan.server.californium.impl.CaliforniumObservationRegistryImpl;
-import org.eclipse.leshan.server.californium.impl.InMemoryLwM2mObservationStore;
+import org.eclipse.leshan.server.californium.impl.InMemoryRegistrationStore;
 import org.eclipse.leshan.server.californium.impl.LeshanServer;
-import org.eclipse.leshan.server.client.ClientRegistry;
 import org.eclipse.leshan.server.impl.ClientRegistryImpl;
 import org.eclipse.leshan.server.impl.SecurityRegistryImpl;
 import org.eclipse.leshan.server.model.LwM2mModelProvider;
@@ -46,9 +45,8 @@ public class LeshanServerBuilder {
     /** IANA assigned UDP port for CoAP with DTLS (so for LWM2M) */
     public static final int PORT_DTLS = 5684;
 
+    private CaliforniumRegistrationStore registrationStore;
     private SecurityRegistry securityRegistry;
-    private CaliforniumObservationRegistry observationRegistry;
-    private ClientRegistry clientRegistry;
     private LwM2mModelProvider modelProvider;
     private InetSocketAddress localAddress;
     private InetSocketAddress localSecureAddress;
@@ -85,13 +83,8 @@ public class LeshanServerBuilder {
         return this;
     }
 
-    public LeshanServerBuilder setClientRegistry(ClientRegistry clientRegistry) {
-        this.clientRegistry = clientRegistry;
-        return this;
-    }
-
-    public LeshanServerBuilder setObservationRegistry(CaliforniumObservationRegistry observationRegistry) {
-        this.observationRegistry = observationRegistry;
+    public LeshanServerBuilder setRegistrationStore(CaliforniumRegistrationStore registrationStore) {
+        this.registrationStore = registrationStore;
         return this;
     }
 
@@ -120,8 +113,8 @@ public class LeshanServerBuilder {
             localAddress = new InetSocketAddress((InetAddress) null, PORT);
         if (localSecureAddress == null)
             localSecureAddress = new InetSocketAddress((InetAddress) null, PORT_DTLS);
-        if (clientRegistry == null)
-            clientRegistry = new ClientRegistryImpl();
+        if (registrationStore == null)
+            registrationStore = new InMemoryRegistrationStore();
         if (securityRegistry == null)
             securityRegistry = new SecurityRegistryImpl();
         if (modelProvider == null)
@@ -131,9 +124,10 @@ public class LeshanServerBuilder {
         if (decoder == null)
             decoder = new DefaultLwM2mNodeDecoder();
 
-        if (observationRegistry == null)
-            observationRegistry = new CaliforniumObservationRegistryImpl(new InMemoryLwM2mObservationStore(),
-                    clientRegistry, modelProvider, decoder);
+        ClientRegistryImpl clientRegistry = new ClientRegistryImpl(registrationStore);
+        
+        CaliforniumObservationRegistryImpl observationRegistry = new CaliforniumObservationRegistryImpl(
+                registrationStore, clientRegistry, modelProvider, decoder);
 
         return new LeshanServer(localAddress, localSecureAddress, clientRegistry, securityRegistry, observationRegistry,
                 modelProvider, encoder, decoder);
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 2c6443ed..4585c1c2 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
@@ -86,7 +86,7 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
         // Create the CoAP request from LwM2m request
         final CoapRequestBuilder coapRequestBuilder = new CoapRequestBuilder(
                 new InetSocketAddress(destination.getAddress(), destination.getPort()), destination.getRootPath(),
-                destination.getRegistrationId(), model, encoder);
+                destination.getRegistrationId(), destination.getEndpoint(), model, encoder);
         request.accept(coapRequestBuilder);
         final Request coapRequest = coapRequestBuilder.getRequest();
 
@@ -123,7 +123,7 @@ public class CaliforniumLwM2mRequestSender implements LwM2mRequestSender {
         // Create the CoAP request from LwM2m request
         final CoapRequestBuilder coapRequestBuilder = new CoapRequestBuilder(
                 new InetSocketAddress(destination.getAddress(), destination.getPort()), destination.getRootPath(),
-                destination.getRegistrationId(), model, encoder);
+                destination.getRegistrationId(), destination.getEndpoint(), model, encoder);
         request.accept(coapRequestBuilder);
         final Request coapRequest = coapRequestBuilder.getRequest();
 
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationRegistryImpl.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationRegistryImpl.java
index 18eb5d36..69c30c42 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationRegistryImpl.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationRegistryImpl.java
@@ -15,16 +15,13 @@
  *******************************************************************************/
 package org.eclipse.leshan.server.californium.impl;
 
-import static org.eclipse.leshan.server.californium.impl.CoapRequestBuilder.*;
+import static org.eclipse.leshan.server.californium.impl.CoapRequestBuilder.CTX_REGID;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -43,6 +40,7 @@ import org.eclipse.leshan.core.observation.Observation;
 import org.eclipse.leshan.core.request.ContentFormat;
 import org.eclipse.leshan.core.response.ObserveResponse;
 import org.eclipse.leshan.server.californium.CaliforniumObservationRegistry;
+import org.eclipse.leshan.server.californium.CaliforniumRegistrationStore;
 import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.client.ClientRegistry;
 import org.eclipse.leshan.server.model.LwM2mModelProvider;
@@ -61,7 +59,7 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
 
     private final Logger LOG = LoggerFactory.getLogger(CaliforniumObservationRegistry.class);
 
-    private final LwM2mObservationStore observationStore;
+    private final CaliforniumRegistrationStore registrationStore;
     private final ClientRegistry clientRegistry;
     private final LwM2mModelProvider modelProvider;
     private final LwM2mNodeDecoder decoder;
@@ -78,9 +76,9 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
      * @param modelProvider instance of {@link LwM2mModelProvider}
      * @param decoder instance of {@link LwM2mNodeDecoder}
      */
-    public CaliforniumObservationRegistryImpl(LwM2mObservationStore store, ClientRegistry clientRegistry,
+    public CaliforniumObservationRegistryImpl(CaliforniumRegistrationStore store, ClientRegistry clientRegistry,
             LwM2mModelProvider modelProvider, LwM2mNodeDecoder decoder) {
-        this.observationStore = store;
+        this.registrationStore = store;
         this.modelProvider = modelProvider;
         this.clientRegistry = clientRegistry;
         this.decoder = decoder;
@@ -120,13 +118,12 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
         if (registrationId == null)
             return 0;
 
-        Collection<org.eclipse.californium.core.observe.Observation> observations = observationStore
-                .removeAll(registrationId);
+        Collection<Observation> observations = registrationStore
+                .removeObservations(registrationId);
         if (observations == null)
             return 0;
 
-        for (org.eclipse.californium.core.observe.Observation cfObs : observations) {
-            Observation observation = build(cfObs);
+        for (Observation observation : observations) {
             if (secureEndpoint != null)
                 secureEndpoint.cancelObservation(observation.getId());
             if (nonSecureEndpoint != null)
@@ -161,7 +158,7 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
             secureEndpoint.cancelObservation(observation.getId());
         if (nonSecureEndpoint != null)
             nonSecureEndpoint.cancelObservation(observation.getId());
-        observationStore.remove(observation.getId());
+        registrationStore.removeObservation(observation.getRegistrationId(), observation.getId());
 
         for (ObservationRegistryListener listener : listeners) {
             listener.cancelled(observation);
@@ -177,12 +174,7 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
         if (registrationId == null)
             return Collections.emptySet();
 
-        Set<Observation> result = new HashSet<>();
-        for (org.eclipse.californium.core.observe.Observation obs : observationStore
-                .getByRegistrationId(registrationId)) {
-            result.add(build(obs));
-        }
-        return result;
+        return new HashSet<Observation>(registrationStore.getObservations(registrationId));
     }
 
     private Set<Observation> getObservations(String registrationId, String resourcePath) {
@@ -201,7 +193,7 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
 
     @Override
     public ObservationStore getObservationStore() {
-        return observationStore;
+        return registrationStore;
     }
 
     @Override
@@ -214,33 +206,6 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
         listeners.remove(listener);
     }
 
-    /* from a Californium observation */
-    private Observation build(org.eclipse.californium.core.observe.Observation cfObs) {
-        if (cfObs == null)
-            return null;
-
-        String regId = null;
-        String lwm2mPath = null;
-        Map<String, String> context = null;
-
-        for (Entry<String, String> ctx : cfObs.getRequest().getUserContext().entrySet()) {
-            switch (ctx.getKey()) {
-            case CTX_REGID:
-                regId = ctx.getValue();
-                break;
-            case CTX_LWM2M_PATH:
-                lwm2mPath = ctx.getValue();
-                break;
-            default:
-                if (context == null) {
-                    context = new HashMap<>();
-                }
-                context.put(ctx.getKey(), ctx.getValue());
-            }
-        }
-        return new Observation(cfObs.getRequest().getToken(), regId, new LwM2mPath(lwm2mPath), context);
-    }
-
     // ********** NotificationListener interface **********//
 
     @Override
@@ -253,8 +218,11 @@ public class CaliforniumObservationRegistryImpl implements CaliforniumObservatio
         if (coapResponse.getCode() == CoAP.ResponseCode.CHANGED
                 || coapResponse.getCode() == CoAP.ResponseCode.CONTENT) {
             try {
+                // get registration Id
+                String regid = coapRequest.getUserContext().get(CTX_REGID);
+                
                 // get observation for this request
-                Observation observation = build(observationStore.get(coapResponse.getToken()));
+                Observation observation = registrationStore.getObservation(regid, coapResponse.getToken());
                 if (observation == null)
                     return;
 
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilder.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilder.java
index 5ee78d0f..9e7e9c15 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilder.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilder.java
@@ -49,11 +49,13 @@ public class CoapRequestBuilder implements DownlinkRequestVisitor {
     private final InetSocketAddress destination;
     private final String rootPath;
     private final String registrationId;
+    private final String endpoint;
 
     private final LwM2mModel model;
     private final LwM2mNodeEncoder encoder;
 
     /* keys used to populate the request context */
+    public static final String CTX_ENDPOINT = "leshan-endpoint";
     public static final String CTX_REGID = "leshan-regId";
     public static final String CTX_LWM2M_PATH = "leshan-path";
 
@@ -61,14 +63,17 @@ public class CoapRequestBuilder implements DownlinkRequestVisitor {
         this.destination = destination;
         this.rootPath = null;
         this.registrationId = null;
+        this.endpoint = null;
         this.model = model;
         this.encoder = encoder;
     }
 
-    public CoapRequestBuilder(InetSocketAddress destination, String rootPath, String registrationId, LwM2mModel model,
+    public CoapRequestBuilder(InetSocketAddress destination, String rootPath, String registrationId, String endpoint,
+            LwM2mModel model,
             LwM2mNodeEncoder encoder) {
         this.destination = destination;
         this.rootPath = rootPath;
+        this.endpoint = endpoint;
         this.registrationId = registrationId;
         this.model = model;
         this.encoder = encoder;
@@ -141,6 +146,7 @@ public class CoapRequestBuilder implements DownlinkRequestVisitor {
 
         // add context info to the observe request
         Map<String, String> context = new HashMap<>();
+        context.put(CTX_ENDPOINT, endpoint);
         context.put(CTX_REGID, registrationId);
         context.put(CTX_LWM2M_PATH, request.getPath().toString());
         for (Entry<String, String> ctx : request.getContext().entrySet()) {
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/InMemoryRegistrationStore.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/InMemoryRegistrationStore.java
new file mode 100644
index 00000000..abdcd0b5
--- /dev/null
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/InMemoryRegistrationStore.java
@@ -0,0 +1,446 @@
+/*******************************************************************************
+ * 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 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.impl;
+
+import static org.eclipse.leshan.server.californium.impl.CoapRequestBuilder.*;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.eclipse.californium.core.coap.Request;
+import org.eclipse.californium.core.network.Exchange.KeyToken;
+import org.eclipse.californium.core.network.serialization.DataParser;
+import org.eclipse.californium.core.network.serialization.DataSerializer;
+import org.eclipse.californium.core.network.serialization.UdpDataParser;
+import org.eclipse.californium.core.network.serialization.UdpDataSerializer;
+import org.eclipse.californium.elements.CorrelationContext;
+import org.eclipse.californium.elements.RawData;
+import org.eclipse.leshan.core.node.LwM2mPath;
+import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.Startable;
+import org.eclipse.leshan.server.Stoppable;
+import org.eclipse.leshan.server.californium.CaliforniumRegistrationStore;
+import org.eclipse.leshan.server.client.Client;
+import org.eclipse.leshan.server.client.ClientUpdate;
+import org.eclipse.leshan.server.registration.Deregistration;
+import org.eclipse.leshan.server.registration.ExpirationListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An in memory store for registration and observation.
+ */
+public class InMemoryRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable {
+    private final Logger LOG = LoggerFactory.getLogger(InMemoryRegistrationStore.class);
+
+    // Data structure
+    private final Map<String /* end-point */, Client> regsByEp = new HashMap<>();
+    private Map<KeyToken, org.eclipse.californium.core.observe.Observation> obsByToken = new HashMap<>();
+    private Map<String, List<KeyToken>> tokensByRegId = new HashMap<>();
+
+    // Request serializer to persist observation
+    private static final DataSerializer serializer = new UdpDataSerializer();
+
+    private final ReadWriteLock lock = new ReentrantReadWriteLock();
+    
+    // Listener use to notify when a registration expires
+    private ExpirationListener expirationListener;
+
+    /* *************** Leshan Registration API **************** */
+
+    @Override
+    public Deregistration addRegistration(Client registration) {
+        try {
+            lock.writeLock().lock();
+
+            Client registrationRemoved = regsByEp.put(registration.getEndpoint(), registration);
+            if (registrationRemoved != null) {
+                Collection<Observation> observationsRemoved = unsafeRemoveAllObservations(registrationRemoved.getRegistrationId());
+                return new Deregistration(registrationRemoved, observationsRemoved);
+            }
+        } finally {
+            lock.writeLock().unlock();
+        }
+        return null;
+    }
+
+    @Override
+    public Client updateRegistration(ClientUpdate update) {
+        try {
+            lock.writeLock().lock();
+
+            Client registration = getRegistration(update.getRegistrationId());
+            if (registration == null) {
+                return null;
+            } else {
+                Client registrationUpdated = update.updateClient(registration);
+                regsByEp.put(registrationUpdated.getEndpoint(), registrationUpdated);
+                return registrationUpdated;
+            }
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    @Override
+    public Client getRegistration(String registrationId) {
+        try {
+            lock.readLock().lock();
+
+            if (registrationId != null) {
+                for (Client registration : regsByEp.values()) {
+                    if (registrationId.equals(registration.getRegistrationId())) {
+                        return registration;
+                    }
+                }
+            }
+            return null;
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public Client getRegistrationByEndpoint(String endpoint) {
+        try {
+            lock.readLock().lock();
+            return regsByEp.get(endpoint);
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public Collection<Client> getRegistrationByAdress(InetSocketAddress address) {
+        // TODO needed to remove getAllRegistration()
+        return null;
+    }
+
+    @Override
+    public Collection<Client> getAllRegistration() {
+        try {
+            lock.readLock().lock();
+            return Collections.unmodifiableCollection(regsByEp.values());
+        } finally {
+            lock.readLock().unlock();
+        }
+
+    }
+
+    @Override
+    public Deregistration removeRegistration(String registrationId) {
+        try {
+            lock.writeLock().lock();
+
+            Client registration = getRegistration(registrationId);
+            if (registration != null) {
+                Collection<Observation> observationsRemoved = unsafeRemoveAllObservations(registration.getRegistrationId());
+                regsByEp.remove(registration.getEndpoint());
+                return new Deregistration(registration, observationsRemoved);
+            }
+            return null;
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    /* *************** Leshan Observation API **************** */
+
+    @Override
+    public Observation addObservation(String registrationId, Observation observation) {
+        // not used observation was added by Californium
+        return null;
+    }
+
+    @Override
+    public Observation removeObservation(String registrationId, byte[] observationId) {
+        try {
+            lock.writeLock().lock();
+
+            Observation observation = build(unsafeGetObservation(new KeyToken(observationId)));
+            if (observation != null && registrationId.equals(observation.getRegistrationId())) {
+                // TODO remove API should returns the observation removed
+                unsafeRemoveObservation(observationId);
+                return observation;
+            }
+            return null;
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    @Override
+    public Observation getObservation(String registrationId, byte[] observationId) {
+        try {
+            lock.readLock().lock();
+            Observation observation = build(unsafeGetObservation(new KeyToken(observationId)));
+            if (observation != null && registrationId.equals(observation.getRegistrationId())) {
+                return observation;
+            }
+            return null;
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public Collection<Observation> getObservations(String registrationId) {
+        try {
+            lock.readLock().lock();
+
+            Collection<Observation> result = new ArrayList<>();
+            List<KeyToken> tokens = tokensByRegId.get(registrationId);
+            if (tokens != null) {
+                for (KeyToken token : tokens) {
+                    Observation obs = build(unsafeGetObservation(token));
+                    if (obs != null) {
+                        result.add(obs);
+                    }
+                }
+            }
+            return result;
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public Collection<Observation> removeObservations(String registrationId) {
+        try {
+            lock.writeLock().lock();
+            return unsafeRemoveAllObservations(registrationId);
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    /* *************** Californium ObservationStore API **************** */
+
+    @Override
+    public void add(org.eclipse.californium.core.observe.Observation obs) {
+        if (obs != null) {
+            try {
+                lock.writeLock().lock();
+
+                validateObservation(obs);
+
+                String registrationId = extractRegistrationId(obs);
+                KeyToken token = new KeyToken(obs.getRequest().getToken());
+                org.eclipse.californium.core.observe.Observation previousObservation = obsByToken.put(token, obs);
+                if (!tokensByRegId.containsKey(registrationId)) {
+                    tokensByRegId.put(registrationId, new ArrayList<KeyToken>());
+                }
+                tokensByRegId.get(registrationId).add(token);
+
+                // log any collisions
+                if (previousObservation != null) {
+                    LOG.warn(
+                            "Token collision ? observation from request [{}] will be replaced by observation from request [{}] ",
+                            previousObservation.getRequest(), obs.getRequest());
+                }
+            } finally {
+                lock.writeLock().unlock();
+            }
+        }
+    }
+
+    @Override
+    public org.eclipse.californium.core.observe.Observation get(byte[] token) {
+        try {
+            lock.readLock().lock();
+            return unsafeGetObservation(new KeyToken(token));
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public void setContext(byte[] token, CorrelationContext ctx) {
+        try {
+            lock.writeLock().lock();
+
+            org.eclipse.californium.core.observe.Observation obs = obsByToken.get(new KeyToken(token));
+            if (obs != null) {
+                obsByToken.put(new KeyToken(token),
+                        new org.eclipse.californium.core.observe.Observation(obs.getRequest(), ctx));
+            }
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    @Override
+    public void remove(byte[] token) {
+        try {
+            lock.writeLock().lock();
+            unsafeRemoveObservation(token);
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    /* *************** Observation utility functions **************** */
+
+    private org.eclipse.californium.core.observe.Observation unsafeGetObservation(KeyToken token) {
+        org.eclipse.californium.core.observe.Observation obs = obsByToken.get(token);
+            if (obs != null) {
+                RawData serialize = serializer.serializeRequest(obs.getRequest(), null);
+                DataParser parser = new UdpDataParser();
+                Request newRequest = (Request) parser.parseMessage(serialize);
+                newRequest.setUserContext(obs.getRequest().getUserContext());
+                return new org.eclipse.californium.core.observe.Observation(newRequest, obs.getContext());
+            }
+            return null;
+    }
+
+    private void unsafeRemoveObservation(byte[] observationId) {
+        KeyToken kToken = new KeyToken(observationId);
+        org.eclipse.californium.core.observe.Observation removed = obsByToken.remove(kToken);
+
+        if (removed != null) {
+            String registrationId = extractRegistrationId(removed);
+            List<KeyToken> tokens = tokensByRegId.get(registrationId);
+            tokens.remove(kToken);
+            if (tokens.isEmpty()) {
+                tokensByRegId.remove(registrationId);
+            }
+        }
+    }
+
+    private Collection<Observation> unsafeRemoveAllObservations(String registrationId) {
+        Collection<Observation> removed = new ArrayList<>();
+        List<KeyToken> tokens = tokensByRegId.get(registrationId);
+        if (tokens != null) {
+            for (KeyToken token : tokens) {
+                Observation observationRemoved = build(obsByToken.remove(token));
+                if (observationRemoved != null) {
+                    removed.add(observationRemoved);
+                }
+            }
+        }
+        tokensByRegId.remove(registrationId);
+        return removed;
+    }
+
+    /* Retrieve the registrationId from the request context */
+    private String extractRegistrationId(org.eclipse.californium.core.observe.Observation observation) {
+        return observation.getRequest().getUserContext().get(CoapRequestBuilder.CTX_REGID);
+    }
+
+    private Observation build(org.eclipse.californium.core.observe.Observation cfObs) {
+        if (cfObs == null)
+            return null;
+
+        String regId = null;
+        String lwm2mPath = null;
+        Map<String, String> context = null;
+
+        for (Entry<String, String> ctx : cfObs.getRequest().getUserContext().entrySet()) {
+            switch (ctx.getKey()) {
+            case CTX_REGID:
+                regId = ctx.getValue();
+                break;
+            case CTX_LWM2M_PATH:
+                lwm2mPath = ctx.getValue();
+                break;
+            default:
+                if (context == null) {
+                    context = new HashMap<>();
+                }
+                context.put(ctx.getKey(), ctx.getValue());
+            }
+        }
+        return new Observation(cfObs.getRequest().getToken(), regId, new LwM2mPath(lwm2mPath), context);
+    }
+
+    private void validateObservation(org.eclipse.californium.core.observe.Observation observation) {
+        if (!observation.getRequest().getUserContext().containsKey(CoapRequestBuilder.CTX_REGID))
+            throw new IllegalStateException("missing registrationId info in the request context");
+        if (!observation.getRequest().getUserContext().containsKey(CoapRequestBuilder.CTX_LWM2M_PATH))
+            throw new IllegalStateException("missing lwm2m path info in the request context");
+        if (getRegistration(observation.getRequest().getUserContext().get(CoapRequestBuilder.CTX_REGID)) == null) {
+            throw new IllegalStateException("no registration for this Id");
+        }
+    }
+
+    /* *************** Expiration handling **************** */
+
+    @Override
+    public void setExpirationListener(ExpirationListener listener) {
+        this.expirationListener = listener;
+    }
+
+    /**
+     * start the registration store, will start regular cleanup of dead registrations.
+     */
+    @Override
+    public void start() {
+        // every 2 seconds clean the registration list
+        // TODO re-consider clean-up interval: wouldn't 5 minutes do as well?
+        schedExecutor.scheduleAtFixedRate(new Cleaner(), 2, 2, TimeUnit.SECONDS);
+    }
+
+    /**
+     * Stop the underlying cleanup of the registrations.
+     */
+    @Override
+    public void stop() {
+        schedExecutor.shutdownNow();
+        try {
+            schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            LOG.warn("Clean up registration thread was interrupted.", e);
+        }
+    }
+
+    private final ScheduledExecutorService schedExecutor = Executors.newScheduledThreadPool(1);
+
+    private class Cleaner implements Runnable {
+
+        @Override
+        public void run() {
+            try {
+                lock.readLock().lock();
+
+                for (Client client : getAllRegistration()) {
+                    synchronized (client) {
+                        if (!client.isAlive()) {
+                            // force de-registration
+                            Deregistration removedRegistration = removeRegistration(client.getRegistrationId());
+                            expirationListener.registrationExpired(removedRegistration.getRegistration(),
+                                    removedRegistration.getObservations());
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                LOG.warn("Unexcepted Exception while registration cleaning", e);
+            } finally {
+                lock.readLock().unlock();
+            }
+        }
+    }
+}
diff --git a/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/LeshanClusterServer.java b/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/LeshanClusterServer.java
index 8b9ea716..beec6efd 100644
--- a/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/LeshanClusterServer.java
+++ b/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/LeshanClusterServer.java
@@ -24,9 +24,7 @@ import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
 import org.eclipse.leshan.core.node.codec.DefaultLwM2mNodeDecoder;
 import org.eclipse.leshan.server.californium.LeshanServerBuilder;
-import org.eclipse.leshan.server.californium.impl.CaliforniumObservationRegistryImpl;
 import org.eclipse.leshan.server.californium.impl.LeshanServer;
-import org.eclipse.leshan.server.client.ClientRegistry;
 import org.eclipse.leshan.server.model.LwM2mModelProvider;
 import org.eclipse.leshan.server.model.StandardModelProvider;
 import org.slf4j.Logger;
@@ -155,15 +153,12 @@ public class LeshanClusterServer {
         LwM2mModelProvider modelProvider = new StandardModelProvider();
         builder.setObjectModelProvider(modelProvider);
 
-        ClientRegistry clientRegistry = new RedisClientRegistry(jedis);
-        builder.setClientRegistry(clientRegistry);
+        RedisRegistrationStore registrationStore = new RedisRegistrationStore(jedis);
+        builder.setRegistrationStore(registrationStore);
 
         // TODO add support of public and private server key
         builder.setSecurityRegistry(new RedisSecurityRegistry(jedis, null, null));
 
-        builder.setObservationRegistry(new CaliforniumObservationRegistryImpl(new RedisObservationStore(jedis),
-                clientRegistry, modelProvider, decoder));
-
         // Create and start LWM2M server
         LeshanServer lwServer = builder.build();
 
@@ -171,8 +166,8 @@ public class LeshanClusterServer {
         RedisTokenHandler tokenHandler = new RedisTokenHandler(jedis, clusterInstanceId);
         new RedisRequestResponseHandler(jedis, lwServer, lwServer.getClientRegistry(), tokenHandler,
                 lwServer.getObservationRegistry());
-        clientRegistry.addListener(tokenHandler);
-        clientRegistry.addListener(new RedisRegistrationEventPublisher(jedis));
+        lwServer.getClientRegistry().addListener(tokenHandler);
+        lwServer.getClientRegistry().addListener(new RedisRegistrationEventPublisher(jedis));
 
         // Start Jetty & Leshan
         lwServer.start();
diff --git a/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/RedisRegistrationStore.java b/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/RedisRegistrationStore.java
new file mode 100644
index 00000000..e4d39ed2
--- /dev/null
+++ b/leshan-server-cluster/src/main/java/org/eclipse/leshan/server/cluster/RedisRegistrationStore.java
@@ -0,0 +1,577 @@
+/*******************************************************************************
+ * 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 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.cluster;
+
+import static org.eclipse.leshan.server.californium.impl.CoapRequestBuilder.*;
+import static org.eclipse.leshan.util.Charsets.UTF_8;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.californium.elements.CorrelationContext;
+import org.eclipse.leshan.core.node.LwM2mPath;
+import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.Startable;
+import org.eclipse.leshan.server.Stoppable;
+import org.eclipse.leshan.server.californium.CaliforniumRegistrationStore;
+import org.eclipse.leshan.server.californium.impl.CoapRequestBuilder;
+import org.eclipse.leshan.server.client.Client;
+import org.eclipse.leshan.server.client.ClientUpdate;
+import org.eclipse.leshan.server.cluster.serialization.ClientSerDes;
+import org.eclipse.leshan.server.cluster.serialization.ObservationSerDes;
+import org.eclipse.leshan.server.registration.Deregistration;
+import org.eclipse.leshan.server.registration.ExpirationListener;
+import org.eclipse.leshan.util.Validate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.ScanParams;
+import redis.clients.jedis.ScanResult;
+import redis.clients.util.Pool;
+
+/**
+ * A RegistrationStore which stores registrations and observations in Redis.
+ */
+public class RedisRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(RedisClientRegistry.class);
+
+    // Redis key prefixes
+    private static final String EP_REG = "EP#REG#";
+    private static final String REGID_EP = "REGID#EP#";
+    private static final String LOCK_EP = "LOCK#EP#";
+    private static final byte[] OBS_TKN = "OBS#TKN#".getBytes(UTF_8);
+    private static final String OBS_REGID = "OBS#REGID#";
+
+    private final Pool<Jedis> pool;
+
+    // Listener use to notify when a registration expires
+    private ExpirationListener expirationListener;
+
+    public RedisRegistrationStore(Pool<Jedis> p) {
+        this.pool = p;
+    }
+
+    /* *************** Redis Key utility function **************** */
+
+    private byte[] toKey(byte[] prefix, byte[] key) {
+        byte[] result = new byte[prefix.length + key.length];
+        System.arraycopy(prefix, 0, result, 0, prefix.length);
+        System.arraycopy(key, 0, result, prefix.length, key.length);
+        return result;
+    }
+
+    private byte[] toKey(String prefix, String registrationID) {
+        return (prefix + registrationID).getBytes();
+    }
+
+    private byte[] toLockKey(String endpoint) {
+        return toKey(LOCK_EP, endpoint);
+    }
+
+    private byte[] toLockKey(byte[] endpoint) {
+        return toKey(LOCK_EP.getBytes(UTF_8), endpoint);
+    }
+
+    /* *************** Leshan Registration API **************** */
+
+    @Override
+    public Deregistration addRegistration(Client registration) {
+        try (Jedis j = pool.getResource()) {
+            byte[] lockValue = null;
+            byte[] lockKey = toLockKey(registration.getEndpoint());
+
+            try {
+                lockValue = RedisLock.acquire(j, lockKey);
+
+                // add registration
+                byte[] k = toEndpointKey(registration.getEndpoint());
+                byte[] old = j.getSet(k, serializeReg(registration));
+
+                // add registration: secondary index
+                byte[] idx = toRegIdKey(registration.getRegistrationId());
+                j.set(idx, registration.getEndpoint().getBytes(UTF_8));
+
+                if (old != null) {
+                    Client oldRegistration = deserializeReg(old);
+                    Collection<Observation> obsRemoved = unsafeRemoveAllObservations(j,
+                            oldRegistration.getRegistrationId());
+                    return new Deregistration(oldRegistration, obsRemoved);
+                }
+
+                return null;
+            } finally {
+                RedisLock.release(j, lockKey, lockValue);
+            }
+        }
+    }
+
+    @Override
+    public Client updateRegistration(ClientUpdate update) {
+        try (Jedis j = pool.getResource()) {
+
+            // fetch the client ep by registration ID index
+            byte[] ep = j.get(toRegIdKey(update.getRegistrationId()));
+            if (ep == null) {
+                return null;
+            }
+
+            // fetch the client
+            byte[] data = j.get(toEndpointKey(ep));
+            if (data == null) {
+                return null;
+            }
+
+            Client r = deserializeReg(data);
+
+            byte[] lockValue = null;
+            byte[] lockKey = toLockKey(r.getEndpoint());
+            try {
+                lockValue = RedisLock.acquire(j, lockKey);
+
+                Client clientUpdated = update.updateClient(r);
+
+                // store the new client
+                j.set(toEndpointKey(clientUpdated.getEndpoint()), serializeReg(clientUpdated));
+
+                return clientUpdated;
+
+            } finally {
+                RedisLock.release(j, lockKey, lockValue);
+            }
+        }
+    }
+
+    @Override
+    public Client getRegistration(String registrationId) {
+        try (Jedis j = pool.getResource()) {
+            byte[] ep = j.get(toRegIdKey(registrationId));
+            if (ep == null) {
+                return null;
+            }
+            byte[] data = j.get(toEndpointKey(ep));
+            if (data == null) {
+                return null;
+            }
+
+            return deserializeReg(data);
+        }
+    }
+
+    @Override
+    public Client getRegistrationByEndpoint(String endpoint) {
+        Validate.notNull(endpoint);
+        try (Jedis j = pool.getResource()) {
+            byte[] data = j.get(toEndpointKey(endpoint));
+            if (data == null) {
+                return null;
+            }
+            Client r = deserializeReg(data);
+            return r.isAlive() ? r : null;
+        }
+    }
+
+    @Override
+    public Collection<Client> getRegistrationByAdress(InetSocketAddress address) {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    @Override
+    public Collection<Client> getAllRegistration() {
+        try (Jedis j = pool.getResource()) {
+            ScanParams params = new ScanParams().match(EP_REG + "*").count(100);
+            Collection<Client> list = new LinkedList<>();
+            String cursor = "0";
+            do {
+                ScanResult<byte[]> res = j.scan(cursor.getBytes(), params);
+                for (byte[] key : res.getResult()) {
+                    byte[] element = j.get(key);
+                    if (element != null) {
+                        Client c = deserializeReg(element);
+                        if (c.isAlive()) {
+                            list.add(c);
+                        }
+                    }
+                }
+                cursor = res.getStringCursor();
+            } while (!"0".equals(cursor));
+            return list;
+        }
+    }
+
+    @Override
+    public Deregistration removeRegistration(String registrationId) {
+        try (Jedis j = pool.getResource()) {
+
+            byte[] regKey = toRegIdKey(registrationId);
+
+            // fetch the client ep by registration ID index
+            byte[] ep = j.get(regKey);
+            if (ep == null) {
+                return null;
+            }
+
+            byte[] data = j.get(toEndpointKey(ep));
+            if (data == null) {
+                return null;
+            }
+
+            Client r = deserializeReg(data);
+            deleteClient(j, r);
+            Collection<Observation> obsRemoved = unsafeRemoveAllObservations(j, r.getRegistrationId());
+            return new Deregistration(r, obsRemoved);
+        }
+    }
+
+    private void deleteClient(Jedis j, Client c) {
+        byte[] lockValue = null;
+        byte[] lockKey = toLockKey(c.getEndpoint());
+        try {
+            lockValue = RedisLock.acquire(j, lockKey);
+
+            // delete all entries
+            j.del(toRegIdKey(c.getRegistrationId()));
+            j.del(toEndpointKey(c.getEndpoint()));
+
+        } finally {
+            RedisLock.release(j, lockKey, lockValue);
+        }
+    }
+
+    private byte[] toRegIdKey(String registrationId) {
+        return toKey(REGID_EP, registrationId);
+    }
+
+    private byte[] toEndpointKey(String endpoint) {
+        return toKey(EP_REG, endpoint);
+    }
+
+    private byte[] toEndpointKey(byte[] endpoint) {
+        return toKey(EP_REG.getBytes(UTF_8), endpoint);
+    }
+
+    private byte[] serializeReg(Client client) {
+        return ClientSerDes.bSerialize(client);
+    }
+
+    private Client deserializeReg(byte[] data) {
+        return ClientSerDes.deserialize(data);
+    }
+
+    /* *************** Leshan Observation API **************** */
+
+    @Override
+    public Observation addObservation(String registrationId, Observation observation) {
+        // not used observation was added by Californium
+        return null;
+    }
+
+    @Override
+    public Observation removeObservation(String registrationId, byte[] observationId) {
+        try (Jedis j = pool.getResource()) {
+
+            // fetch the client ep by registration ID index
+            byte[] ep = j.get(toRegIdKey(registrationId));
+            if (ep == null) {
+                return null;
+            }
+
+            // remove observation
+            byte[] lockValue = null;
+            byte[] lockKey = toLockKey(ep);
+            try {
+                lockValue = RedisLock.acquire(j, lockKey);
+
+                Observation observation = build(get(observationId));
+                if (observation != null && registrationId.equals(observation.getRegistrationId())) {
+                    unsafeRemoveObservation(j, registrationId, observationId);
+                    return observation;
+                }
+                return null;
+
+            } finally {
+                RedisLock.release(j, lockKey, lockValue);
+            }
+        }
+    }
+
+    @Override
+    public Observation getObservation(String registrationId, byte[] ObservationId) {
+        return build(get(ObservationId));
+    }
+
+    @Override
+    public Collection<Observation> getObservations(String registrationId) {
+        Collection<Observation> result = new ArrayList<>();
+        try (Jedis j = pool.getResource()) {
+            for (byte[] token : j.lrange(toKey(OBS_REGID, registrationId), 0, -1)) {
+                byte[] obs = j.get(toKey(OBS_TKN, token));
+                if (obs != null) {
+                    result.add(build(deserializeObs(obs)));
+                }
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public Collection<Observation> removeObservations(String registrationId) {
+        try (Jedis j = pool.getResource()) {
+            // check registration exists
+            Client registration = getRegistration(registrationId);
+            if (registration == null)
+                return Collections.emptyList();
+
+            // get endpoint and create lock
+            String endpoint = registration.getEndpoint();
+            byte[] lockValue = null;
+            byte[] lockKey = toKey(LOCK_EP, endpoint);
+            try {
+                lockValue = RedisLock.acquire(j, lockKey);
+
+                return unsafeRemoveAllObservations(j, registrationId);
+            } finally {
+                RedisLock.release(j, lockKey, lockValue);
+            }
+        }
+    }
+
+    /* *************** Californium ObservationStore API **************** */
+
+    @Override
+    public void add(org.eclipse.californium.core.observe.Observation obs) {
+        String endpoint = this.validateObservation(obs);
+
+        try (Jedis j = pool.getResource()) {
+            byte[] lockValue = null;
+            byte[] lockKey = toKey(LOCK_EP, endpoint);
+            try {
+                lockValue = RedisLock.acquire(j, lockKey);
+
+                String registrationId = obs.getRequest().getUserContext().get(CTX_REGID);
+                if (!j.exists(toRegIdKey(registrationId)))
+                    throw new IllegalStateException("no registration for this Id");
+
+                byte[] previousValue = j.getSet(toKey(OBS_TKN, obs.getRequest().getToken()), serializeObs(obs));
+
+                // secondary index to get the list by registrationId
+                j.lpush(toKey(OBS_REGID, registrationId), obs.getRequest().getToken());
+
+                // log any collisions
+                if (previousValue != null && previousValue.length != 0) {
+                    org.eclipse.californium.core.observe.Observation previousObservation = deserializeObs(
+                            previousValue);
+                    LOG.warn(
+                            "Token collision ? observation from request [{}] will be replaced by observation from request [{}] ",
+                            previousObservation.getRequest(), obs.getRequest());
+                }
+            } finally {
+                RedisLock.release(j, lockKey, lockValue);
+            }
+        }
+    }
+
+    @Override
+    public void remove(byte[] token) {
+        try (Jedis j = pool.getResource()) {
+            byte[] tokenKey = toKey(OBS_TKN, token);
+
+            // fetch the observation by token
+            byte[] serializedObs = j.get(tokenKey);
+            if (serializedObs == null)
+                return;
+
+            org.eclipse.californium.core.observe.Observation obs = deserializeObs(serializedObs);
+            String registrationId = extractRegistrationId(obs);
+            Client registration = getRegistration(registrationId);
+            String endpoint = registration.getEndpoint();
+
+            byte[] lockValue = null;
+            byte[] lockKey = toKey(LOCK_EP, endpoint);
+            try {
+                lockValue = RedisLock.acquire(j, lockKey);
+
+                unsafeRemoveObservation(j, registrationId, token);
+            } finally {
+                RedisLock.release(j, lockKey, lockValue);
+            }
+        }
+
+    }
+
+    @Override
+    public org.eclipse.californium.core.observe.Observation get(byte[] token) {
+        try (Jedis j = pool.getResource()) {
+            byte[] obs = j.get(toKey(OBS_TKN, token));
+            if (obs == null) {
+                return null;
+            } else {
+                return deserializeObs(obs);
+            }
+        }
+    }
+
+    /* *************** Observation utility functions **************** */
+
+    private void unsafeRemoveObservation(Jedis j, String registrationId,
+            byte[] observationId) {
+        if (j.del(observationId) > 0L) {
+            j.lrem(toKey(OBS_REGID, registrationId), 0, observationId);
+        }
+    }
+
+    private Collection<Observation> unsafeRemoveAllObservations(Jedis j, String registrationId) {
+        Collection<Observation> removed = new ArrayList<>();
+        byte[] regIdKey = toKey(OBS_REGID, registrationId);
+
+        // fetch all observations by token
+        for (byte[] token : j.lrange(regIdKey, 0, -1)) {
+            byte[] obs = j.get(toKey(OBS_TKN, token));
+            if (obs != null) {
+                removed.add(build(deserializeObs(obs)));
+                }
+            j.del(toKey(OBS_TKN, token));
+        }
+        j.del(regIdKey);
+
+        return removed;
+    }
+
+    @Override
+    public void setContext(byte[] token, CorrelationContext correlationContext) {
+        // TODO should be implemented
+    }
+
+    private byte[] serializeObs(org.eclipse.californium.core.observe.Observation obs) {
+        return ObservationSerDes.serialize(obs);
+    }
+
+    private org.eclipse.californium.core.observe.Observation deserializeObs(byte[] data) {
+        return ObservationSerDes.deserialize(data);
+    }
+
+    /* Retrieve the registrationId from the request context */
+    private String extractRegistrationId(org.eclipse.californium.core.observe.Observation observation) {
+        return observation.getRequest().getUserContext().get(CoapRequestBuilder.CTX_REGID);
+    }
+
+    private Observation build(org.eclipse.californium.core.observe.Observation cfObs) {
+        if (cfObs == null)
+            return null;
+
+        String regId = null;
+        String lwm2mPath = null;
+        Map<String, String> context = null;
+
+        for (Entry<String, String> ctx : cfObs.getRequest().getUserContext().entrySet()) {
+            switch (ctx.getKey()) {
+            case CTX_REGID:
+                regId = ctx.getValue();
+                break;
+            case CTX_LWM2M_PATH:
+                lwm2mPath = ctx.getValue();
+                break;
+            default:
+                if (context == null) {
+                    context = new HashMap<>();
+                }
+                context.put(ctx.getKey(), ctx.getValue());
+            }
+        }
+        return new Observation(cfObs.getRequest().getToken(), regId, new LwM2mPath(lwm2mPath), context);
+    }
+
+    private String validateObservation(org.eclipse.californium.core.observe.Observation observation) {
+        if (!observation.getRequest().getUserContext().containsKey(CoapRequestBuilder.CTX_REGID))
+            throw new IllegalStateException("missing registrationId info in the request context");
+        if (!observation.getRequest().getUserContext().containsKey(CoapRequestBuilder.CTX_LWM2M_PATH))
+            throw new IllegalStateException("missing lwm2m path info in the request context");
+
+        String endpoint = observation.getRequest().getUserContext().get(CoapRequestBuilder.CTX_ENDPOINT);
+        if (endpoint == null)
+            throw new IllegalStateException("missing endpoint info in the request context");
+
+        return endpoint;
+    }
+
+    /* *************** Expiration handling **************** */
+
+    /**
+     * Start regular cleanup of dead registrations.
+     */
+    @Override
+    public void start() {
+        // clean the registration list every minute
+        schedExecutor.scheduleAtFixedRate(new Cleaner(), 1, 1, TimeUnit.MINUTES);
+    }
+
+    /**
+     * Stop the underlying cleanup of the registrations.
+     */
+    @Override
+    public void stop() {
+        schedExecutor.shutdownNow();
+        try {
+            schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            LOG.warn("Clean up registration thread was interrupted.", e);
+        }
+    }
+
+    private final ScheduledExecutorService schedExecutor = Executors.newScheduledThreadPool(1);
+
+    private class Cleaner implements Runnable {
+
+        @Override
+        public void run() {
+
+            try (Jedis j = pool.getResource()) {
+                ScanParams params = new ScanParams().match(EP_REG + "*").count(100);
+                String cursor = "0";
+                do {
+                    // TODO we probably need a lock here
+                    ScanResult<byte[]> res = j.scan(cursor.getBytes(), params);
+                    for (byte[] key : res.getResult()) {
+                        Client c = deserializeReg(j.get(key));
+                        if (!c.isAlive()) {
+                            deleteClient(j, c);
+                            expirationListener.registrationExpired(c, new ArrayList<Observation>());
+                        }
+                    }
+                    cursor = res.getStringCursor();
+                } while (!"0".equals(cursor));
+            } catch (Exception e) {
+                LOG.warn("Unexcepted Exception while registration cleaning", e);
+            }
+        }
+    }
+
+    @Override
+    public void setExpirationListener(ExpirationListener listener) {
+        expirationListener = listener;
+    }
+}
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/impl/ClientRegistryImpl.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/impl/ClientRegistryImpl.java
index 7f585f6a..8b932fed 100644
--- a/leshan-server-core/src/main/java/org/eclipse/leshan/server/impl/ClientRegistryImpl.java
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/impl/ClientRegistryImpl.java
@@ -18,19 +18,18 @@ package org.eclipse.leshan.server.impl;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 
+import org.eclipse.leshan.core.observation.Observation;
 import org.eclipse.leshan.server.Startable;
 import org.eclipse.leshan.server.Stoppable;
 import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.client.ClientRegistry;
 import org.eclipse.leshan.server.client.ClientRegistryListener;
 import org.eclipse.leshan.server.client.ClientUpdate;
+import org.eclipse.leshan.server.registration.Deregistration;
+import org.eclipse.leshan.server.registration.ExpirationListener;
+import org.eclipse.leshan.server.registration.RegistrationStore;
 import org.eclipse.leshan.util.Validate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,14 +37,19 @@ import org.slf4j.LoggerFactory;
 /**
  * In memory client registry
  */
-public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable {
+public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable, ExpirationListener {
 
     private static final Logger LOG = LoggerFactory.getLogger(ClientRegistryImpl.class);
 
-    private final Map<String /* end-point */, Client> clientsByEp = new ConcurrentHashMap<>();
-
     private final List<ClientRegistryListener> listeners = new CopyOnWriteArrayList<>();
 
+    private RegistrationStore store;
+
+    public ClientRegistryImpl(RegistrationStore store) {
+        this.store = store;
+        store.setExpirationListener(this);
+    }
+
     @Override
     public void addListener(ClientRegistryListener listener) {
         listeners.add(listener);
@@ -58,12 +62,12 @@ public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable
 
     @Override
     public Collection<Client> allClients() {
-        return Collections.unmodifiableCollection(clientsByEp.values());
+        return Collections.unmodifiableCollection(store.getAllRegistration());
     }
 
     @Override
     public Client get(String endpoint) {
-        return clientsByEp.get(endpoint);
+        return store.getRegistrationByEndpoint(endpoint);
     }
 
     @Override
@@ -72,10 +76,10 @@ public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable
 
         LOG.debug("Registering new client: {}", client);
 
-        Client previous = clientsByEp.put(client.getEndpoint(), client);
+        Deregistration previous = store.addRegistration(client);
         if (previous != null) {
             for (ClientRegistryListener l : listeners) {
-                l.unregistered(previous);
+                l.unregistered(previous.getRegistration());
             }
         }
         for (ClientRegistryListener l : listeners) {
@@ -90,19 +94,15 @@ public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable
         Validate.notNull(update);
 
         LOG.debug("Updating registration for client: {}", update);
-        Client client = findByRegistrationId(update.getRegistrationId());
-        if (client == null) {
-            return null;
-        } else {
-            Client clientUpdated = update.updateClient(client);
-            clientsByEp.put(clientUpdated.getEndpoint(), clientUpdated);
-
+        Client clientUpdated = store.updateRegistration(update);
+        if (clientUpdated != null) {
             // notify listener
             for (ClientRegistryListener l : listeners) {
                 l.updated(update, clientUpdated);
             }
             return clientUpdated;
         }
+        return null;
     }
 
     @Override
@@ -111,31 +111,17 @@ public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable
 
         LOG.debug("Deregistering client with registrationId: {}", registrationId);
 
-        Client toBeUnregistered = findByRegistrationId(registrationId);
-        if (toBeUnregistered == null) {
-            return null;
-        } else {
-            Client unregistered = clientsByEp.remove(toBeUnregistered.getEndpoint());
-            for (ClientRegistryListener l : listeners) {
-                l.unregistered(unregistered);
-            }
-            LOG.debug("Deregistered client: {}", unregistered);
-            return unregistered;
+        Deregistration unregistered = store.removeRegistration(registrationId);
+        for (ClientRegistryListener l : listeners) {
+            l.unregistered(unregistered.getRegistration());
         }
+        LOG.debug("Deregistered client: {}", unregistered);
+        return unregistered.getRegistration();
     }
 
     @Override
     public Client findByRegistrationId(String id) {
-        Client result = null;
-        if (id != null) {
-            for (Client client : clientsByEp.values()) {
-                if (id.equals(client.getRegistrationId())) {
-                    result = client;
-                    break;
-                }
-            }
-        }
-        return result;
+        return store.getRegistration(id);
     }
 
     /**
@@ -143,9 +129,8 @@ public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable
      */
     @Override
     public void start() {
-        // every 2 seconds clean the registration list
-        // TODO re-consider clean-up interval: wouldn't 5 minutes do as well?
-        schedExecutor.scheduleAtFixedRate(new Cleaner(), 2, 2, TimeUnit.SECONDS);
+        if (store instanceof Startable)
+            ((Startable) store).start();
     }
 
     /**
@@ -153,32 +138,14 @@ public class ClientRegistryImpl implements ClientRegistry, Startable, Stoppable
      */
     @Override
     public void stop() {
-        schedExecutor.shutdownNow();
-        try {
-            schedExecutor.awaitTermination(5, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            LOG.warn("Clean up registration thread was interrupted.", e);
-        }
+        if (store instanceof Stoppable)
+            ((Stoppable) store).stop();
     }
 
-    private final ScheduledExecutorService schedExecutor = Executors.newScheduledThreadPool(1);
-
-    private class Cleaner implements Runnable {
-
-        @Override
-        public void run() {
-            try {
-                for (Client client : clientsByEp.values()) {
-                    synchronized (client) {
-                        if (!client.isAlive()) {
-                            // force de-registration
-                            deregisterClient(client.getRegistrationId());
-                        }
-                    }
-                }
-            } catch (Exception e) {
-                LOG.warn("Unexcepted Exception while registration cleaning", e);
-            }
+    @Override
+    public void registrationExpired(Client registration, Collection<Observation> observation) {
+        for (ClientRegistryListener l : listeners) {
+            l.unregistered(registration);
         }
     }
 }
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/Deregistration.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/Deregistration.java
new file mode 100644
index 00000000..d7e82515
--- /dev/null
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/Deregistration.java
@@ -0,0 +1,48 @@
+/*******************************************************************************
+ * 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 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.registration;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.client.Client;
+
+/**
+ * A Deregistration contains all informations which are removed after a client was unregistered.
+ * 
+ * @see RegistrationStore
+ */
+public class Deregistration {
+    final Client registration;
+    final Collection<Observation> observations;
+
+    public Deregistration(Client registration, Collection<Observation> observations) {
+        this.registration = registration;
+        if (observations == null)
+            this.observations = Collections.emptyList();
+        else
+            this.observations = observations;
+    }
+
+    public Client getRegistration() {
+        return registration;
+    }
+
+    public Collection<Observation> getObservations() {
+        return observations;
+    }
+}
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/ExpirationListener.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/ExpirationListener.java
new file mode 100644
index 00000000..c4a9b903
--- /dev/null
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/ExpirationListener.java
@@ -0,0 +1,31 @@
+/*******************************************************************************
+ * 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 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.registration;
+
+import java.util.Collection;
+
+import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.client.Client;
+
+/**
+ * A listener to be aware of registration expiration.
+ * 
+ * @see RegistrationStore
+ */
+public interface ExpirationListener {
+
+    void registrationExpired(Client registration, Collection<Observation> observation);
+}
diff --git a/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/RegistrationStore.java b/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/RegistrationStore.java
new file mode 100644
index 00000000..edc681e1
--- /dev/null
+++ b/leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/RegistrationStore.java
@@ -0,0 +1,120 @@
+/*******************************************************************************
+ * 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 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.registration;
+
+import java.net.InetSocketAddress;
+import java.util.Collection;
+
+import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.server.client.Client;
+import org.eclipse.leshan.server.client.ClientUpdate;
+
+/**
+ * A store for registrations and observations. This interface is also responsible to handle registration expiration.
+ */
+public interface RegistrationStore {
+
+    /**
+     * Add a new registration. If there is already a registration with the same endpoint removed it.
+     * 
+     * @param registration the new registration.
+     * @return the old registration and its observations removed or null.
+     */
+    Deregistration addRegistration(Client registration);
+
+    /**
+     * Update an existing registration
+     * 
+     * @param update data to update
+     * @return the registration updated
+     */
+    Client updateRegistration(ClientUpdate update);
+
+    /**
+     * Get the registration by registration Id.
+     * 
+     * @param registrationId of the registration.
+     * @return the registration or null if there is no registration with this id.
+     */
+    Client getRegistration(String registrationId);
+
+    /**
+     * Get the registration by endpoint.
+     * 
+     * @param endpoint of the registration.
+     * @return the registration or null if there is no registration with this endpoint.
+     */
+    Client getRegistrationByEndpoint(String endpoint);
+
+    /**
+     * Get the registration by socket address.
+     * 
+     * @param address of the client registered.
+     * @return the registration or null if there is no client registered with this socket address.
+     */
+    Collection<Client> getRegistrationByAdress(InetSocketAddress address);
+
+    /**
+     * @return all registrations in this store.
+     * @Deprecated should be replaced by an iterator.
+     */
+    // TODO Should be replaced by an iterator.
+    @Deprecated
+    Collection<Client> getAllRegistration();
+
+    /**
+     * Remove the registration with the given registration Id
+     * 
+     * @param registrationId the id of the registration to removed
+     * @return the registration and observations removed or null if there is no registration for this Id.
+     */
+    Deregistration removeRegistration(String registrationId);
+
+    /**
+     * Add a new Observation for a given registration.
+     * 
+     * @param registrationId the id of the registration
+     * @param observation the observation to add
+     * 
+     * @return the previous observation or null if any.
+     */
+    Observation addObservation(String registrationId, Observation observation);
+
+    /**
+     * Get the observation for the given registration with the given observationId
+     */
+    Observation getObservation(String registrationId, byte[] observationId);
+
+    /**
+     * Remove the observation for the given registration with the given observationId
+     */
+    Observation removeObservation(String registrationId, byte[] observationId);
+
+    /**
+     * Get all observations for the given registrationId
+     */
+    Collection<Observation> getObservations(String registrationId);
+
+    /**
+     * Remove all observations for the given registrationId
+     */
+    Collection<Observation> removeObservations(String registrationId);
+
+    /**
+     * set a listener for registration expiration.
+     */
+    void setExpirationListener(ExpirationListener listener);
+}
diff --git a/leshan-server-demo/src/main/java/org/eclipse/leshan/server/demo/LeshanServerDemo.java b/leshan-server-demo/src/main/java/org/eclipse/leshan/server/demo/LeshanServerDemo.java
index dd9904d1..5b81d712 100644
--- a/leshan-server-demo/src/main/java/org/eclipse/leshan/server/demo/LeshanServerDemo.java
+++ b/leshan-server-demo/src/main/java/org/eclipse/leshan/server/demo/LeshanServerDemo.java
@@ -45,11 +45,8 @@ import org.eclipse.leshan.core.node.codec.DefaultLwM2mNodeDecoder;
 import org.eclipse.leshan.core.node.codec.DefaultLwM2mNodeEncoder;
 import org.eclipse.leshan.core.node.codec.LwM2mNodeDecoder;
 import org.eclipse.leshan.server.californium.LeshanServerBuilder;
-import org.eclipse.leshan.server.californium.impl.CaliforniumObservationRegistryImpl;
 import org.eclipse.leshan.server.californium.impl.LeshanServer;
-import org.eclipse.leshan.server.client.ClientRegistry;
-import org.eclipse.leshan.server.cluster.RedisClientRegistry;
-import org.eclipse.leshan.server.cluster.RedisObservationStore;
+import org.eclipse.leshan.server.cluster.RedisRegistrationStore;
 import org.eclipse.leshan.server.cluster.RedisSecurityRegistry;
 import org.eclipse.leshan.server.demo.servlet.ClientServlet;
 import org.eclipse.leshan.server.demo.servlet.EventServlet;
@@ -217,11 +214,8 @@ public class LeshanServerDemo {
                 builder.setSecurityRegistry(new SecurityRegistryImpl(privateKey, publicKey));
             } else {
                 // use Redis
-                ClientRegistry clientRegistry = new RedisClientRegistry(jedis);
                 builder.setSecurityRegistry(new RedisSecurityRegistry(jedis, privateKey, publicKey));
-                builder.setClientRegistry(clientRegistry);
-                builder.setObservationRegistry(new CaliforniumObservationRegistryImpl(
-                        new RedisObservationStore(jedis), clientRegistry, modelProvider, decoder));
+                builder.setRegistrationStore(new RedisRegistrationStore(jedis));
             }
         } catch (InvalidKeySpecException | NoSuchAlgorithmException | InvalidParameterSpecException e) {
             LOG.error("Unable to initialize RPK.", e);


Test patch:
diff --git a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeIntegrationTestHelper.java b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeIntegrationTestHelper.java
index 282852e1..bc0e8f6b 100644
--- a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeIntegrationTestHelper.java
+++ b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/QueueModeIntegrationTestHelper.java
@@ -43,8 +43,7 @@ import org.eclipse.leshan.integration.tests.util.QueueModeLeshanServer;
 import org.eclipse.leshan.integration.tests.util.QueuedModeLeshanClient;
 import org.eclipse.leshan.server.californium.impl.CaliforniumLwM2mRequestSender;
 import org.eclipse.leshan.server.californium.impl.CaliforniumObservationRegistryImpl;
-import org.eclipse.leshan.server.californium.impl.InMemoryLwM2mObservationStore;
-import org.eclipse.leshan.server.californium.impl.LwM2mObservationStore;
+import org.eclipse.leshan.server.californium.impl.InMemoryRegistrationStore;
 import org.eclipse.leshan.server.californium.impl.RegisterResource;
 import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.client.ClientRegistry;
@@ -85,7 +84,8 @@ public class QueueModeIntegrationTestHelper extends IntegrationTestHelper {
         super.deregisterLatch = new CountDownLatch(1);
         super.updateLatch = new CountDownLatch(1);
 
-        ClientRegistry clientRegistry = new ClientRegistryImpl();
+        InMemoryRegistrationStore registrationStore = new InMemoryRegistrationStore();
+        ClientRegistry clientRegistry = new ClientRegistryImpl(registrationStore);
         SecurityRegistry securityRegistry = new SecurityRegistryImpl() {
             @Override
             protected void loadFromFile() {
@@ -97,15 +97,14 @@ public class QueueModeIntegrationTestHelper extends IntegrationTestHelper {
                 // do not save to file
             }
         };
-        LwM2mObservationStore observationStore = new InMemoryLwM2mObservationStore();
 
         // coap server
         noSecureEndpoint = new CoapEndpoint(
                 new InetSocketAddress(InetAddress.getLoopbackAddress(), networkConfig.getInt(Keys.COAP_PORT)),
-                networkConfig, observationStore);
+                networkConfig, registrationStore);
         secureEndpoint = new CoapEndpoint(
                 new InetSocketAddress(InetAddress.getLoopbackAddress(), networkConfig.getInt(Keys.COAP_SECURE_PORT)),
-                networkConfig, observationStore);
+                networkConfig, registrationStore);
         coapServer = new CoapServer(networkConfig);
         coapServer.addEndpoint(noSecureEndpoint);
         coapServer.addEndpoint(secureEndpoint);
@@ -118,7 +117,7 @@ public class QueueModeIntegrationTestHelper extends IntegrationTestHelper {
         LwM2mNodeEncoder encoder = new DefaultLwM2mNodeEncoder();
         LwM2mNodeDecoder decoder = new DefaultLwM2mNodeDecoder();
         CaliforniumObservationRegistryImpl observationRegistry = new CaliforniumObservationRegistryImpl(
-                observationStore, clientRegistry, modelProvider, decoder);
+                registrationStore, clientRegistry, modelProvider, decoder);
         observationRegistry.setSecureEndpoint(secureEndpoint);
         secureEndpoint.addNotificationListener(observationRegistry);
         observationRegistry.setNonSecureEndpoint(noSecureEndpoint);
diff --git a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisIntegrationTestHelper.java b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisIntegrationTestHelper.java
index f0ad7c25..0bd856f8 100644
--- a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisIntegrationTestHelper.java
+++ b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisIntegrationTestHelper.java
@@ -20,13 +20,10 @@ import java.net.InetSocketAddress;
 
 import org.eclipse.leshan.core.node.codec.DefaultLwM2mNodeDecoder;
 import org.eclipse.leshan.server.californium.LeshanServerBuilder;
-import org.eclipse.leshan.server.californium.impl.CaliforniumObservationRegistryImpl;
 import org.eclipse.leshan.server.client.Client;
-import org.eclipse.leshan.server.client.ClientRegistry;
 import org.eclipse.leshan.server.client.ClientRegistryListener;
 import org.eclipse.leshan.server.client.ClientUpdate;
-import org.eclipse.leshan.server.cluster.RedisClientRegistry;
-import org.eclipse.leshan.server.cluster.RedisObservationStore;
+import org.eclipse.leshan.server.cluster.RedisRegistrationStore;
 import org.eclipse.leshan.server.impl.SecurityRegistryImpl;
 import org.eclipse.leshan.server.model.StaticModelProvider;
 
@@ -64,11 +61,7 @@ public class RedisIntegrationTestHelper extends IntegrationTestHelper {
         if (redisURI == null)
             redisURI = "";
         Pool<Jedis> jedis = new JedisPool(redisURI);
-        ClientRegistry clientRegistry = new RedisClientRegistry(jedis);
-        builder.setClientRegistry(clientRegistry);
-        builder.setObservationRegistry(
-                new CaliforniumObservationRegistryImpl(new RedisObservationStore(jedis), clientRegistry, modelProvider,
-                        decoder));
+        builder.setRegistrationStore(new RedisRegistrationStore(jedis));
 
         // Build server !
         server = builder.build();
diff --git a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisSecureIntegrationTestHelper.java b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisSecureIntegrationTestHelper.java
index 07831c32..f9d6f1a1 100644
--- a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisSecureIntegrationTestHelper.java
+++ b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RedisSecureIntegrationTestHelper.java
@@ -21,13 +21,10 @@ import java.security.cert.Certificate;
 
 import org.eclipse.leshan.core.node.codec.DefaultLwM2mNodeDecoder;
 import org.eclipse.leshan.server.californium.LeshanServerBuilder;
-import org.eclipse.leshan.server.californium.impl.CaliforniumObservationRegistryImpl;
 import org.eclipse.leshan.server.client.Client;
-import org.eclipse.leshan.server.client.ClientRegistry;
 import org.eclipse.leshan.server.client.ClientRegistryListener;
 import org.eclipse.leshan.server.client.ClientUpdate;
-import org.eclipse.leshan.server.cluster.RedisClientRegistry;
-import org.eclipse.leshan.server.cluster.RedisObservationStore;
+import org.eclipse.leshan.server.cluster.RedisRegistrationStore;
 import org.eclipse.leshan.server.cluster.RedisSecurityRegistry;
 import org.eclipse.leshan.server.model.StaticModelProvider;
 
@@ -51,10 +48,7 @@ public class RedisSecureIntegrationTestHelper extends SecureIntegrationTestHelpe
         if (redisURI == null)
             redisURI = "";
         Pool<Jedis> jedis = new JedisPool(redisURI);
-        ClientRegistry clientRegistry = new RedisClientRegistry(jedis);
-        builder.setClientRegistry(clientRegistry);
-        builder.setObservationRegistry(new CaliforniumObservationRegistryImpl(new RedisObservationStore(jedis),
-                clientRegistry, modelProvider, decoder));
+        builder.setRegistrationStore(new RedisRegistrationStore(jedis));
         builder.setSecurityRegistry(new RedisSecurityRegistry(jedis, null, null));
 
         // Build server !
diff --git a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RegistrationTest.java b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RegistrationTest.java
index aea55b70..97c2fa4f 100644
--- a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RegistrationTest.java
+++ b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/RegistrationTest.java
@@ -17,6 +17,8 @@
 package org.eclipse.leshan.integration.tests;
 
 import static org.eclipse.leshan.integration.tests.IntegrationTestHelper.LIFETIME;
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
 import static org.junit.Assert.*;
 
 import java.net.InetAddress;
@@ -25,19 +27,27 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
+import org.eclipse.californium.core.coap.Response;
 import org.eclipse.californium.core.network.Endpoint;
 import org.eclipse.leshan.LinkObject;
+import org.eclipse.leshan.ResponseCode;
 import org.eclipse.leshan.client.californium.LeshanClient;
 import org.eclipse.leshan.client.californium.impl.CaliforniumLwM2mClientRequestSender;
 import org.eclipse.leshan.client.resource.LwM2mInstanceEnabler;
 import org.eclipse.leshan.client.resource.LwM2mObjectEnabler;
 import org.eclipse.leshan.client.resource.ObjectEnabler;
+import org.eclipse.leshan.core.node.LwM2mPath;
+import org.eclipse.leshan.core.observation.Observation;
 import org.eclipse.leshan.core.request.DeregisterRequest;
+import org.eclipse.leshan.core.request.ObserveRequest;
 import org.eclipse.leshan.core.request.ReadRequest;
 import org.eclipse.leshan.core.request.RegisterRequest;
+import org.eclipse.leshan.core.response.ObserveResponse;
 import org.eclipse.leshan.core.response.ReadResponse;
 import org.eclipse.leshan.core.response.RegisterResponse;
+import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.security.NonUniqueSecurityInfoException;
 import org.junit.After;
 import org.junit.Before;
@@ -183,6 +193,50 @@ public class RegistrationTest {
         helper.assertClientRegisterered();
     }
 
+    @Test
+    public void register_observe_deregister_observe() throws NonUniqueSecurityInfoException, InterruptedException {
+        // Check client is not registered
+        helper.assertClientNotRegisterered();
+
+        // Start it and wait for registration
+        helper.client.start();
+        helper.waitForRegistration(1);
+
+        // Check client is well registered
+        helper.assertClientRegisterered();
+
+        // observe device timezone
+        ObserveResponse observeResponse = helper.server.send(helper.getCurrentRegistration(), new ObserveRequest(3, 0));
+        assertEquals(ResponseCode.CONTENT, observeResponse.getCode());
+        assertNotNull(observeResponse.getCoapResponse());
+        assertThat(observeResponse.getCoapResponse(), is(instanceOf(Response.class)));
+
+        // check observation registry is not null
+        Client currentRegistration = helper.getCurrentRegistration();
+        Set<Observation> observations = helper.server.getObservationRegistry()
+                .getObservations(currentRegistration);
+        assertEquals(1, observations.size());
+        Observation obs = observations.iterator().next();
+        assertEquals(currentRegistration.getRegistrationId(), obs.getRegistrationId());
+        assertEquals(new LwM2mPath(3, 0), obs.getPath());
+
+        // Check de-registration
+        helper.client.stop(true);
+        helper.waitForDeregistration(1);
+        helper.assertClientNotRegisterered();
+        observations = helper.server.getObservationRegistry()
+                .getObservations(currentRegistration);
+        assertTrue(observations.isEmpty());
+
+        // try to send a new observation
+        observeResponse = helper.server.send(currentRegistration, new ObserveRequest(3, 0), 50);
+        assertNull(observeResponse);
+
+        // check observationStore is empty
+        observations = helper.server.getObservationRegistry().getObservations(currentRegistration);
+        assertTrue(observations.isEmpty());
+    }
+
     // TODO not really a registration test
     @Test(expected = IllegalArgumentException.class)
     public void fail_to_create_client_with_same_object_twice() {
diff --git a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationTest.java b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationTest.java
index b728594c..dcfd3e39 100644
--- a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationTest.java
+++ b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CaliforniumObservationTest.java
@@ -15,6 +15,9 @@
  *******************************************************************************/
 package org.eclipse.leshan.server.californium.impl;
 
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -24,6 +27,8 @@ import org.eclipse.leshan.core.node.LwM2mPath;
 import org.eclipse.leshan.core.node.codec.DefaultLwM2mNodeDecoder;
 import org.eclipse.leshan.core.observation.Observation;
 import org.eclipse.leshan.server.californium.CaliforniumObservationRegistry;
+import org.eclipse.leshan.server.californium.CaliforniumRegistrationStore;
+import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.impl.ClientRegistryImpl;
 import org.eclipse.leshan.server.model.StandardModelProvider;
 import org.junit.Assert;
@@ -35,15 +40,16 @@ public class CaliforniumObservationTest {
     Request coapRequest;
     LwM2mPath target;
     CaliforniumObservationRegistry registry;
-    InMemoryLwM2mObservationStore store;
+    CaliforniumRegistrationStore store;
 
     private CaliforniumTestSupport support = new CaliforniumTestSupport();
 
     @Before
     public void setUp() throws Exception {
         support.givenASimpleClient();
-        store = new InMemoryLwM2mObservationStore();
-        registry = new CaliforniumObservationRegistryImpl(store, new ClientRegistryImpl(), new StandardModelProvider(),
+        store = new InMemoryRegistrationStore();
+        registry = new CaliforniumObservationRegistryImpl(store, new ClientRegistryImpl(store),
+                new StandardModelProvider(),
                 new DefaultLwM2mNodeDecoder());
     }
 
@@ -101,7 +107,7 @@ public class CaliforniumObservationTest {
     }
 
     @Test
-    public void cancel_by_observation() {
+    public void cancel_by_observation() throws UnknownHostException {
         // create some observations and add it to registry
         givenAnObservation(support.client.getRegistrationId(), new LwM2mPath(3, 0, 13));
         givenAnObservation(support.client.getRegistrationId(), new LwM2mPath(3, 0, 12));
@@ -123,6 +129,9 @@ public class CaliforniumObservationTest {
     }
 
     private Observation givenAnObservation(String registrationId, LwM2mPath target) {
+        if (store.getRegistration(registrationId) == null)
+            store.addRegistration(givenASimpleClient(registrationId));
+        
         coapRequest = Request.newGet();
         coapRequest.setToken(CaliforniumTestSupport.createToken());
         coapRequest.getOptions().addUriPath(String.valueOf(target.getObjectId()));
@@ -143,4 +152,15 @@ public class CaliforniumObservationTest {
         return observation;
     }
 
+    public Client givenASimpleClient(String registrationId) {
+        InetSocketAddress registrationAddress = InetSocketAddress.createUnresolved("localhost", 5683);
+        Client.Builder builder;
+        try {
+            builder = new Client.Builder(registrationId, registrationId + "_ep", InetAddress.getLocalHost(), 10000,
+                    registrationAddress);
+            return builder.build();
+        } catch (UnknownHostException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }
diff --git a/leshan-server-core/src/test/java/org/eclipse/leshan/server/impl/ClientRegistryImplTest.java b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ClientRegistryImplTest.java
similarity index 94%
rename from leshan-server-core/src/test/java/org/eclipse/leshan/server/impl/ClientRegistryImplTest.java
rename to leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ClientRegistryImplTest.java
index f81b0bd8..640db922 100644
--- a/leshan-server-core/src/test/java/org/eclipse/leshan/server/impl/ClientRegistryImplTest.java
+++ b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ClientRegistryImplTest.java
@@ -13,7 +13,7 @@
  * Contributors:
  *     Sierra Wireless - initial API and implementation
  *******************************************************************************/
-package org.eclipse.leshan.server.impl;
+package org.eclipse.leshan.server.californium.impl;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -22,6 +22,7 @@ import org.eclipse.leshan.LinkObject;
 import org.eclipse.leshan.core.request.BindingMode;
 import org.eclipse.leshan.server.client.Client;
 import org.eclipse.leshan.server.client.ClientUpdate;
+import org.eclipse.leshan.server.impl.ClientRegistryImpl;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -42,7 +43,7 @@ public class ClientRegistryImplTest {
     @Before
     public void setUp() throws Exception {
         address = InetAddress.getLocalHost();
-        registry = new ClientRegistryImpl();
+        registry = new ClientRegistryImpl(new InMemoryRegistrationStore());
     }
 
     @Test
diff --git a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilderTest.java b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilderTest.java
index 8d3e2044..4ae67392 100644
--- a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilderTest.java
+++ b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/CoapRequestBuilderTest.java
@@ -89,7 +89,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         ReadRequest request = new ReadRequest(3, 0);
         builder.visit(request);
 
@@ -108,7 +108,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         ReadRequest request = new ReadRequest(3, 0, 1);
         builder.visit(request);
 
@@ -124,7 +124,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         ReadRequest request = new ReadRequest(3);
         builder.visit(request);
 
@@ -140,7 +140,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         DiscoverRequest request = new DiscoverRequest(3, 0);
         builder.visit(request);
 
@@ -160,7 +160,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         WriteRequest request = new WriteRequest(Mode.UPDATE, 3, 0, LwM2mSingleResource.newStringResource(4, "value"));
         builder.visit(request);
 
@@ -185,7 +185,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         WriteRequest request = new WriteRequest(3, 0, 14, "value");
         builder.visit(request);
 
@@ -201,7 +201,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         WriteAttributesRequest request = new WriteAttributesRequest(3, 0, 14,
                 new ObserveSpec.Builder().minPeriod(10).maxPeriod(100).build());
         builder.visit(request);
@@ -221,7 +221,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         ExecuteRequest request = new ExecuteRequest(3, 0, 12, "params");
         builder.visit(request);
 
@@ -241,7 +241,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         CreateRequest request = new CreateRequest(12, LwM2mSingleResource.newStringResource(0, "value"));
         builder.visit(request);
 
@@ -265,7 +265,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         CreateRequest request = new CreateRequest(12,
                 new LwM2mObjectInstance(26, LwM2mSingleResource.newStringResource(0, "value")));
         builder.visit(request);
@@ -291,7 +291,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         DeleteRequest request = new DeleteRequest(12, 0);
         builder.visit(request);
 
@@ -310,7 +310,7 @@ public class CoapRequestBuilderTest {
         // test
         CoapRequestBuilder builder = new CoapRequestBuilder(
                 new InetSocketAddress(client.getAddress(), client.getPort()), client.getRootPath(),
-                client.getRegistrationId(), model, encoder);
+                client.getRegistrationId(), client.getEndpoint(), model, encoder);
         ObserveRequest request = new ObserveRequest(12, 0);
         builder.visit(request);