Repository: eclipse/leshan
Pull request #477 (2018-02-26 17:20:59)

Integration of Californium 2.0.0 M8
Aims to solve issue about #476 

boaks: Very good. Can I close my old PR? It doesn't make sense, if you complete it here. 
sbernard31: I close the old PR.
@boaks does this PR sounds good to you ? Should I merge it ?

Patch:
diff --git a/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/LeshanClientBuilder.java b/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/LeshanClientBuilder.java
index 1b97c662..9d47bccc 100644
--- a/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/LeshanClientBuilder.java
+++ b/leshan-client-cf/src/main/java/org/eclipse/leshan/client/californium/LeshanClientBuilder.java
@@ -246,7 +246,10 @@ public class LeshanClientBuilder {
             if (endpointFactory != null) {
                 unsecuredEndpoint = endpointFactory.createUnsecuredEndpoint(localAddress, coapConfig, null);
             } else {
-                unsecuredEndpoint = new CoapEndpoint(localAddress, coapConfig);
+                CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+                builder.setInetSocketAddress(localAddress);
+                builder.setNetworkConfig(coapConfig);
+                unsecuredEndpoint = builder.build();
             }
         }
 
@@ -255,7 +258,10 @@ public class LeshanClientBuilder {
             if (endpointFactory != null) {
                 securedEndpoint = endpointFactory.createSecuredEndpoint(dtlsConfig, coapConfig, null);
             } else {
-                securedEndpoint = new CoapEndpoint(new DTLSConnector(dtlsConfig), coapConfig, null, null);
+                CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+                builder.setConnector(new DTLSConnector(dtlsConfig));
+                builder.setNetworkConfig(coapConfig);
+                securedEndpoint = builder.build();
             }
         }
 
@@ -264,6 +270,7 @@ public class LeshanClientBuilder {
                     "All CoAP enpoints are deactivated, at least one endpoint should be activated");
         }
 
-        return new LeshanClient(endpoint, unsecuredEndpoint, securedEndpoint, objectEnablers, coapConfig, additionalAttributes);
+        return new LeshanClient(endpoint, unsecuredEndpoint, securedEndpoint, objectEnablers, coapConfig,
+                additionalAttributes);
     }
 }
diff --git a/leshan-core/src/main/java/org/eclipse/leshan/util/Key.java b/leshan-core/src/main/java/org/eclipse/leshan/util/Key.java
deleted file mode 100644
index 72fec51a..00000000
--- a/leshan-core/src/main/java/org/eclipse/leshan/util/Key.java
+++ /dev/null
@@ -1,39 +0,0 @@
-package org.eclipse.leshan.util;
-
-import java.util.Arrays;
-
-public class Key {
-
-    private final byte[] bytes;
-
-    public Key(byte[] bytes) {
-        this.bytes = bytes;
-    }
-
-    @Override
-    public String toString() {
-        return Hex.encodeHexString(bytes);
-    }
-
-    @Override
-    public int hashCode() {
-        final int prime = 31;
-        int result = 1;
-        result = prime * result + Arrays.hashCode(bytes);
-        return result;
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (this == obj)
-            return true;
-        if (obj == null)
-            return false;
-        if (getClass() != obj.getClass())
-            return false;
-        Key other = (Key) obj;
-        if (!Arrays.equals(bytes, other.bytes))
-            return false;
-        return true;
-    }
-}
\ No newline at end of file
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanBootstrapServerBuilder.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanBootstrapServerBuilder.java
index b6f2e339..f9e05caf 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanBootstrapServerBuilder.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/LeshanBootstrapServerBuilder.java
@@ -12,6 +12,7 @@
  * 
  * Contributors:
  *     Sierra Wireless - initial API and implementation
+ *     Achim Kraus (Bosch Software Innovations GmbH) - use CoapEndpointBuilder
  *******************************************************************************/
 package org.eclipse.leshan.server.californium;
 
