RedissonConcurrentMap merged with RedissonMap

src/main/java/org/redisson/Redisson.java

@@ -28,14 +28,17 @@ public class Redisson {
     }
 
     public <K, V> Map<K, V> getMap(String name) {
-        RedisConnection<Object, Object> connection = redisClient.connect(codec);
-        return new RedissonMap<K, V>(connection, name);
+        return new RedissonMap<K, V>(this, connect(), name);
+    }
+
+    RedisConnection<Object, Object> connect() {
+        return redisClient.connect(codec);
     }
 
     public Lock getLock(String name) {
         RedissonLock lock = locksMap.get(name);
         if (lock == null) {
-            RedisConnection<Object, Object> connection = redisClient.connect(codec);
+            RedisConnection<Object, Object> connection = connect();
             RedisPubSubConnection<Object, Object> pubSubConnection = redisClient.connectPubSub(codec);
 
             lock = new RedissonLock(pubSubConnection, connection, name);

src/main/java/org/redisson/RedissonConcurrentMap.java

@@ -1,80 +0,0 @@
-package org.redisson;
-
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.locks.Lock;
-
-import com.lambdaworks.redis.RedisConnection;
-
-public class RedissonConcurrentMap<K, V> extends RedissonMap<K, V> implements ConcurrentMap<K, V> {
-
-    private final Redisson redisson;
-
-    RedissonConcurrentMap(Redisson redisson, RedisConnection<Object, Object> connection, String name) {
-        super(connection, name);
-        this.redisson = redisson;
-    }
-
-    @Override
-    public V putIfAbsent(K key, V value) {
-        while (true) {
-            Boolean res = getConnection().hsetnx(getName(), key, value);
-            if (!res) {
-                V result = get(key);
-                if (result != null) {
-                    return result;
-                }
-            } else {
-                return null;
-            }
-        }
-    }
-
-    @Override
-    public boolean remove(Object key, Object value) {
-        // TODO use murmur-hashing as lock key table
-        Lock lock = redisson.getLock(getName() + "__" + key);
-        lock.lock();
-        try {
-            if (containsKey(key) && get(key).equals(value)) {
-                remove(key);
-                return true;
-            }
-            return false;
-        } finally {
-            lock.unlock();
-        }
-    }
-
-    @Override
-    public boolean replace(K key, V oldValue, V newValue) {
-        // TODO user murmur-hashing as lock key table
-        Lock lock = redisson.getLock(getName() + "__" + key);
-        lock.lock();
-        try {
-            if (containsKey(key) && get(key).equals(oldValue)) {
-                getConnection().hset(getName(), key, newValue);
-                return true;
-            } else {
-                return false;
-            }
-        } finally {
-            lock.unlock();
-        }
-    }
-
-    @Override
-    public V replace(K key, V value) {
-        // TODO use murmur-hashing as lock key table
-        Lock lock = redisson.getLock(getName() + "__" + key);
-        lock.lock();
-        try {
-            if (containsKey(key)) {
-                return put(key, value);
-            }
-            return null;
-        } finally {
-            lock.unlock();
-        }
-    }
-
-}

src/main/java/org/redisson/RedissonMap.java

@@ -5,15 +5,19 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
 
 import com.lambdaworks.redis.RedisConnection;
 
-public class RedissonMap<K, V> implements Map<K, V> {
+//TODO make keys watching instead of map name
+public class RedissonMap<K, V> implements ConcurrentMap<K, V> {
 
     private final RedisConnection<Object, Object> connection;
     private final String name;
+    private final Redisson redisson;
 
-    RedissonMap(RedisConnection<Object, Object> connection, String name) {
+    RedissonMap(Redisson redisson, RedisConnection<Object, Object> connection, String name) {
+        this.redisson = redisson;
         this.connection = connection;
         this.name = name;
     }
@@ -96,4 +100,83 @@ public class RedissonMap<K, V> implements Map<K, V> {
         return result.entrySet();
     }
 
+    @Override
+    public V putIfAbsent(K key, V value) {
+        while (true) {
+            Boolean res = getConnection().hsetnx(getName(), key, value);
+            if (!res) {
+                V result = get(key);
+                if (result != null) {
+                    return result;
+                }
+            } else {
+                return null;
+            }
+        }
+    }
+
+    @Override
+    public boolean remove(Object key, Object value) {
+        RedisConnection<Object, Object> connection = redisson.connect();
+        try {
+            while (true) {
+                connection.watch(getName());
+                if (connection.hexists(getName(), key)
+                        && connection.hget(getName(), key).equals(value)) {
+                    connection.multi();
+                    connection.hdel(getName(), key);
+                    if (connection.exec().size() == 1) {
+                        return true;
+                    }
+                } else {
+                    return false;
+                }
+            }
+        } finally {
+            connection.close();
+        }
+    }
+
+    @Override
+    public boolean replace(K key, V oldValue, V newValue) {
+        RedisConnection<Object, Object> connection = redisson.connect();
+        try {
+            while (true) {
+                connection.watch(getName());
+                if (connection.hexists(getName(), key)
+                        && connection.hget(getName(), key).equals(oldValue)) {
+                    connection.multi();
+                    connection.hset(getName(), key, newValue);
+                    if (connection.exec().size() == 1) {
+                        return true;
+                    }
+                } else {
+                    return false;
+                }
+            }
+        } finally {
+            connection.close();
+        }
+    }
+
+    @Override
+    public V replace(K key, V value) {
+        RedisConnection<Object, Object> connection = redisson.connect();
+        try {
+            while (true) {
+                connection.watch(getName());
+                if (connection.hexists(getName(), key)) {
+                    V prev = (V) connection.hget(getName(), key);
+                    connection.hset(getName(), key, value);
+                    if (connection.exec().size() == 1) {
+                        return prev;
+                    }
+                }
+                return null;
+            }
+        } finally {
+            connection.close();
+        }
+    }
+
 }