Merge branch 'master' of github.com:redisson/redisson

redisson/src/main/java/org/redisson/RedissonMapCache.java

@@ -82,6 +82,7 @@ public class RedissonMapCache<K, V> extends RedissonMap<K, V> implements RMapCac
     private static final RedisCommand<Boolean> EVAL_REMOVE_VALUE = new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 5, ValueType.MAP);
     private static final RedisCommand<Object> EVAL_PUT_TTL = new RedisCommand<Object>("EVAL", 9, ValueType.MAP, ValueType.MAP_VALUE);
     private static final RedisCommand<Boolean> EVAL_FAST_PUT_TTL = new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 9, ValueType.MAP, ValueType.MAP_VALUE);
+    private static final RedisCommand<Boolean> EVAL_FAST_PUT_TTL_IF_ABSENT = new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 10, ValueType.MAP, ValueType.MAP_VALUE);
     private static final RedisCommand<Object> EVAL_GET_TTL = new RedisCommand<Object>("EVAL", 7, ValueType.MAP_KEY, ValueType.MAP_VALUE);
     private static final RedisCommand<Boolean> EVAL_CONTAINS_KEY = new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 7, ValueType.MAP_KEY);
     static final RedisCommand<Boolean> EVAL_CONTAINS_VALUE = new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 7, ValueType.MAP_VALUE);
@@ -687,6 +688,99 @@ public class RedissonMapCache<K, V> extends RedissonMap<K, V> implements RMapCac
              Arrays.<Object>asList(getName(key), getTimeoutSetNameByKey(key), getIdleSetNameByKey(key)), System.currentTimeMillis(), key, value);
     }
     
+    @Override
+	public boolean fastPutIfAbsent(K key, V value, long ttl, TimeUnit ttlUnit) {
+		return fastPutIfAbsent(key, value, ttl, ttlUnit, 0, null);
+	}
+    
+    @Override
+    public boolean fastPutIfAbsent(K key, V value, long ttl, TimeUnit ttlUnit, long maxIdleTime, TimeUnit maxIdleUnit) {
+    	return get(fastPutIfAbsentAsync(key, value, ttl, ttlUnit, maxIdleTime, maxIdleUnit));
+    }
+	
+    @Override
+	public RFuture<Boolean> fastPutIfAbsentAsync(K key, V value, long ttl, TimeUnit ttlUnit, long maxIdleTime, TimeUnit maxIdleUnit) {
+		if (ttl < 0) {
+            throw new IllegalArgumentException("ttl can't be negative");
+        }
+        if (maxIdleTime < 0) {
+            throw new IllegalArgumentException("maxIdleTime can't be negative");
+        }
+        if (ttl == 0 && maxIdleTime == 0) {
+            return fastPutIfAbsentAsync(key, value);
+        }
+
+        if (ttl > 0 && ttlUnit == null) {
+            throw new NullPointerException("ttlUnit param can't be null");
+        }
+
+        if (maxIdleTime > 0 && maxIdleUnit == null) {
+            throw new NullPointerException("maxIdleUnit param can't be null");
+        }
+
+        long ttlTimeout = 0;
+        if (ttl > 0) {
+            ttlTimeout = System.currentTimeMillis() + ttlUnit.toMillis(ttl);
+        }
+
+        long maxIdleTimeout = 0;
+        long maxIdleDelta = 0;
+        if (maxIdleTime > 0) {
+            maxIdleDelta = maxIdleUnit.toMillis(maxIdleTime);
+            maxIdleTimeout = System.currentTimeMillis() + maxIdleDelta;
+        }
+
+        return commandExecutor.evalWriteAsync(getName(key), codec, EVAL_FAST_PUT_TTL_IF_ABSENT,
+            "local insertable = false; "
+            + "local value = redis.call('hget', KEYS[1], ARGV[5]); "
+            + "if value == false then "
+            	+ "insertable = true; "
+        	+ "else "
+        		+ "if insertable == false then "
+	        		+ "local t, val = struct.unpack('dLc0', value); "
+	                + "local expireDate = 92233720368547758; "
+	                + "local expireDateScore = redis.call('zscore', KEYS[2], ARGV[5]); "
+	                + "if expireDateScore ~= false then "
+	                    + "expireDate = tonumber(expireDateScore) "
+	                + "end; "
+	                + "if t ~= 0 then "
+	                    + "local expireIdle = redis.call('zscore', KEYS[3], ARGV[5]); "
+	                    + "if expireIdle ~= false then "
+	                        + "expireDate = math.min(expireDate, tonumber(expireIdle)) "
+	                    + "end; "
+	                + "end; "
+	                + "if expireDate <= tonumber(ARGV[1]) then "
+	                    + "insertable = true; "
+	                + "end; "        	
+        		+ "end; "
+			+ "end; "
+        		
+			+ "if insertable == true then "
+				// ttl
+				+ "if tonumber(ARGV[2]) > 0 then "
+					+ "redis.call('zadd', KEYS[2], ARGV[2], ARGV[5]); "
+				+ "else "
+					+ "redis.call('zrem', KEYS[2], ARGV[5]); "
+				+ "end; "
+					
+				// idle
+				+ "if tonumber(ARGV[3]) > 0 then "
+					+ "redis.call('zadd', KEYS[3], ARGV[3], ARGV[5]); "
+				+ "else "
+					+ "redis.call('zrem', KEYS[3], ARGV[5]); "
+				+ "end; "
+				
+				// value
+				+ "local val = struct.pack('dLc0', ARGV[4], string.len(ARGV[6]), ARGV[6]); "
+				+ "redis.call('hset', KEYS[1], ARGV[5], val); "
+				
+				+ "return 1; "
+			+ "else "
+				+ "return 0; "
+			+ "end; ",
+			Arrays.<Object>asList(getName(key), getTimeoutSetNameByKey(key), getIdleSetNameByKey(key)), System.currentTimeMillis(), ttlTimeout, maxIdleTimeout, maxIdleDelta, key, value);
+	}
+
     @Override
     public RFuture<Boolean> replaceAsync(K key, V oldValue, V newValue) {
         return commandExecutor.evalWriteAsync(getName(key), codec, EVAL_REPLACE_VALUE,
@@ -912,5 +1006,4 @@ public class RedissonMapCache<K, V> extends RedissonMap<K, V> implements RMapCac
              "return result;",
              Arrays.<Object>asList(getName(), getTimeoutSetName(), getIdleSetName()), System.currentTimeMillis());
     }
