redis building exception added; redis cluster builder implemented; re…

…dis sentinel builder made resettable; redis server builder made resettable; basic sentinel test;

src/main/java/redis/embedded/RedisCluster.java

@@ -12,7 +12,7 @@ public class RedisCluster implements Redis {
     private final List<Redis> sentinels = new LinkedList<>();
     private final List<Redis> servers = new LinkedList<>();
 
-    public RedisCluster(List<Redis> sentinels, List<Redis> servers) {
+    RedisCluster(List<Redis> sentinels, List<Redis> servers) {
         this.servers.addAll(servers);
         this.sentinels.addAll(sentinels);
     }
@@ -33,4 +33,8 @@ public void stop() throws EmbeddedRedisException {
         servers.parallelStream().forEach(Redis::stop);
         sentinels.parallelStream().forEach(Redis::stop);
     }
+
+    public static RedisClusterBuilder builder() {
+        return new RedisClusterBuilder();
+    }
 }

src/main/java/redis/embedded/RedisClusterBuilder.java

@@ -0,0 +1,122 @@
+package redis.embedded;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Created by piotrturek on 22/01/15.
+ */
+public class RedisClusterBuilder {
+    private RedisSentinelBuilder sentinelBuilder = new RedisSentinelBuilder();
+    private RedisServerBuilder serverBuilder = new RedisServerBuilder();
+    private int sentinelCount = 1;
+    private int quorumSize = 1;
+    private int currentSentinelPort = 26379;
+    private int currentReplicationGroupPort = 6379;
+    private final List<ReplicationGroup> groups = new LinkedList<>();
+
+    public RedisClusterBuilder withSentinelBuilder(RedisSentinelBuilder sentinelBuilder) {
+        this.sentinelBuilder = sentinelBuilder;
+        return this;
+    }
+
+    public RedisClusterBuilder withServerBuilder(RedisServerBuilder serverBuilder) {
+        this.serverBuilder = serverBuilder;
+        return this;
+    }
+
+    public RedisClusterBuilder sentinelCount(int sentinelCount) {
+        this.sentinelCount = sentinelCount;
+        return this;
+    }
+
+    public RedisClusterBuilder quorumSize(int quorumSize) {
+        this.quorumSize = quorumSize;
+        return this;
+    }
+
+    public RedisClusterBuilder replicationGroup(String masterName, int slaveCount) {
+        this.groups.add(new ReplicationGroup(masterName, slaveCount, nextReplicationGroupPort(slaveCount)));
+        return this;
+    }
+
+    public RedisCluster build() {
+        final List<Redis> sentinels = buildSentinels();
+        final List<Redis> servers = buildServers();
+        return new RedisCluster(sentinels, servers);
+    }
+
+    private List<Redis> buildServers() {
+        return groups.stream().flatMap(g -> {
+            final Stream.Builder<Redis> builder = Stream.builder();
+            builder.accept(buildMaster(g));
+            buildSlaves(builder, g);
+            return builder.build();
+        }).collect(Collectors.toList());
+    }
+
+    private void buildSlaves(Stream.Builder<Redis> builder, ReplicationGroup g) {
+        for (Integer slavePort : g.slavePorts) {
+            serverBuilder.reset();
+            serverBuilder.port(slavePort);
+            serverBuilder.slaveOf("localhost", g.masterPort);
+            final RedisServer slave = serverBuilder.build();
+            builder.accept(slave);
+        }
+    }
+
+    private Redis buildMaster(ReplicationGroup g) {
+        serverBuilder.reset();
+        return serverBuilder.port(g.masterPort).build();
+    }
+
+    private List<Redis> buildSentinels() {
+        int toBuild = this.sentinelCount;
+        final List<Redis> sentinels = new LinkedList<>();
+        while (toBuild-- > 0) {
+            sentinels.add(buildSentinel());
+        }
+        return sentinels;
+    }
+
+    private Redis buildSentinel() {
+        sentinelBuilder.reset();
+        sentinelBuilder.port(nextSentinelPort());
+        groups.stream().forEach(g -> {
+            sentinelBuilder.masterName(g.masterName);
+            sentinelBuilder.masterPort(g.masterPort);
+            sentinelBuilder.quorumSize(quorumSize);
+            sentinelBuilder.addDefaultReplicationGroup();
+        });
+        return sentinelBuilder.build();
+    }
+
+    private int nextSentinelPort() {
+        return currentSentinelPort++;
+    }
+
+    private int nextReplicationGroupPort(int slaveCount) {
+        final int toReturn = this.currentReplicationGroupPort;
+        currentReplicationGroupPort += slaveCount + 1;
+        return toReturn;
+    }
+
+    private static class ReplicationGroup {
+        private final int slaveCount;
+        private final String masterName;
+        private final int masterPort;
+        private final List<Integer> slavePorts = new LinkedList<>();
+
+        private ReplicationGroup(String masterName, int slaveCount, int portsStart) {
+            this.masterName = masterName;
+            this.slaveCount = slaveCount;
+            masterPort = portsStart;
+            while (slaveCount-- > 0) {
+                portsStart += 1;
+                slavePorts.add(portsStart);
+            }
+        }
+    }
+}

src/main/java/redis/embedded/RedisSentinelBuilder.java

@@ -2,6 +2,7 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.io.Files;
+import redis.embedded.exceptions.RedisBuildingException;
 import redis.embedded.util.JarUtil;
 
 import java.io.File;
@@ -16,18 +17,20 @@
 public class RedisSentinelBuilder {
     private static final String LINE_SEPARATOR = System.getProperty("line.separator");
     private static final String CONF_FILENAME = "embedded-redis-sentinel";
-    private static final String MASTER_MONITOR_LINE = "sentinel monitor %s 127.0.0.1 %d 1";
+    private static final String MASTER_MONITOR_LINE = "sentinel monitor %s 127.0.0.1 %d %d";
     private static final String DOWN_AFTER_LINE = "sentinel down-after-milliseconds %s %d";
     private static final String FAILOVER_LINE = "sentinel failover-timeout %s %d";
     private static final String PARALLEL_SYNCS_LINE = "sentinel parallel-syncs %s %d";
+    private static final String PORT_LINE = "port %d";
 
     private File executable;
-    private Integer port;
+    private Integer port = 26379;
     private int masterPort = 6379;
     private String masterName = "mymaster";
     private long downAfterMilliseconds = 60000L;
     private long failoverTimeout = 180000L;
     private int parallelSyncs = 1;
+    private int quorumSize = 1;
     private String sentinelConf;
 
     private StringBuilder redisConfigBuilder;
@@ -57,6 +60,11 @@ public RedisSentinelBuilder masterName(String masterName) {
         return this;
     }
 
+    public RedisSentinelBuilder quorumSize(int quorumSize) {
+        this.quorumSize = quorumSize;
+        return this;
+    }
+
     public RedisSentinelBuilder downAfterMilliseconds(Long downAfterMilliseconds) {
         this.downAfterMilliseconds = downAfterMilliseconds;
         return this;
@@ -94,35 +102,51 @@ public RedisSentinelBuilder setting(String configLine) {
         return this;
     }
 
-    public RedisSentinel build() throws IOException {
-        if (sentinelConf == null) {
-            resolveSentinelConf();
-        }
-        if (executable == null) {
-            executable = JarUtil.extractExecutableFromJar(RedisRunScriptEnum.getRedisRunScript());
-        }
-
+    public RedisSentinel build() {
+        tryResolveConfAndExec();
         List<String> args = buildCommandArgs();
         return new RedisSentinel(args);
     }
 
+    private void tryResolveConfAndExec() {
+        try {
+            if (sentinelConf == null) {
+                resolveSentinelConf();
+            }
+            if (executable == null) {
+                executable = JarUtil.extractExecutableFromJar(RedisRunScriptEnum.getRedisRunScript());
+            }
+        } catch (IOException e) {
+            throw new RedisBuildingException("Could not build sentinel instance", e);
+        }
+    }
+
+    public void reset() {
+        this.redisConfigBuilder = null;
+    }
+
+    public void addDefaultReplicationGroup() {
+        setting(String.format(MASTER_MONITOR_LINE, masterName, masterPort, quorumSize));
+        setting(String.format(DOWN_AFTER_LINE, masterName, downAfterMilliseconds));
+        setting(String.format(FAILOVER_LINE, masterName, failoverTimeout));
+        setting(String.format(PARALLEL_SYNCS_LINE, masterName, parallelSyncs));
+    }
+
     private void resolveSentinelConf() throws IOException {
         if (redisConfigBuilder == null) {
-            useDefaultConfig();
+            addDefaultReplicationGroup();
         }
+        setting(String.format(PORT_LINE, port));
         final String configString = redisConfigBuilder.toString();
 
-        File redisConfigFile = File.createTempFile(CONF_FILENAME, ".conf");
+        File redisConfigFile = File.createTempFile(resolveConfigName(), ".conf");
         redisConfigFile.deleteOnExit();
         Files.write(configString, redisConfigFile, Charset.forName("UTF-8"));
         sentinelConf = redisConfigFile.getAbsolutePath();
     }
 
-    private void useDefaultConfig() {
-        setting(String.format(MASTER_MONITOR_LINE, masterName, masterPort));
-        setting(String.format(DOWN_AFTER_LINE, masterName, downAfterMilliseconds));
-        setting(String.format(FAILOVER_LINE, masterName, failoverTimeout));
-        setting(String.format(PARALLEL_SYNCS_LINE, masterName, parallelSyncs));
+    private String resolveConfigName() {
+        return CONF_FILENAME + "_" + port;
     }
 
     private List<String> buildCommandArgs() {

src/main/java/redis/embedded/RedisServerBuilder.java

@@ -2,6 +2,7 @@
 
 import com.google.common.base.Strings;
 import com.google.common.io.Files;
+import redis.embedded.exceptions.RedisBuildingException;
 import redis.embedded.util.JarUtil;
 
 import java.io.File;
@@ -72,9 +73,28 @@ public RedisServerBuilder setting(String configLine) {
         return this;
     }
 
-    public RedisServer build() throws IOException {
+    public RedisServer build() {
+        tryResolveConfAndExec();
+        List<String> args = buildCommandArgs();
+        return new RedisServer(args);
+    }
+
+    public void reset() {
+        this.redisConfigBuilder = null;
+        this.slaveOf = null;
+    }
+
+    private void tryResolveConfAndExec() {
+        try {
+            resolveConfAndExec();
+        } catch (IOException e) {
+            throw new RedisBuildingException("Could not build server instance", e);
+        }
+    }
+
+    private void resolveConfAndExec() throws IOException {
         if (redisConf == null && redisConfigBuilder != null) {
-            File redisConfigFile = File.createTempFile(CONF_FILENAME, ".conf");
+            File redisConfigFile = File.createTempFile(resolveConfigName(), ".conf");
             redisConfigFile.deleteOnExit();
             Files.write(redisConfigBuilder.toString(), redisConfigFile, Charset.forName("UTF-8"));
             redisConf = redisConfigFile.getAbsolutePath();
@@ -83,16 +103,14 @@ public RedisServer build() throws IOException {
         if (executable == null) {
             executable = JarUtil.extractExecutableFromJar(RedisRunScriptEnum.getRedisRunScript());
         }
+    }
 
-        List<String> args = buildCommandArgs();
-        return new RedisServer(args);
+    private String resolveConfigName() {
+        return CONF_FILENAME + "_" + port;
     }
 
-    private List<String> buildCommandArgs() throws IOException {
+    private List<String> buildCommandArgs() {
         List<String> args = new ArrayList<>();
-        if (executable == null) {
-            executable = JarUtil.extractExecutableFromJar(RedisRunScriptEnum.getRedisRunScript());
-        }
         args.add(executable.getAbsolutePath());
 
         if (!Strings.isNullOrEmpty(redisConf)) {

src/main/java/redis/embedded/exceptions/RedisBuildingException.java

@@ -0,0 +1,10 @@
+package redis.embedded.exceptions;
+
+/**
+ * Created by piotrturek on 26/01/15.
+ */
+public class RedisBuildingException extends RuntimeException {
+    public RedisBuildingException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/test/java/redis/embedded/RedisSentinelTest.java

@@ -2,6 +2,8 @@
 
 import org.junit.Test;
 
+import java.util.concurrent.TimeUnit;
+
 public class RedisSentinelTest {
     private RedisSentinel sentinel;
     private RedisServer server;
@@ -12,7 +14,7 @@ public void testSimpleRun() throws Exception {
         sentinel = RedisSentinel.builder().build();
         sentinel.start();
         server.start();
-        Thread.sleep(1000L);
+        TimeUnit.SECONDS.sleep(1);
         server.stop();
         sentinel.stop();
     }