@@ -252,7 +253,10 @@ public class LeshanBootstrapServerBuilder {
             if (endpointFactory != null) {
                 unsecuredEndpoint = endpointFactory.createUnsecuredEndpoint(localAddress, coapConfig, null);
             } else {
-                unsecuredEndpoint = new CoapEndpoint(localAddress, coapConfig);
+                CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+                builder.setInetSocketAddress(localAddress);
+                builder.setNetworkConfig(coapConfig);
+                unsecuredEndpoint = builder.build();
             }
         }
 
@@ -261,8 +265,11 @@ public class LeshanBootstrapServerBuilder {
             if (endpointFactory != null) {
                 securedEndpoint = endpointFactory.createSecuredEndpoint(dtlsConfig, coapConfig, null);
             } else {
-                securedEndpoint = new CoapEndpoint(new DTLSConnector(dtlsConfig), coapConfig, null, null,
-                        new Lwm2mEndpointContextMatcher());
+                CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+                builder.setConnector(new DTLSConnector(dtlsConfig));
+                builder.setNetworkConfig(coapConfig);
+                builder.setEndpointContextMatcher(new Lwm2mEndpointContextMatcher());
+                securedEndpoint = builder.build();
             }
         }
 
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 d900989d..554afcbc 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
@@ -14,6 +14,7 @@
  *     Sierra Wireless - initial API and implementation
  *     Achim Kraus (Bosch Software Innovations GmbH) - use Lwm2mEndpointContextMatcher
  *                                                     for secure endpoint.
+ *     Achim Kraus (Bosch Software Innovations GmbH) - use CoapEndpointBuilder
  *******************************************************************************/
 package org.eclipse.leshan.server.californium;
 
@@ -435,7 +436,11 @@ public class LeshanServerBuilder {
                 unsecuredEndpoint = endpointFactory.createUnsecuredEndpoint(localAddress, coapConfig,
                         registrationStore);
             } else {
-                unsecuredEndpoint = new CoapEndpoint(localAddress, coapConfig, registrationStore);
+                CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+                builder.setInetSocketAddress(localAddress);
+                builder.setNetworkConfig(coapConfig);
+                builder.setObservationStore(registrationStore);
+                unsecuredEndpoint = builder.build();
             }
         }
 
@@ -444,8 +449,12 @@ public class LeshanServerBuilder {
             if (endpointFactory != null) {
                 securedEndpoint = endpointFactory.createSecuredEndpoint(dtlsConfig, coapConfig, registrationStore);
             } else {
-                securedEndpoint = new CoapEndpoint(new DTLSConnector(dtlsConfig), coapConfig, registrationStore, null,
-                        new Lwm2mEndpointContextMatcher());
+                CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+                builder.setConnector(new DTLSConnector(dtlsConfig));
+                builder.setNetworkConfig(coapConfig);
+                builder.setObservationStore(registrationStore);
+                builder.setEndpointContextMatcher(new Lwm2mEndpointContextMatcher());
+                securedEndpoint = builder.build();
             }
         }
 
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/ObserveUtil.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/ObserveUtil.java
index 3460f965..caa12a4c 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/ObserveUtil.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/ObserveUtil.java
@@ -60,7 +60,7 @@ public class ObserveUtil {
                 context.put(ctx.getKey(), ctx.getValue());
             }
         }
-        return new Observation(request.getToken(), regId, new LwM2mPath(lwm2mPath), context);
+        return new Observation(request.getToken().getBytes(), regId, new LwM2mPath(lwm2mPath), context);
     }
 
     /**
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
index 7cbb7277..8935f2e1 100644
--- 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
@@ -19,6 +19,8 @@
  *                                                     setContext().
  *     Achim Kraus (Bosch Software Innovations GmbH) - rename CorrelationContext to
  *                                                     EndpointContext
+ *     Achim Kraus (Bosch Software Innovations GmbH) - update to modified 
+ *                                                     ObservationStore API
  *******************************************************************************/
 package org.eclipse.leshan.server.californium.impl;
 
