curator-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From randg...@apache.org
Subject [1/3] git commit: start writing RPC tests
Date Mon, 02 Jun 2014 16:20:47 GMT
Repository: curator
Updated Branches:
  refs/heads/curator-rpc f3ec63cb6 -> 9077ad001


start writing RPC tests


Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/bee07a52
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/bee07a52
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/bee07a52

Branch: refs/heads/curator-rpc
Commit: bee07a525682436ce33e7110b8f6688ec1df285f
Parents: f3ec63c
Author: randgalt <randgalt@apache.org>
Authored: Sun Jun 1 16:11:02 2014 -0500
Committer: randgalt <randgalt@apache.org>
Committed: Sun Jun 1 16:11:02 2014 -0500

----------------------------------------------------------------------
 .../curator/x/rpc/CuratorProjectionServer.java  |  14 +-
 .../java/org/apache/curator/x/rpc/RpcTests.java | 207 +++++++++++++++++++
 2 files changed, 217 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/bee07a52/curator-x-rpc/src/main/java/org/apache/curator/x/rpc/CuratorProjectionServer.java
----------------------------------------------------------------------
diff --git a/curator-x-rpc/src/main/java/org/apache/curator/x/rpc/CuratorProjectionServer.java
b/curator-x-rpc/src/main/java/org/apache/curator/x/rpc/CuratorProjectionServer.java
index 806f06e..a01f462 100644
--- a/curator-x-rpc/src/main/java/org/apache/curator/x/rpc/CuratorProjectionServer.java
+++ b/curator-x-rpc/src/main/java/org/apache/curator/x/rpc/CuratorProjectionServer.java
@@ -77,10 +77,7 @@ public class CuratorProjectionServer
             configurationSource = args[0];
         }
 
-        Configuration configuration = new ConfigurationBuilder(configurationSource).build();
-
-        final CuratorProjectionServer server = new CuratorProjectionServer(configuration);
-        server.start();
+        final CuratorProjectionServer server = startServer(configurationSource);
 
         Runnable shutdown = new Runnable()
         {
@@ -94,6 +91,15 @@ public class CuratorProjectionServer
         Runtime.getRuntime().addShutdownHook(hook);
     }
 
