Repository: eclipse/leshan
Pull request #212 (2016-12-02 16:37:35)

fixed humitidy minimum value to be float instead of string
Signed-off-by: Joakim Eriksson <joakime@sics.se>

I Had some issues with the humidity sensors and it seems to be the JSON description file that was configured for string intstead of float for the minimum humidity value. This is a fix for that.

BTW: Is there anyone planning to add the other IPSO objects that are not yet supported?


Patch:
diff --git a/leshan-core/src/main/resources/oma-objects-spec.json b/leshan-core/src/main/resources/oma-objects-spec.json
index dc139f9a..330e4519 100644
--- a/leshan-core/src/main/resources/oma-objects-spec.json
+++ b/leshan-core/src/main/resources/oma-objects-spec.json
@@ -1705,7 +1705,7 @@
         "operations": "R",
         "instancetype": "single",
         "mandatory": false,
-        "type": "string",
+        "type": "float",
         "range": "",
         "units": "Defined by “Units” resource.",
         "description": "The minimum value that can be measured by the sensor"
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);