Merge pull request buaazp#2 from sumory/master

etcd cluster support for java

juq/README.md

@@ -14,6 +14,6 @@ public enum RedisCommand {
 }
 ```
 
-see the [tests](src/test/java/com/sumory/juq/JuqTest.java) for usage detail.
+check [single node usage](src/test/java/com/sumory/juq/JuqTest.java) and [cluster usage](src/test/java/com/sumory/juq/EtcdTest.java) for detail.
 
 

juq/src/main/java/com/sumory/juq/conn/ConnRedis.java

@@ -9,7 +9,7 @@
 import java.util.Random;
 
 import org.apache.commons.lang3.StringUtils;
-import org.boon.etcd.EtcdClient;
+import org.boon.etcd.Etcd;
 import org.boon.etcd.Node;
 import org.boon.etcd.Response;
 import org.slf4j.Logger;
@@ -21,19 +21,24 @@
 
 public class ConnRedis implements Connection {
     private final static Logger logger = LoggerFactory.getLogger(ConnRedis.class);
-    // private static final Log logger = LogFactory.getLog(ConnRedis.class);
     private String addr;
-    private EtcdClient etcdClient;
+    private Etcd etcdClient;
     private String etcdKey;
     private String[] addrs;
     private Map<String, RedisConnection> conns;
 
     private int maxRetry;
 
-    public ConnRedis(String ip, int port) throws JuqException {
-        String addr = ip + ":" + port;
-        this.addr = addr;
-        this.maxRetry = 1;
+    public ConnRedis(String ip, int port, int maxRetry) throws JuqException {
+        this.addr = ip + ":" + port;
+        this.maxRetry = maxRetry;
+        this.updateConnPool();
+    }
+
+    public ConnRedis(Etcd etcdClient, String etcdKey, int maxRetry) throws JuqException {
+        this.etcdClient = etcdClient;
+        this.etcdKey = etcdKey;
+        this.maxRetry = maxRetry;
         this.updateConnPool();
     }
 

juq/src/test/java/com/sumory/juq/EtcdTest.java

@@ -3,16 +3,19 @@
 import static org.boon.Boon.puts;
 
 import java.net.URI;
+import java.util.Map;
 
 import org.boon.etcd.ClientBuilder;
 import org.boon.etcd.Etcd;
 import org.boon.etcd.Response;
 import org.junit.Assert;
 import org.junit.Test;
 
+import com.sumory.juq.conn.ConnRedis;
+
 public class EtcdTest {
     @Test
-    public void test() {
+    public void testEtcd() {
         Etcd client = ClientBuilder.builder().hosts(//
                 URI.create("http://localhost:4001"),//
                 URI.create("http://localhost:4002"),//
@@ -31,4 +34,59 @@ public void test() {
         r = client.get("foo");
         Assert.assertEquals(r.node().getValue(), "bar");
     }
+
+    @Test
+    public void testNewEtcdConn() {
+        Etcd etcdClient = ClientBuilder.builder().hosts(//
+                URI.create("http://localhost:4001"),//
+                URI.create("http://localhost:4002"),//
+                URI.create("http://localhost:4003")).createClient();
+
+        try {
+            int maxRetry = 1;
+            Response r;
+            r = etcdClient.get("uq");
+            ConnRedis conn = new ConnRedis(etcdClient, "uq", maxRetry);
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
+
+    @Test
+    public void testBasics() {
+        Etcd etcdClient = ClientBuilder.builder().hosts(//
+                URI.create("http://localhost:4001"),//
+                URI.create("http://localhost:4002"),//
+                URI.create("http://localhost:4003")).createClient();
+
+        try {
+            int maxRetry = 1;
+            Response r = etcdClient.get("uq");
+            ConnRedis conn = new ConnRedis(etcdClient, "uq", maxRetry);
+
+            String topic = "topic_from_juq1";
+            String line = "line";
+            String msg = "msg001";
+
+            conn.add(topic);
+            conn.add(topic, line, 10);
+            conn.push(topic, msg.getBytes());
+            Map<String, Object> result = conn.pop(topic + "/" + line);
+            Assert.assertNotNull(result);
+            Assert.assertNotNull(result.get("cid"));
+            System.out.println((String) result.get("cid"));
+            Assert.assertArrayEquals(msg.getBytes(), (byte[]) result.get("value"));
+
+            String fullId = (String) result.get("cid");
+            System.out.println("fullId:" + fullId);
+            conn.del(fullId);
+
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
 }

juq/src/test/java/com/sumory/juq/JuqTest.java

@@ -17,8 +17,9 @@ public void testBasics() {
             String topic = "topic_from_juq1";
             String line = "line";
             String msg = "msg001";
+            int maxRetry = 1;
 
-            Connection conn = new ConnRedis(host, port);
+            Connection conn = new ConnRedis(host, port, maxRetry);
             conn.add(topic);
             conn.add(topic, line, 10);
             conn.push(topic, msg.getBytes());
@@ -45,9 +46,10 @@ public void testDel() {
         String topic = "topic_from_juq2";
         String line = "line_with_recycle";
         String msg = "msg001";
+        int maxRetry = 1;
 
         try {
-            Connection conn = new ConnRedis(host, port);
+            Connection conn = new ConnRedis(host, port, maxRetry);
             conn.add(topic);
             conn.add(topic, line, 10);
             conn.push(topic, msg.getBytes());