+    public static CuratorProjectionServer startServer(String configurationSource) throws
Exception
+    {
+        Configuration configuration = new ConfigurationBuilder(configurationSource).build();
+
+        final CuratorProjectionServer server = new CuratorProjectionServer(configuration);
+        server.start();
+        return server;
+    }
+
     public CuratorProjectionServer(Configuration configuration)
     {
         this.configuration = configuration;

http://git-wip-us.apache.org/repos/asf/curator/blob/bee07a52/curator-x-rpc/src/test/java/org/apache/curator/x/rpc/RpcTests.java
----------------------------------------------------------------------
diff --git a/curator-x-rpc/src/test/java/org/apache/curator/x/rpc/RpcTests.java b/curator-x-rpc/src/test/java/org/apache/curator/x/rpc/RpcTests.java
new file mode 100644
index 0000000..5a83b25
--- /dev/null
+++ b/curator-x-rpc/src/test/java/org/apache/curator/x/rpc/RpcTests.java
@@ -0,0 +1,207 @@
+package org.apache.curator.x.rpc;
+
+import org.apache.curator.generated.*;
+import org.apache.curator.test.BaseClassForTests;
+import org.apache.curator.test.InstanceSpec;
+import org.apache.curator.test.Timing;
+import org.apache.curator.utils.ThreadUtils;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.node.ArrayNode;
+import org.codehaus.jackson.node.ObjectNode;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import java.nio.ByteBuffer;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class RpcTests extends BaseClassForTests
+{
+    private Timing timing = new Timing();
+    private CuratorProjectionServer thriftServer;
+    private CuratorService.Client curatorServiceClient;
+    private EventService.Client eventServiceClient;
+    private int thriftPort;
+
+    @BeforeMethod
+    @Override
+    public void setup() throws Exception
+    {
+        super.setup();
+
+        ObjectMapper mapper = new ObjectMapper();
+
+        ObjectNode connectionNode = mapper.createObjectNode();
+        connectionNode.put("name", "test");
+        connectionNode.put("connectionString", server.getConnectString());
+
+        ObjectNode thriftNode = mapper.createObjectNode();
+        thriftPort = InstanceSpec.getRandomPort();
+        thriftNode.put("port", thriftPort);
+
+        ArrayNode connections = mapper.createArrayNode();
+        connections.add(connectionNode);
+
+        ObjectNode node = mapper.createObjectNode();
+        node.put("connections", connections);
+        node.put("thrift", thriftNode);
+
+        final String configurationJson = mapper.writeValueAsString(node);
+
+        thriftServer = CuratorProjectionServer.startServer(configurationJson);
+
+        TSocket clientTransport = new TSocket("localhost", thriftPort);
+        clientTransport.setTimeout(timing.connection());
+        clientTransport.open();
+        TProtocol clientProtocol = new TBinaryProtocol(clientTransport);
+        curatorServiceClient = new CuratorService.Client(clientProtocol);
+
+        TSocket eventTransport = new TSocket("localhost", thriftPort);
+        eventTransport.setTimeout(timing.connection());
+        eventTransport.open();
+        TProtocol eventProtocol = new TBinaryProtocol(eventTransport);
+        eventServiceClient = new EventService.Client(eventProtocol);
+
+    }
+
+    @AfterMethod
+    @Override
+    public void teardown() throws Exception
+    {
+        thriftServer.stop();
+
+        super.teardown();
+    }
+
+    @Test
+    public void testBasic() throws Exception
+    {
+        CuratorProjection curatorProjection = curatorServiceClient.newCuratorProjection("test");
+        CreateSpec spec = new CreateSpec();
+        spec.path = "/test";
+        spec.data = ByteBuffer.wrap("value".getBytes());
+        OptionalPath node = curatorServiceClient.createNode(curatorProjection, spec);
+        Assert.assertEquals(node.path, "/test");
+
+        GetDataSpec dataSpec = new GetDataSpec();
+        dataSpec.path = "/test";
+        OptionalData data = curatorServiceClient.getData(curatorProjection, dataSpec);
+        Assert.assertEquals(data.data, ByteBuffer.wrap("value".getBytes()));
+    }
+
+    @Test
+    public void testEvents() throws Exception
+    {
+        final CuratorProjection curatorProjection = curatorServiceClient.newCuratorProjection("test");
+
+        final CountDownLatch connectedLatch = new CountDownLatch(1);
+        final CountDownLatch nodeCreatedLatch = new CountDownLatch(1);
+        Callable<Void> proc = new Callable<Void>()
+        {
+            @Override
+            public Void call() throws Exception
+            {
+                while ( !Thread.currentThread().isInterrupted() )
+                {
+                    CuratorEvent event = eventServiceClient.getNextEvent(curatorProjection);
+                    if ( event.type == CuratorEventType.CONNECTION_CONNECTED )
+                    {
+                        connectedLatch.countDown();
+                    }
+                    else if ( event.type == CuratorEventType.WATCHED )
+                    {
+                        if ( event.watchedEvent.eventType == EventType.NodeCreated )
+                        {
+                            nodeCreatedLatch.countDown();
+                        }
+                    }
+                }
+                return null;
+            }
+        };
+        Future<Void> eventFuture = ThreadUtils.newSingleThreadExecutor("test").submit(proc);
+
+        Assert.assertTrue(timing.awaitLatch(connectedLatch));
+
+        ExistsSpec spec = new ExistsSpec();
+        spec.path = "/test";
+        spec.watched = true;
+        curatorServiceClient.exists(curatorProjection, spec);
+
+        CreateSpec createSpec = new CreateSpec();
+        createSpec.path = "/test";
+        curatorServiceClient.createNode(curatorProjection, createSpec);
+
+        Assert.assertTrue(timing.awaitLatch(nodeCreatedLatch));
+
+        eventFuture.cancel(true);
+    }
+
+    @Test
+    public void testLockMultiThread() throws Exception
+    {
+        final Timing timing = new Timing();
+
+        TSocket clientTransport = new TSocket("localhost", thriftPort);
+        clientTransport.setTimeout(timing.connection());
+        clientTransport.open();
+        TProtocol clientProtocol = new TBinaryProtocol(clientTransport);
+        final CuratorService.Client secondCuratorServiceClient = new CuratorService.Client(clientProtocol);
+        ExecutorService service = ThreadUtils.newFixedThreadPool(2, "test");
+        ExecutorCompletionService<Void> completer = new ExecutorCompletionService<Void>(service);
+
+        final CountDownLatch lockLatch = new CountDownLatch(2);
+        final AtomicBoolean hasTheLock = new AtomicBoolean();
+        for ( int i = 0; i < 2; ++i )
+        {
+            final CuratorService.Client client = (i == 0) ? curatorServiceClient : secondCuratorServiceClient;
+            Callable<Void> proc = new Callable<Void>()
+            {
+                @Override
+                public Void call() throws Exception
+                {
+                    CuratorProjection curatorProjection = client.newCuratorProjection("test");
+                    LockProjection lockProjection = client.acquireLock(curatorProjection,
"/lock", timing.forWaiting().milliseconds());
+                    if ( lockProjection.id == null )
+                    {
+                        throw new Exception("Could not acquire lock");
+                    }
+                    try
+                    {
+                        if ( !hasTheLock.compareAndSet(false, true) )
+                        {
+                            throw new Exception("Two lockers");
+                        }
+
+                        timing.sleepABit();
+                    }
+                    finally
+                    {
+                        hasTheLock.set(false);
+                        lockLatch.countDown();
+                        client.closeGenericProjection(curatorProjection, lockProjection.id);
+                    }
+
+                    return null;
+                }
+            };
+            completer.submit(proc);
+        }
+
+        completer.take().get();
+        completer.take().get();
+
+        Assert.assertTrue(timing.awaitLatch(lockLatch));
+
+        service.shutdownNow();
+    }
+}
+


Mime
View raw message