@@ -27,15 +29,18 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 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.Token;
 import org.eclipse.californium.core.observe.ObservationUtil;
 import org.eclipse.californium.elements.EndpointContext;
 import org.eclipse.leshan.core.observation.Observation;
@@ -48,7 +53,6 @@ import org.eclipse.leshan.server.registration.ExpirationListener;
 import org.eclipse.leshan.server.registration.Registration;
 import org.eclipse.leshan.server.registration.RegistrationUpdate;
 import org.eclipse.leshan.server.registration.UpdatedRegistration;
-import org.eclipse.leshan.util.Key;
 import org.eclipse.leshan.util.NamedThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,8 +65,8 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
 
     // Data structure
     private final Map<String /* end-point */, Registration> regsByEp = new HashMap<>();
-    private Map<Key, org.eclipse.californium.core.observe.Observation> obsByToken = new HashMap<>();
-    private Map<String, List<Key>> tokensByRegId = new HashMap<>();
+    private Map<Token, org.eclipse.californium.core.observe.Observation> obsByToken = new HashMap<>();
+    private Map<String, Set<Token>> tokensByRegId = new HashMap<>();
 
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
@@ -212,7 +216,7 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
             // cancel existing observations for the same path and registration id.
             for (Observation obs : unsafeGetObservations(registrationId)) {
                 if (observation.getPath().equals(obs.getPath()) && !Arrays.equals(observation.getId(), obs.getId())) {
-                    unsafeRemoveObservation(obs.getId());
+                    unsafeRemoveObservation(new Token(obs.getId()));
                     removed.add(obs);
                 }
             }
