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();
    }
}


@@ -7656,11 +7656,729 @@
 ;%0A    %7D%0A
-%7D
 %0A
+    @Test%0A    public void testRecoverableException() throws Exception%0A    %7B%0A        CuratorProjection curatorProjection = curatorServiceClient.newCuratorProjection(%22test%22);%0A        CreateSpec spec = new CreateSpec();%0A        spec.path = %22/this/wont/work%22;%0A        spec.data = ByteBuffer.wrap(%22value%22.getBytes());%0A        try%0A        %7B%0A            curatorServiceClient.createNode(curatorProjection, spec);%0A            Assert.fail(%22Should have failed%22);%0A        %7D%0A        catch ( CuratorException e )%0A        %7B%0A            Assert.assertEquals(e.getType(), ExceptionType.NODE);%0A            Assert.assertNotNull(e.nodeException);%0A            Assert.assertEquals(e.nodeException, NodeExceptionType.NONODE);%0A        %7D%0A    %7D%0A%7D
 %0A

test recoverable exceptions