-    
 }

redisson/src/main/java/org/redisson/api/RMapCache.java

@@ -162,6 +162,53 @@ public interface RMapCache<K, V> extends RMap<K, V>, RMapCacheAsync<K, V> {
      */
     boolean fastPut(K key, V value, long ttl, TimeUnit ttlUnit, long maxIdleTime, TimeUnit maxIdleUnit);
     
+    /**
+     * If the specified key is not already associated
+     * with a value, associate it with the given value.
+     * <p>
+     * Stores value mapped by key with specified time to live.
+     * Entry expires after specified time to live.
+     * <p>
+     * Works faster than usual {@link #putIfAbsent(Object, Object, long, TimeUnit)}
+     * as it not returns previous value.
+     *
+     * @param key - map key
+     * @param value - map value
+     * @param ttl - time to live for key\value entry.
+     *              If <code>0</code> then stores infinitely.
+     * @param ttlUnit - time unit
+     * @return <code>true</code> if key is a new key in the hash and value was set.
+     *         <code>false</code> if key already exists in the hash
+     */
+    boolean fastPutIfAbsent(K key, V value, long ttl, TimeUnit ttlUnit);
+
+    /**
+     * If the specified key is not already associated
+     * with a value, associate it with the given value.
+     * <p>
+     * Stores value mapped by key with specified time to live and max idle time.
+     * Entry expires when specified time to live or max idle time has expired.
+     * <p>
+     * Works faster than usual {@link #putIfAbsent(Object, Object, long, TimeUnit, long, TimeUnit)}
+     * as it not returns previous value.
+     * 
+     * @param key - map key
+     * @param value - map value
+     * @param ttl - time to live for key\value entry.
+     *              If <code>0</code> then time to live doesn't affect entry expiration.
+     * @param ttlUnit - time unit
+     * @param maxIdleTime - max idle time for key\value entry.
+     *              If <code>0</code> then max idle time doesn't affect entry expiration.
+     * @param maxIdleUnit - time unit
+     * <p>
+     * if <code>maxIdleTime</code> and <code>ttl</code> params are equal to <code>0</code>
+     * then entry stores infinitely.
+     *
+     * @return <code>true</code> if key is a new key in the hash and value was set.
+     *         <code>false</code> if key already exists in the hash.
+     */
+    boolean fastPutIfAbsent(K key, V value, long ttl, TimeUnit ttlUnit, long maxIdleTime, TimeUnit maxIdleUnit);
+
     /**
      * Returns the number of entries in cache.
      * This number can reflects expired entries too

redisson/src/main/java/org/redisson/api/RMapCacheAsync.java

@@ -164,6 +164,32 @@ public interface RMapCacheAsync<K, V> extends RMapAsync<K, V> {
      */
     RFuture<Boolean> fastPutAsync(K key, V value, long ttl, TimeUnit ttlUnit, long maxIdleTime, TimeUnit maxIdleUnit);
     
+    /**
+     * If the specified key is not already associated
+     * with a value, associate it with the given value.
+     * <p>
+     * Stores value mapped by key with specified time to live and max idle time.
+     * Entry expires when specified time to live or max idle time has expired.
+     * <p>
+     * Works faster than usual {@link #putIfAbsentAsync(Object, Object, long, TimeUnit, long, TimeUnit)}
+     * as it not returns previous value.
+     *
+     * @param key - map key
+     * @param value - map value
+     * @param ttl - time to live for key\value entry.
+     *              If <code>0</code> then time to live doesn't affect entry expiration.
+     * @param ttlUnit - time unit
+     * @param maxIdleTime - max idle time for key\value entry.
+     *              If <code>0</code> then max idle time doesn't affect entry expiration.
+     * @param maxIdleUnit - time unit
+     * <p>
+     * if <code>maxIdleTime</code> and <code>ttl</code> params are equal to <code>0</code>
+     * then entry stores infinitely.
+     *
+     * @return previous associated value
+     */
+    RFuture<Boolean> fastPutIfAbsentAsync(K key, V value, long ttl, TimeUnit ttlUnit, long maxIdleTime, TimeUnit maxIdleUnit);
+
     /**
      * Returns the number of entries in cache.
      * This number can reflects expired entries too

redisson/src/main/resources/META-INF/spring.schemas

@@ -1,2 +1,2 @@
-http\://redisson.org/schema/redisson.xsd=org/redisson/spring/support/redisson-1.0.xsd
-http\://redisson.org/schema/redisson-1.0.xsd=org/redisson/spring/support/redisson-1.0.xsd
+http\://redisson.org/schema/redisson/redisson.xsd=org/redisson/spring/support/redisson-1.0.xsd
+http\://redisson.org/schema/redisson/redisson-1.0.xsd=org/redisson/spring/support/redisson-1.0.xsd

redisson/src/test/java/org/redisson/ClusterRunner.java

@@ -17,9 +17,10 @@ import java.util.List;
 public class ClusterRunner {
     
     private final LinkedHashMap<RedisRunner, String> nodes = new LinkedHashMap<>();
+    private final LinkedHashMap<String, String> masters = new LinkedHashMap<>();
     
     public ClusterRunner addNode(RedisRunner runner) {
-        nodes.put(runner, getRandomId());
+        nodes.putIfAbsent(runner, getRandomId());
         if (!runner.hasOption(RedisRunner.REDIS_OPTIONS.CLUSTER_ENABLED)) {
             runner.clusterEnabled(true);
         }
@@ -36,11 +37,20 @@ public class ClusterRunner {
         return this;
     }
     
+    public ClusterRunner addNode(RedisRunner master, RedisRunner... slaves) {
+        addNode(master);
+        for (RedisRunner slave : slaves) {
+            addNode(slave);
+            masters.put(nodes.get(slave), nodes.get(master));
+        }
+        return this;
+    }
+    
     public List<RedisRunner.RedisProcess> run() throws IOException, InterruptedException, RedisRunner.FailedToStartRedisException {
         ArrayList<RedisRunner.RedisProcess> processes = new ArrayList<>();
         for (RedisRunner runner : nodes.keySet()) {
             List<String> options = getClusterConfig(runner);
-            String confFile = runner.defaultDir() + File.pathSeparator + nodes.get(runner) + ".conf";
+            String confFile = runner.dir() + File.separatorChar + nodes.get(runner) + ".conf";
             System.out.println("WRITING CONFIG: for " + nodes.get(runner));
             try (PrintWriter printer = new PrintWriter(new FileWriter(confFile))) {
                 options.stream().forEach((line) -> {
@@ -64,13 +74,20 @@ public class ClusterRunner {
         List<String> nodeConfig = new ArrayList<>();
         int c = 0;
         for (RedisRunner node : nodes.keySet()) {
+            String nodeId = nodes.get(node);
             StringBuilder sb = new StringBuilder();
             String nodeAddr = node.getInitialBindAddr() + ":" + node.getPort();
-            sb.append(nodes.get(node)).append(" ");
+            sb.append(nodeId).append(" ");
             sb.append(nodeAddr).append(" ");
             sb.append(me.equals(nodeAddr)
                     ? "myself,"
-                    : "").append("master -").append(" ");
+                    : "");
+            if (!masters.containsKey(nodeId)) {
+                 sb.append("master -");
+            } else {
+                sb.append("slave ").append(masters.get(nodeId));
+            }
+            sb.append(" ");
             sb.append("0").append(" ");
             sb.append(me.equals(nodeAddr)
                     ? "0"

redisson/src/test/java/org/redisson/RedisRunner.java

@@ -191,6 +191,7 @@ public class RedisRunner {
     protected static RedisRunner.RedisProcess defaultRedisInstance;
     private static int defaultRedisInstanceExitCode;
 
+    private String path = "";
     private String defaultDir = Paths.get("").toString();
     private boolean nosave = false;
     private boolean randomDir = false;
@@ -459,6 +460,7 @@ public class RedisRunner {
     public RedisRunner dir(String dir) {
         if (!randomDir) {
             addConfigOption(REDIS_OPTIONS.DIR, dir);
+            this.path = dir;
         }
         return this;
     }
@@ -825,6 +827,10 @@ public class RedisRunner {
         return this.defaultDir;
     }
     
+    public String dir() {
+        return this.path;
+    }
+
     public String getInitialBindAddr() {
         return bindAddr.size() > 0 ? bindAddr.get(0) : "localhost";
     }
@@ -849,7 +855,7 @@ public class RedisRunner {
 
     public boolean deleteClusterFile() {
         File f = new File(clusterFile);
-        if (f.exists()) {
+        if (f.exists() && isRandomDir()) {
             System.out.println("REDIS RUNNER: Deleting cluster config file " + f.getAbsolutePath());
             return f.delete();
         }

redisson/src/test/java/org/redisson/RedissonMapCacheTest.java

@@ -838,4 +838,29 @@ public class RedissonMapCacheTest extends BaseTest {
             this.testField = testField;
         }
     }
+    
+    @Test
+    public void testFastPutIfAbsentWithTTL() throws Exception {
+        RMapCache<SimpleKey, SimpleValue> map = redisson.getMapCache("simpleTTL");
+        SimpleKey key = new SimpleKey("1");
+        SimpleValue value = new SimpleValue("2");
+        map.fastPutIfAbsent(key, value, 1, TimeUnit.SECONDS);
+        assertThat(map.fastPutIfAbsent(key, new SimpleValue("3"), 1, TimeUnit.SECONDS)).isFalse();
+        assertThat(map.get(key)).isEqualTo(value);
+        
+        Thread.sleep(1100);
+        
+        assertThat(map.fastPutIfAbsent(key, new SimpleValue("3"), 1, TimeUnit.SECONDS)).isTrue();
+        assertThat(map.get(key)).isEqualTo(new SimpleValue("3"));
+        
+        assertThat(map.fastPutIfAbsent(key, new SimpleValue("4"), 1, TimeUnit.SECONDS)).isFalse();
+        assertThat(map.get(key)).isEqualTo(new SimpleValue("3"));
+        
+        Thread.sleep(1100);
+        assertThat(map.fastPutIfAbsent(key, new SimpleValue("4"), 1, TimeUnit.SECONDS, 500, TimeUnit.MILLISECONDS)).isTrue();
+        
+        Thread.sleep(550);
+        assertThat(map.fastPutIfAbsent(key, new SimpleValue("5"), 1, TimeUnit.SECONDS, 500, TimeUnit.MILLISECONDS)).isTrue();
+
+    }
 }