@@ -227,10 +231,10 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     public Observation removeObservation(String registrationId, byte[] observationId) {
         try {
             lock.writeLock().lock();
-
-            Observation observation = build(unsafeGetObservation(new Key(observationId)));
+            Token token = new Token(observationId);
+            Observation observation = build(unsafeGetObservation(token));
             if (observation != null && registrationId.equals(observation.getRegistrationId())) {
-                unsafeRemoveObservation(observationId);
+                unsafeRemoveObservation(token);
                 return observation;
             }
             return null;
@@ -243,7 +247,7 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     public Observation getObservation(String registrationId, byte[] observationId) {
         try {
             lock.readLock().lock();
-            Observation observation = build(unsafeGetObservation(new Key(observationId)));
+            Observation observation = build(unsafeGetObservation(new Token(observationId)));
             if (observation != null && registrationId.equals(observation.getRegistrationId())) {
                 return observation;
             }
@@ -276,7 +280,17 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     /* *************** Californium ObservationStore API **************** */
 
     @Override
-    public void add(org.eclipse.californium.core.observe.Observation obs) {
+    public org.eclipse.californium.core.observe.Observation putIfAbsent(Token token, org.eclipse.californium.core.observe.Observation obs) {
+        return add(token, obs, true);
+    }
+
+    @Override
+    public org.eclipse.californium.core.observe.Observation put(Token token, org.eclipse.californium.core.observe.Observation obs) {
+        return add(token, obs, false);
+    }
+
+    private org.eclipse.californium.core.observe.Observation add(Token token, org.eclipse.californium.core.observe.Observation obs, boolean ifAbsent) {
+        org.eclipse.californium.core.observe.Observation previousObservation = null;
         if (obs != null) {
             try {
                 lock.writeLock().lock();
@@ -284,10 +298,16 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
                 validateObservation(obs);
 
                 String registrationId = ObserveUtil.extractRegistrationId(obs);
-                Key token = new Key(obs.getRequest().getToken());
-                org.eclipse.californium.core.observe.Observation previousObservation = obsByToken.put(token, obs);
+                if (ifAbsent) {
+                    if (!obsByToken.containsKey(token))
+                        previousObservation = obsByToken.put(token, obs);
+                    else
+                        return obsByToken.get(token);
+                } else {
+                    previousObservation = obsByToken.put(token, obs);
+                }
                 if (!tokensByRegId.containsKey(registrationId)) {
-                    tokensByRegId.put(registrationId, new ArrayList<Key>());
+                    tokensByRegId.put(registrationId, new HashSet<Token>());
                 }
                 tokensByRegId.get(registrationId).add(token);
 
@@ -301,26 +321,26 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
                 lock.writeLock().unlock();
             }
         }
+        return previousObservation;
     }
 
     @Override
-    public org.eclipse.californium.core.observe.Observation get(byte[] token) {
+    public org.eclipse.californium.core.observe.Observation get(Token token) {
         try {
             lock.readLock().lock();
-            return unsafeGetObservation(new Key(token));
+            return unsafeGetObservation(token);
         } finally {
             lock.readLock().unlock();
         }
     }
 
     @Override
-    public void setContext(byte[] token, EndpointContext ctx) {
+    public void setContext(Token token, EndpointContext ctx) {
         try {
             lock.writeLock().lock();
-            Key key = new Key(token);
-            org.eclipse.californium.core.observe.Observation obs = obsByToken.get(key);
+            org.eclipse.californium.core.observe.Observation obs = obsByToken.get(token);
             if (obs != null) {
-                obsByToken.put(key, new org.eclipse.californium.core.observe.Observation(obs.getRequest(), ctx));
+                obsByToken.put(token, new org.eclipse.californium.core.observe.Observation(obs.getRequest(), ctx));
             }
         } finally {
             lock.writeLock().unlock();
@@ -328,7 +348,7 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
     }
 
     @Override
-    public void remove(byte[] token) {
+    public void remove(Token token) {
         try {
             lock.writeLock().lock();
             unsafeRemoveObservation(token);
@@ -339,19 +359,18 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
 
     /* *************** Observation utility functions **************** */
 
-    private org.eclipse.californium.core.observe.Observation unsafeGetObservation(Key token) {
+    private org.eclipse.californium.core.observe.Observation unsafeGetObservation(Token token) {
         org.eclipse.californium.core.observe.Observation obs = obsByToken.get(token);
         return ObservationUtil.shallowClone(obs);
     }
 
-    private void unsafeRemoveObservation(byte[] observationId) {
-        Key kToken = new Key(observationId);
-        org.eclipse.californium.core.observe.Observation removed = obsByToken.remove(kToken);
+    private void unsafeRemoveObservation(Token observationId) {
+        org.eclipse.californium.core.observe.Observation removed = obsByToken.remove(observationId);
 
         if (removed != null) {
             String registrationId = ObserveUtil.extractRegistrationId(removed);
-            List<Key> tokens = tokensByRegId.get(registrationId);
-            tokens.remove(kToken);
+            Set<Token> tokens = tokensByRegId.get(registrationId);
+            tokens.remove(observationId);
             if (tokens.isEmpty()) {
                 tokensByRegId.remove(registrationId);
             }
@@ -360,9 +379,9 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
 
     private Collection<Observation> unsafeRemoveAllObservations(String registrationId) {
         Collection<Observation> removed = new ArrayList<>();
-        List<Key> tokens = tokensByRegId.get(registrationId);
+        Set<Token> tokens = tokensByRegId.get(registrationId);
         if (tokens != null) {
-            for (Key token : tokens) {
+            for (Token token : tokens) {
                 Observation observationRemoved = build(obsByToken.remove(token));
                 if (observationRemoved != null) {
                     removed.add(observationRemoved);
@@ -375,9 +394,9 @@ public class InMemoryRegistrationStore implements CaliforniumRegistrationStore,
 
     private Collection<Observation> unsafeGetObservations(String registrationId) {
         Collection<Observation> result = new ArrayList<>();
-        List<Key> tokens = tokensByRegId.get(registrationId);
+        Set<Token> tokens = tokensByRegId.get(registrationId);
         if (tokens != null) {
-            for (Key token : tokens) {
+            for (Token token : tokens) {
                 Observation obs = build(unsafeGetObservation(token));
                 if (obs != null) {
                     result.add(obs);
diff --git a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/ObservationServiceImpl.java b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/ObservationServiceImpl.java
index f3c9c216..9cecabe2 100644
--- a/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/ObservationServiceImpl.java
+++ b/leshan-server-cf/src/main/java/org/eclipse/leshan/server/californium/impl/ObservationServiceImpl.java
@@ -27,6 +27,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import org.eclipse.californium.core.coap.CoAP;
 import org.eclipse.californium.core.coap.Request;
 import org.eclipse.californium.core.coap.Response;
+import org.eclipse.californium.core.coap.Token;
 import org.eclipse.californium.core.network.Endpoint;
 import org.eclipse.californium.core.observe.NotificationListener;
 import org.eclipse.californium.core.observe.ObservationStore;
@@ -139,10 +140,11 @@ public class ObservationServiceImpl implements ObservationService, NotificationL
     }
 
     private void cancel(Observation observation) {
+        Token token = new Token(observation.getId());
         if (secureEndpoint != null)
-            secureEndpoint.cancelObservation(observation.getId());
+            secureEndpoint.cancelObservation(token);
         if (nonSecureEndpoint != null)
-            nonSecureEndpoint.cancelObservation(observation.getId());
+            nonSecureEndpoint.cancelObservation(token);
 
         for (ObservationListener listener : listeners) {
             listener.cancelled(observation);
@@ -205,10 +207,10 @@ public class ObservationServiceImpl implements ObservationService, NotificationL
         String regid = coapRequest.getUserContext().get(ObserveUtil.CTX_REGID);
 
         // get observation for this request
-        Observation observation = registrationStore.getObservation(regid, coapResponse.getToken());
+        Observation observation = registrationStore.getObservation(regid, coapResponse.getToken().getBytes());
         if (observation == null) {
             LOG.error("Unexpected error: Unable to find observation with token {} for registration {}",
-                    Hex.encodeHexString(coapResponse.getToken()), regid);
+                    coapResponse.getToken(), regid);
             return;
         }
 
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
index 2c8d3eb4..90ec2e55 100644
--- 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
@@ -14,6 +14,8 @@
  *     Sierra Wireless - initial API and implementation
  *     Achim Kraus (Bosch Software Innovations GmbH) - rename CorrelationContext to
  *                                                     EndpointContext
+ *     Achim Kraus (Bosch Software Innovations GmbH) - update to modified 
+ *                                                     ObservationStore API
  *******************************************************************************/
 package org.eclipse.leshan.server.cluster;
 
@@ -31,6 +33,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import org.eclipse.californium.core.coap.Token;
 import org.eclipse.californium.elements.EndpointContext;
 import org.eclipse.leshan.core.observation.Observation;
 import org.eclipse.leshan.server.Startable;
@@ -409,7 +412,7 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
             try {
                 lockValue = RedisLock.acquire(j, lockKey);
 
-                Observation observation = build(get(observationId));
+                Observation observation = build(get(new Token(observationId)));
                 if (observation != null && registrationId.equals(observation.getRegistrationId())) {
                     unsafeRemoveObservation(j, registrationId, observationId);
                     return observation;
@@ -424,7 +427,7 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
 
     @Override
     public Observation getObservation(String registrationId, byte[] observationId) {
-        return build(get(observationId));
+        return build(get(new Token(observationId)));
     }
 
     @Override
@@ -470,9 +473,19 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
     /* *************** Californium ObservationStore API **************** */
 
     @Override
-    public void add(org.eclipse.californium.core.observe.Observation obs) {
-        String endpoint = ObserveUtil.validateCoapObservation(obs);
+    public org.eclipse.californium.core.observe.Observation putIfAbsent(Token token, org.eclipse.californium.core.observe.Observation obs) {
+        return add(token, obs, true);
+    }
+
+    @Override
+    public org.eclipse.californium.core.observe.Observation put(Token token, org.eclipse.californium.core.observe.Observation obs) {
+        return add(token, obs, false);
+    }
 
+    private org.eclipse.californium.core.observe.Observation add(Token token, org.eclipse.californium.core.observe.Observation obs, boolean ifAbsent) {
+        String endpoint = ObserveUtil.validateCoapObservation(obs);
+        org.eclipse.californium.core.observe.Observation previousObservation = null;
+        
         try (Jedis j = pool.getResource()) {
             byte[] lockValue = null;
             byte[] lockKey = toKey(LOCK_EP, endpoint);
@@ -482,16 +495,26 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
                 String registrationId = ObserveUtil.extractRegistrationId(obs);
                 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));
+                byte[] key = toKey(OBS_TKN, obs.getRequest().getToken().getBytes());
+                byte[] serializeObs = serializeObs(obs);
+                byte[] previousValue = null;
+                if (ifAbsent) {
+                    previousValue = j.get(key);
+                    if (previousValue == null || previousValue.length == 0) {
+                        j.set(key, serializeObs);
+                    } else {
+                        return deserializeObs(previousValue);
+                    }
+                } else {
+                    previousValue = j.getSet(key, serializeObs);
+                }
 
                 // secondary index to get the list by registrationId
-                j.lpush(toKey(OBS_TKNS_REGID_IDX, registrationId), obs.getRequest().getToken());
+                j.lpush(toKey(OBS_TKNS_REGID_IDX, registrationId), obs.getRequest().getToken().getBytes());
 
                 // log any collisions
                 if (previousValue != null && previousValue.length != 0) {
-                    org.eclipse.californium.core.observe.Observation previousObservation = deserializeObs(
-                            previousValue);
+                    previousObservation = deserializeObs(previousValue);
                     LOG.warn(
                             "Token collision ? observation from request [{}] will be replaced by observation from request [{}] ",
                             previousObservation.getRequest(), obs.getRequest());
@@ -500,12 +523,13 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
                 RedisLock.release(j, lockKey, lockValue);
             }
         }
+        return previousObservation;
     }
 
     @Override
-    public void remove(byte[] token) {
+    public void remove(Token token) {
         try (Jedis j = pool.getResource()) {
-            byte[] tokenKey = toKey(OBS_TKN, token);
+            byte[] tokenKey = toKey(OBS_TKN, token.getBytes());
 
             // fetch the observation by token
             byte[] serializedObs = j.get(tokenKey);
@@ -527,7 +551,7 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
             try {
                 lockValue = RedisLock.acquire(j, lockKey);
 
-                unsafeRemoveObservation(j, registrationId, token);
+                unsafeRemoveObservation(j, registrationId, token.getBytes());
             } finally {
                 RedisLock.release(j, lockKey, lockValue);
             }
@@ -536,9 +560,9 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
     }
 
     @Override
-    public org.eclipse.californium.core.observe.Observation get(byte[] token) {
+    public org.eclipse.californium.core.observe.Observation get(Token token) {
         try (Jedis j = pool.getResource()) {
-            byte[] obs = j.get(toKey(OBS_TKN, token));
+            byte[] obs = j.get(toKey(OBS_TKN, token.getBytes()));
             if (obs == null) {
                 return null;
             } else {
@@ -586,7 +610,7 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta
     }
 
     @Override
-    public void setContext(byte[] token, EndpointContext correlationContext) {
+    public void setContext(Token token, EndpointContext correlationContext) {
         // TODO should be implemented
     }
 
diff --git a/pom.xml b/pom.xml
index 0c2f2e0b..f72a2699 100644
--- a/pom.xml
+++ b/pom.xml
@@ -105,7 +105,7 @@ Contributors:
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <californium.version>2.0.0-M6</californium.version>
+        <californium.version>2.0.0-M8</californium.version>
         <logback.version>1.2.3</logback.version>
         <slf4j.api.version>1.7.25</slf4j.api.version>
         <jetty.version>9.1.4.v20140401</jetty.version>


Test patch:
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 a70203de..0a3917f5 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
@@ -301,7 +301,9 @@ public class RegistrationTest {
         coapRequest.getOptions().addUriQuery("ep=" + helper.currentEndpointIdentifier);
 
         // send request
-        CoapEndpoint coapEndpoint = new CoapEndpoint(new InetSocketAddress(0));
+        CoapEndpoint.CoapEndpointBuilder builder = new CoapEndpoint.CoapEndpointBuilder();
+        builder.setInetSocketAddress(new InetSocketAddress(0));
+        CoapEndpoint coapEndpoint = builder.build();
         coapEndpoint.start();
         coapEndpoint.sendRequest(coapRequest);
 
diff --git a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/SecureIntegrationTestHelper.java b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/SecureIntegrationTestHelper.java
index 904611cf..09b91f47 100644
--- a/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/SecureIntegrationTestHelper.java
+++ b/leshan-integration-tests/src/test/java/org/eclipse/leshan/integration/tests/SecureIntegrationTestHelper.java
@@ -213,7 +213,10 @@ public class SecureIntegrationTestHelper extends IntegrationTestHelper {
         config.setIdentity(clientPrivateKey, clientPublicKey);
 
         CoapServer coapServer = new CoapServer();
-        coapServer.addEndpoint(new CoapEndpoint(new DTLSConnector(config.build()), NetworkConfig.getStandard()));
+        CoapEndpoint.CoapEndpointBuilder coapBuilder = new CoapEndpoint.CoapEndpointBuilder();
+        coapBuilder.setConnector(new DTLSConnector(config.build()));
+        coapBuilder.setNetworkConfig(new NetworkConfig());
+        coapServer.addEndpoint(coapBuilder.build());
 
         LeshanClientBuilder builder = new LeshanClientBuilder(getCurrentEndpoint());
         builder.setLocalAddress(clientAddress.getHostString(), clientAddress.getPort());
@@ -240,7 +243,10 @@ public class SecureIntegrationTestHelper extends IntegrationTestHelper {
         config.setTrustStore(trustedCertificates);
 
         CoapServer coapServer = new CoapServer();
-        coapServer.addEndpoint(new CoapEndpoint(new DTLSConnector(config.build()), NetworkConfig.getStandard()));
+        CoapEndpoint.CoapEndpointBuilder coapBuilder = new CoapEndpoint.CoapEndpointBuilder();
+        coapBuilder.setConnector(new DTLSConnector(config.build()));
+        coapBuilder.setNetworkConfig(new NetworkConfig());
+        coapServer.addEndpoint(coapBuilder.build());
 
         LeshanClientBuilder builder = new LeshanClientBuilder(getCurrentEndpoint());
         builder.setLocalAddress(clientAddress.getHostString(), clientAddress.getPort());
diff --git a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ObservationServiceTest.java b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ObservationServiceTest.java
index e9937a53..35a38dc0 100644
--- a/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ObservationServiceTest.java
+++ b/leshan-server-cf/src/test/java/org/eclipse/leshan/server/californium/impl/ObservationServiceTest.java
@@ -146,9 +146,9 @@ public class ObservationServiceTest {
                 registrationId, new ObserveRequest(target.toString()));
         coapRequest.setUserContext(context);
 
-        store.add(new org.eclipse.californium.core.observe.Observation(coapRequest, null));
+        store.put(coapRequest.getToken(), new org.eclipse.californium.core.observe.Observation(coapRequest, null));
 
-        Observation observation = new Observation(coapRequest.getToken(), registrationId, target, null);
+        Observation observation = new Observation(coapRequest.getToken().getBytes(), registrationId, target, null);
         observationService.addObservation(registration, observation);
 
         return observation;