Merge pull request kstyrc#39 from rwinch/jdk6

Support JDK 6

README.md

@@ -165,12 +165,16 @@ Contributors
  * anthonyu ([@anthonyu](http://github.com/anthonyu))
  * Artem Orobets ([@enisher](http://github.com/enisher))
  * Sean Simonsen ([@SeanSimonsen](http://github.com/SeanSimonsen))
+ * Rob Winch ([@rwinch](http://github.com/rwinch))
 
 
 Changelog
 ==============
 
-### 0.5 
+### 0.6
+ * Support JDK 6 +
+
+### 0.5
  * OS detection fix
  * redis binary per OS/arch pair
  * Updated to 2.8.19 binary for Windows

pom.xml

@@ -100,9 +100,9 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>2.3.2</version>
                 <configuration>
-                    <compilerVersion>1.8</compilerVersion>
-                    <source>1.8</source>
-                    <target>1.8</target>
+                    <compilerVersion>1.6</compilerVersion>
+                    <source>1.6</source>
+                    <target>1.6</target>
                 </configuration>
             </plugin>
 

src/main/java/redis/embedded/AbstractRedisInstance.java

@@ -9,6 +9,8 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
+import org.apache.commons.io.IOUtils;
+
 abstract class AbstractRedisInstance implements Redis {
     protected List<String> args = Collections.emptyList();
     private volatile boolean active = false;
@@ -43,21 +45,14 @@ public synchronized void start() throws EmbeddedRedisException {
 
     private void logErrors() {
         final InputStream errorStream = redisProcess.getErrorStream();
-        executor.submit(() -> {
-            try (BufferedReader reader = new BufferedReader(new InputStreamReader(errorStream))) {
-                String line;
-                while ((line = reader.readLine()) != null) {
-                    System.out.println(line);
-                }
-            } catch (IOException e) {
-                e.printStackTrace();
-            }
-        });
-
+        BufferedReader reader = new BufferedReader(new InputStreamReader(errorStream));
+        Runnable printReaderTask = new PrintReaderRunnable(reader);
+        executor.submit(printReaderTask);
     }
 
     private void awaitRedisServerReady() throws IOException {
-        try (BufferedReader reader = new BufferedReader(new InputStreamReader(redisProcess.getInputStream()))) {
+        BufferedReader reader = new BufferedReader(new InputStreamReader(redisProcess.getInputStream()));
+        try {
             String outputLine;
             do {
                 outputLine = reader.readLine();
@@ -66,6 +61,8 @@ private void awaitRedisServerReady() throws IOException {
                     throw new RuntimeException("Can't start redis server. Check logs for details.");
                 }
             } while (!outputLine.matches(redisReadyPattern()));
+        } finally {
+            IOUtils.closeQuietly(reader);
         }
     }
 
@@ -99,4 +96,31 @@ private void tryWaitFor() {
     public List<Integer> ports() {
         return Arrays.asList(port);
     }
+
+    private static class PrintReaderRunnable implements Runnable {
+        private final BufferedReader reader;
+
+        private PrintReaderRunnable(BufferedReader reader) {
+            this.reader = reader;
+        }
+
+        public void run() {
+            try {
+                readLines();
+            } finally {
+                IOUtils.closeQuietly(reader);
+            }
+        }
+
+        public void readLines() {
+            try {
+                String line;
+                while ((line = reader.readLine()) != null) {
+                    System.out.println(line);
+                }
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
 }

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

@@ -3,14 +3,13 @@
 import com.google.common.collect.Lists;
 import redis.embedded.exceptions.EmbeddedRedisException;
 
+import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 public class RedisCluster implements Redis {
-    private final List<Redis> sentinels = new LinkedList<>();
-    private final List<Redis> servers = new LinkedList<>();
+    private final List<Redis> sentinels = new LinkedList<Redis>();
+    private final List<Redis> servers = new LinkedList<Redis>();
 
     RedisCluster(List<Redis> sentinels, List<Redis> servers) {
         this.servers.addAll(servers);
@@ -19,43 +18,69 @@ public class RedisCluster implements Redis {
 
     @Override
     public boolean isActive() {
-        return sentinels.stream().allMatch(Redis::isActive) && servers.stream().allMatch(Redis::isActive);
+        for(Redis redis : sentinels) {
+            if(!redis.isActive()) {
+                return false;
+            }
+        }
+        for(Redis redis : servers) {
+            if(!redis.isActive()) {
+                return false;
+            }
+        }
+        return true;
     }
 
     @Override
     public void start() throws EmbeddedRedisException {
-        sentinels.parallelStream().forEach(Redis::start);
-        servers.parallelStream().forEach(Redis::start);
+        for(Redis redis : sentinels) {
+            redis.start();
+        }
+        for(Redis redis : servers) {
+            redis.start();
+        }
     }
 
     @Override
     public void stop() throws EmbeddedRedisException {
-        servers.parallelStream().forEach(Redis::stop);
-        sentinels.parallelStream().forEach(Redis::stop);
+        for(Redis redis : sentinels) {
+            redis.stop();
+        }
+        for(Redis redis : servers) {
+            redis.stop();
+        }
     }
 
     @Override
     public List<Integer> ports() {
-        return Stream.concat(
-                sentinels.stream().flatMap(s -> s.ports().stream()),
-                servers.stream().flatMap(s -> s.ports().stream())
-        ).collect(Collectors.toList());
+        List<Integer> ports = new ArrayList<Integer>();
+        ports.addAll(sentinelPorts());
+        ports.addAll(serverPorts());
+        return ports;
     }
 
     public List<Redis> sentinels() {
         return Lists.newLinkedList(sentinels);
     }
 
     public List<Integer> sentinelPorts() {
-        return sentinels.stream().flatMap(s -> s.ports().stream()).collect(Collectors.toList());
+        List<Integer> ports = new ArrayList<Integer>();
+        for(Redis redis : sentinels) {
+            ports.addAll(redis.ports());
+        }
+        return ports;
     }
 
     public List<Redis> servers() {
         return Lists.newLinkedList(servers);
     }
 
     public List<Integer> serverPorts() {
-        return servers.stream().flatMap(s -> s.ports().stream()).collect(Collectors.toList());
+        List<Integer> ports = new ArrayList<Integer>();
+        for(Redis redis : servers) {
+            ports.addAll(redis.ports());
+        }
+        return ports;
     }
 
     public static RedisClusterBuilder builder() {

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

@@ -4,11 +4,10 @@
 import redis.embedded.ports.PredefinedPortProvider;
 import redis.embedded.ports.SequencePortProvider;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 public class RedisClusterBuilder {
     private RedisSentinelBuilder sentinelBuilder = new RedisSentinelBuilder();
@@ -17,7 +16,7 @@ public class RedisClusterBuilder {
     private int quorumSize = 1;
     private PortProvider sentinelPortProvider = new SequencePortProvider(26379);
     private PortProvider replicationGroupPortProvider = new SequencePortProvider(6379);
-    private final List<ReplicationGroup> groups = new LinkedList<>();
+    private final List<ReplicationGroup> groups = new LinkedList<ReplicationGroup>();
 
     public RedisClusterBuilder withSentinelBuilder(RedisSentinelBuilder sentinelBuilder) {
         this.sentinelBuilder = sentinelBuilder;
@@ -84,21 +83,21 @@ public RedisCluster build() {
     }
 
     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());
+        List<Redis> servers = new ArrayList<Redis>();
+        for(ReplicationGroup g : groups) {
+            servers.add(buildMaster(g));
+            buildSlaves(servers, g);
+        }
+        return servers;
     }
 
-    private void buildSlaves(Stream.Builder<Redis> builder, ReplicationGroup g) {
+    private void buildSlaves(List<Redis> servers, 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);
+            servers.add(slave);
         }
     }
 
@@ -109,7 +108,7 @@ private Redis buildMaster(ReplicationGroup g) {
 
     private List<Redis> buildSentinels() {
         int toBuild = this.sentinelCount;
-        final List<Redis> sentinels = new LinkedList<>();
+        final List<Redis> sentinels = new LinkedList<Redis>();
         while (toBuild-- > 0) {
             sentinels.add(buildSentinel());
         }
@@ -119,12 +118,12 @@ private List<Redis> buildSentinels() {
     private Redis buildSentinel() {
         sentinelBuilder.reset();
         sentinelBuilder.port(nextSentinelPort());
-        groups.stream().forEach(g -> {
+        for(ReplicationGroup g : groups) {
             sentinelBuilder.masterName(g.masterName);
             sentinelBuilder.masterPort(g.masterPort);
             sentinelBuilder.quorumSize(quorumSize);
             sentinelBuilder.addDefaultReplicationGroup();
-        });
+        }
         return sentinelBuilder.build();
     }
 
@@ -135,7 +134,7 @@ private int nextSentinelPort() {
     private static class ReplicationGroup {
         private final String masterName;
         private final int masterPort;
-        private final List<Integer> slavePorts = new LinkedList<>();
+        private final List<Integer> slavePorts = new LinkedList<Integer>();
 
         private ReplicationGroup(String masterName, int slaveCount, PortProvider portProvider) {
             this.masterName = masterName;

src/main/java/redis/embedded/RedisSentinel.java

@@ -8,7 +8,7 @@ public class RedisSentinel extends AbstractRedisInstance {
 
     public RedisSentinel(List<String> args, int port) {
         super(port);
-        this.args = new ArrayList<>(args);
+        this.args = new ArrayList<String>(args);
     }
 
     public static RedisSentinelBuilder builder() { return new RedisSentinelBuilder(); }

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

@@ -143,7 +143,7 @@ private String resolveConfigName() {
     private List<String> buildCommandArgs() {
         Preconditions.checkNotNull(sentinelConf);
 
-        List<String> args = new ArrayList<>();
+        List<String> args = new ArrayList<String>();
         args.add(executable.getAbsolutePath());
         args.add(sentinelConf);
         args.add("--sentinel");

src/main/java/redis/embedded/RedisServer.java

@@ -41,7 +41,7 @@ public RedisServer(RedisExecProvider redisExecProvider, Integer port) throws IOE
 
     RedisServer(List<String> args, int port) {
         super(port);
-        this.args = new ArrayList<>(args);
+        this.args = new ArrayList<String>(args);
     }
 
     public static RedisServerBuilder builder() {

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

@@ -106,7 +106,7 @@ private String resolveConfigName() {
     }
 
     private List<String> buildCommandArgs() {
-        List<String> args = new ArrayList<>();
+        List<String> args = new ArrayList<String>();
         args.add(executable.getAbsolutePath());
 
         if (!Strings.isNullOrEmpty(redisConf)) {

src/main/java/redis/embedded/ports/PredefinedPortProvider.java

@@ -9,7 +9,7 @@
 import java.util.List;
 
 public class PredefinedPortProvider implements PortProvider {
-    private final List<Integer> ports = new LinkedList<>();
+    private final List<Integer> ports = new LinkedList<Integer>();
     private final Iterator<Integer> current;
 
     public PredefinedPortProvider(Collection<Integer> ports) {

src/main/java/redis/embedded/util/JedisUtil.java

@@ -3,9 +3,9 @@
 import redis.embedded.Redis;
 import redis.embedded.RedisCluster;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 public class JedisUtil {
     public static Set<String> jedisHosts(Redis redis) {
@@ -19,6 +19,10 @@ public static Set<String> sentinelHosts(RedisCluster cluster) {
     }
 
     public static Set<String> portsToJedisHosts(List<Integer> ports) {
-        return ports.stream().map(p -> "localhost:" + p).collect(Collectors.toSet());
+        Set<String> hosts = new HashSet<String>();
+        for(Integer p : ports) {
+            hosts.add("localhost:" + p);
+        }
+        return hosts;
     }
 }