/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.tinkerpop.gremlin.server;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.tinkerpop.gremlin.driver.Client;
import org.apache.tinkerpop.gremlin.driver.Cluster;
import org.apache.tinkerpop.gremlin.driver.RequestOptions;
import org.apache.tinkerpop.gremlin.driver.Result;
import org.apache.tinkerpop.gremlin.driver.ResultSet;
import org.apache.tinkerpop.gremlin.driver.Tokens;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseStatusCode;
import org.apache.tinkerpop.gremlin.driver.simple.SimpleClient;
import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
import org.apache.tinkerpop.gremlin.server.op.session.Session;
import org.apache.tinkerpop.gremlin.server.op.session.SessionOpProcessor;
import org.apache.tinkerpop.gremlin.util.Log4jRecordingAppender;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.IsIterableContaining.hasItem;
import static org.hamcrest.core.StringStartsWith.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.junit.Assume.assumeThat;

/**
 * @author Stephen Mallette (http://stephen.genoprime.com)
 */
public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerIntegrationTest {
    private Log4jRecordingAppender recordingAppender = null;
    private Level originalLevel;

    @Before
    public void setupForEachTest() {
        recordingAppender = new Log4jRecordingAppender();
        final Logger rootLogger = Logger.getRootLogger();
        originalLevel = rootLogger.getLevel();
        final String nameOfTest = name.getMethodName();
        switch (nameOfTest) {
            case "shouldCloseSessionOnceOnRequest":
            case "shouldHaveTheSessionTimeout":
            case "shouldCloseSessionOnClientClose":
                final org.apache.log4j.Logger sessionLogger = org.apache.log4j.Logger.getLogger(Session.class);
                sessionLogger.setLevel(Level.DEBUG);
                break;
        }
        rootLogger.addAppender(recordingAppender);
    }

    @After
    public void teardownForEachTest() {
        final org.apache.log4j.Logger sessionLogger = org.apache.log4j.Logger.getLogger(Session.class);
        sessionLogger.setLevel(originalLevel);
        sessionLogger.removeAppender(recordingAppender);
    }

    /**
     * Configure specific Gremlin Server settings for specific tests.
     */
    @Override
    public Settings overrideSettings(final Settings settings) {
        final String nameOfTest = name.getMethodName();
        switch (nameOfTest) {
            case "shouldHaveTheSessionTimeout":
            case "shouldCloseSessionOnceOnRequest":
                settings.processors.clear();

                // OpProcessor setting
                final Settings.ProcessorSettings processorSettings = new Settings.ProcessorSettings();
                processorSettings.className = SessionOpProcessor.class.getCanonicalName();
                processorSettings.config = new HashMap<>();
                processorSettings.config.put(SessionOpProcessor.CONFIG_SESSION_TIMEOUT, 3000L);
                settings.processors.add(processorSettings);

                // Unified setting
                settings.sessionLifetimeTimeout = 3000L;
                break;
            case "shouldCloseSessionOnClientClose":
            case "shouldCloseSessionOnClientCloseWithStateMaintainedBetweenExceptions":
                clearNeo4j(settings);
                break;
            case "shouldEnsureSessionBindingsAreThreadSafe":
                settings.threadPoolWorker = 2;
                break;
            case "shouldUseGlobalFunctionCache":
                // OpProcessor settings are good by default
                // UnifiedHandler settings
                settings.useCommonEngineForSessions = false;
                settings.useGlobalFunctionCacheForSessions = true;

                break;
            case "shouldNotUseGlobalFunctionCache":
                settings.processors.clear();

                // OpProcessor settings
                final Settings.ProcessorSettings processorSettingsForDisableFunctionCache = new Settings.ProcessorSettings();
                processorSettingsForDisableFunctionCache.className = SessionOpProcessor.class.getCanonicalName();
                processorSettingsForDisableFunctionCache.config = new HashMap<>();
                processorSettingsForDisableFunctionCache.config.put(SessionOpProcessor.CONFIG_GLOBAL_FUNCTION_CACHE_ENABLED, false);
                settings.processors.add(processorSettingsForDisableFunctionCache);

                // UnifiedHandler settings
                settings.useCommonEngineForSessions = false;
                settings.useGlobalFunctionCacheForSessions = false;

                break;
            case "shouldExecuteInSessionAndSessionlessWithoutOpeningTransactionWithSingleClient":
            case "shouldExecuteInSessionWithTransactionManagement":
            case "shouldRollbackOnEvalExceptionForManagedTransaction":
            case "shouldNotExecuteQueuedRequestsIfOneInFrontOfItFails":
                clearNeo4j(settings);
                break;
            case "shouldBlowTheSessionQueueSize":
                clearNeo4j(settings);
                settings.maxSessionTaskQueueSize = 1;
                break;
        }

        return settings;
    }

    private static void clearNeo4j(Settings settings) {
        deleteDirectory(new File("/tmp/neo4j"));
        settings.graphs.put("graph", "conf/neo4j-empty.properties");
    }

    @Test
    public void shouldBlowTheSessionQueueSize() throws Exception {
        assumeNeo4jIsPresent();
        assumeThat(isUsingUnifiedChannelizer(), is(true));

        final Cluster cluster = TestClientFactory.open();
        final Client session = cluster.connect(name.getMethodName());

        // maxSessionQueueSize=1
        // we should be able to do one request at a time serially
        assertEquals("test1", session.submit("'test1'").all().get().get(0).getString());
        assertEquals("test2", session.submit("'test2'").all().get().get(0).getString());
        assertEquals("test3", session.submit("'test3'").all().get().get(0).getString());

        final AtomicBoolean errorTriggered = new AtomicBoolean();
        final ResultSet r1 = session.submitAsync("Thread.sleep(1000);'test4'").get();

        final List<CompletableFuture<List<Result>>> blockers = new ArrayList<>();
        for (int ix = 0; ix < 512 && !errorTriggered.get(); ix++) {
            blockers.add(session.submit("'test'").all().exceptionally(t -> {
                final ResponseException re = (ResponseException) t.getCause();
                errorTriggered.compareAndSet(false, ResponseStatusCode.TOO_MANY_REQUESTS == re.getResponseStatusCode());
                return null;
            }));

            // low resource environments like travis might need a break
            if (ix % 32 == 0) Thread.sleep(500);
        }

        // wait for the blockage to clear for sure
        assertEquals("test4", r1.all().get().get(0).getString());
        blockers.forEach(CompletableFuture::join);

        assertThat(errorTriggered.get(), is(true));

        // should be accepting test6 now
        assertEquals("test6", session.submit("'test6'").all().get().get(0).getString());

        session.close();
        cluster.close();
    }

    @Test
    public void shouldNotExecuteQueuedRequestsIfOneInFrontOfItFails() throws Exception {
        assumeNeo4jIsPresent();
        assumeThat(isUsingUnifiedChannelizer(), is(true));

        final Cluster cluster = TestClientFactory.open();
        final Client session = cluster.connect(name.getMethodName());
        final List<CompletableFuture<ResultSet>> futuresAfterFailure = new ArrayList<>();

        // short, sweet and successful - but will rollback after close
        final CompletableFuture<List<Result>> first = session.submit("g.addV('person').iterate();1").all();
        assertEquals(1, first.get().get(0).getInt());

        // leave enough time for the results to queue behind this
        final int followOnRequestCount = 8;
        final CompletableFuture<ResultSet> second = session.submitAsync("g.addV('person').sideEffect{Thread.sleep(60000)}");

        // these requests will queue
        for (int ix = 0; ix < followOnRequestCount; ix ++) {
            futuresAfterFailure.add(session.submitAsync("g.addV('software')"));
        }

        // close the session which will interrupt "second" request and flush the queue of requests behind it
        session.close();

        try {
            second.join().all().join();
            fail("should have been interrupted by channel close");
        } catch (Exception ex) {
            assertEquals("Connection to server is no longer active", ex.getCause().getMessage());
        }

        final AtomicInteger fails = new AtomicInteger(0);
        futuresAfterFailure.forEach(f -> {
            try {
                f.join().all().join();
                fail("should have been interrupted by channel close");
            } catch (Exception ex) {
                assertEquals("Connection to server is no longer active", ex.getCause().getMessage());
                fails.incrementAndGet();
            }
        });

        // all are failures
        assertEquals(followOnRequestCount, fails.get());

        // validate the rollback
        final Client client = cluster.connect();
        assertEquals(0, client.submit("g.V().count()").all().get().get(0).getInt());
        cluster.close();
    }

    @Test
    public void shouldCloseSessionOnClientClose() throws Exception {
        assumeNeo4jIsPresent();

        final Cluster cluster1 = TestClientFactory.open();
        final Client client1 = cluster1.connect(name.getMethodName());
        client1.submit("x = 1").all().join();
        client1.submit("graph.addVertex()").all().join();
        client1.close();
        cluster1.close();

        // the following session close log message is no longer relevant as
        if (isUsingUnifiedChannelizer()) {
            assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
        } else {
            assertThat(recordingAppender.getMessages(), hasItem("DEBUG - Skipped attempt to close open graph transactions on shouldCloseSessionOnClientClose - close was forced\n"));
            assertThat(recordingAppender.getMessages(), hasItem("DEBUG - Session shouldCloseSessionOnClientClose closed\n"));
        }

        // try to reconnect to that session and make sure no state is there
        final Cluster clusterReconnect = TestClientFactory.open();
        final Client clientReconnect = clusterReconnect.connect(name.getMethodName());

        // should get an error because "x" is not defined as this is a new session
        try {
            clientReconnect.submit("y=100").all().join();
            clientReconnect.submit("x").all().join();
            fail("Should not have been successful as 'x' was only defined in the old session");
        } catch(Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertThat(root.getMessage(), startsWith("No such property"));
        }

        // the commit from client1 should not have gone through so there should be no data present.
        assertEquals(0, clientReconnect.submit("graph.traversal().V().count()").all().join().get(0).getInt());

        // must turn on maintainStateAfterException for unified channelizer
        if (!isUsingUnifiedChannelizer()) {
            assertEquals(100, clientReconnect.submit("y").all().join().get(0).getInt());
        }

        clusterReconnect.close();

        if (isUsingUnifiedChannelizer()) {
            assertEquals(0, ((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().getActiveSessionCount());
        }
    }

    @Test
    public void shouldCloseSessionOnClientCloseWithStateMaintainedBetweenExceptions() throws Exception {
        assumeNeo4jIsPresent();
        assumeThat("Must use UnifiedChannelizer", isUsingUnifiedChannelizer(), is(true));

        final Cluster cluster1 = TestClientFactory.open();
        final Client client1 = cluster1.connect(name.getMethodName());
        client1.submit("x = 1").all().join();
        client1.submit("graph.addVertex()").all().join();
        client1.close();
        cluster1.close();

        assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));

        // try to reconnect to that session and make sure no state is there
        final Cluster clusterReconnect = TestClientFactory.open();

        // this configures the client to behave like OpProcessor for UnifiedChannelizer
        final Client.SessionSettings settings = Client.SessionSettings.build().
                sessionId(name.getMethodName()).maintainStateAfterException(true).create();
        final Client clientReconnect = clusterReconnect.connect(Client.Settings.build().useSession(settings).create());

        // should get an error because "x" is not defined as this is a new session
        try {
            clientReconnect.submit("y=100").all().join();
            clientReconnect.submit("x").all().join();
            fail("Should not have been successful as 'x' was only defined in the old session");
        } catch(Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertThat(root.getMessage(), startsWith("No such property"));
        }

        // the commit from client1 should not have gone through so there should be no data present.
        assertEquals(0, clientReconnect.submit("graph.traversal().V().count()").all().join().get(0).getInt());

        // since maintainStateAfterException is enabled the UnifiedChannelizer works like OpProcessor
        assertEquals(100, clientReconnect.submit("y").all().join().get(0).getInt());

        clusterReconnect.close();

        assertEquals(0, ((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().getActiveSessionCount());
    }

    @Test
    public void shouldUseGlobalFunctionCache() throws Exception {
        final Cluster cluster = TestClientFactory.open();
        final Client session = cluster.connect(name.getMethodName());
        final Client client = cluster.connect();

        assertEquals(3, session.submit("def sumItUp(x,y){x+y};sumItUp(1,2)").all().get().get(0).getInt());
        assertEquals(3, session.submit("sumItUp(1,2)").all().get().get(0).getInt());

        try {
            client.submit("sumItUp(1,2)").all().get().get(0).getInt();
            fail("Global functions should not be cached so the call to sumItUp() should fail");
        } catch (Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertThat(root.getMessage(), startsWith("No signature of method"));
        } finally {
            cluster.close();
        }
    }

    @Test
    public void shouldNotUseGlobalFunctionCache() throws Exception {
        final Cluster cluster = TestClientFactory.open();
        final Client client = cluster.connect(name.getMethodName());

        try {
            assertEquals(3, client.submit("def sumItUp(x,y){x+y};sumItUp(1,2)").all().get().get(0).getInt());
        } catch (Exception ex) {
            cluster.close();
            throw ex;
        }

        try {
            client.submit("sumItUp(1,2)").all().get().get(0).getInt();
            fail("Global functions should not be cached so the call to addItUp() should fail");
        } catch (Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertThat(root.getMessage(), startsWith("No signature of method"));
        } finally {
            cluster.close();
        }
    }

    @Test
    public void shouldNotAllowMoreThanOneClientPerSession() throws Exception {
        final Cluster cluster1 = TestClientFactory.open();
        final Client client1 = cluster1.connect(name.getMethodName());
        client1.submit("1+1").all().join();
        final Cluster cluster2 = TestClientFactory.open();
        final Client.SessionSettings sessionSettings = Client.SessionSettings.build()
                .sessionId(name.getMethodName())
                .forceClosed(true).create();
        final Client client2 = cluster2.connect(Client.Settings.build().useSession(sessionSettings).create());

        try {
            client2.submit("2+2").all().join();
            fail("Can't have more than one client connecting to the same session");
        } catch (Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertEquals("Session shouldNotAllowMoreThanOneClientPerSession is not bound to the connecting client", root.getMessage());
        }

        cluster1.close();
        cluster2.close();
    }

    @Test
    public void shouldRollbackOnEvalExceptionForManagedTransaction() throws Exception {
        assumeNeo4jIsPresent();

        final Cluster cluster = TestClientFactory.open();
        final Client.SessionSettings sessionSettings = Client.SessionSettings.build().
                sessionId(name.getMethodName()).
                manageTransactions(true).
                maintainStateAfterException(false).
                create();
        final Client.Settings clientSettings = Client.Settings.build().useSession(sessionSettings).create();
        final Client client = cluster.connect(clientSettings);

        try {
            client.submit("graph.addVertex(); throw new Exception('no worky')").all().get();
            fail("Should have tossed the manually generated exception");
        } catch (Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertEquals("no worky", root.getMessage());
        }

        try {
            // just force a commit here of "something" to ensure the rollback of the previous request
            client.submit("graph.addVertex(); graph.tx().commit()").all().get(10, TimeUnit.SECONDS);

            // the transaction is managed so a rollback should have executed
            assertEquals(1, client.submit("g.V().count()").all().get(10, TimeUnit.SECONDS).get(0).getInt());
        } finally {
            cluster.close();
        }
    }

    @Test
    public void shouldCloseSessionOnceOnRequest() throws Exception {
        final Cluster cluster = TestClientFactory.open();
        final Client client = cluster.connect(name.getMethodName());

        final ResultSet results1 = client.submit("x = [1,2,3,4,5,6,7,8,9]");
        final AtomicInteger counter = new AtomicInteger(0);
        results1.stream().map(i -> i.get(Integer.class) * 2).forEach(i -> assertEquals(counter.incrementAndGet() * 2, Integer.parseInt(i.toString())));

        final ResultSet results2 = client.submit("x[0]+1");
        assertEquals(2, results2.all().get().get(0).getInt());

        // explicitly close the session
        client.close();

        // wait past automatic session expiration
        Thread.sleep(3500);

        try {
            // the original session should be dead so this call will open a new session with the same name but fail
            // because the state is now gone - x is an invalid property
            client.submit("x[1]+2").all().get();
            fail("Session should be dead");
        } catch (Exception ex) {
            final Throwable root = ExceptionUtils.getRootCause(ex);
            assertThat(root, instanceOf(IllegalStateException.class));
        } finally {
            cluster.close();
        }

        if (isUsingUnifiedChannelizer()) {
            assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
        } else {
            assertEquals(1, recordingAppender.getMessages().stream()
                    .filter(msg -> msg.equals("DEBUG - Session shouldCloseSessionOnceOnRequest closed\n")).count());
        }
    }

    @Test
    public void shouldHaveTheSessionTimeout() throws Exception {
        final Cluster cluster = TestClientFactory.open();
        final Client.SessionSettings settings = Client.SessionSettings.build().
                sessionId(name.getMethodName()).maintainStateAfterException(true).create();
        final Client session = cluster.connect(Client.Settings.build().useSession(settings).create());

        final ResultSet results1 = session.submit("x = [1,2,3,4,5,6,7,8,9]");
        final AtomicInteger counter = new AtomicInteger(0);
        results1.stream().map(i -> i.get(Integer.class) * 2).forEach(i -> assertEquals(counter.incrementAndGet() * 2, Integer.parseInt(i.toString())));

        final ResultSet results2 = session.submit("x[0]+1");
        assertEquals(2, results2.all().get().get(0).getInt());

        // session times out in 3 seconds but slow environments like travis might not be so exacting in the
        // shutdown process so be patient
        if (isUsingUnifiedChannelizer()) {
            boolean sessionAlive = true;
            for (int ix = 1; ix < 11; ix++) {
                sessionAlive = ((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName());
                if (!sessionAlive) break;
                Thread.sleep(ix * 500);
            }

            assertThat(sessionAlive, is(false));
        } else {
            Thread.sleep(5000);
        }

        try {
            // the original session should be dead so this call will open a new session with the same name but fail
            // because the state is now gone - x is an invalid property.
            session.submit("x[1]+2").all().get();
            fail("Session should be dead");
        } catch (Exception ex) {
            final Throwable cause = ExceptionUtils.getCause(ex);
            assertThat(cause, instanceOf(ResponseException.class));
            assertEquals(ResponseStatusCode.SERVER_ERROR_EVALUATION, ((ResponseException) cause).getResponseStatusCode());

            // validate that we can still send messages to the server
            assertEquals(2, session.submit("1+1").all().join().get(0).getInt());
        } finally {
            cluster.close();
        }

        if (isUsingUnifiedChannelizer()) {
            assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
        } else {
            // there will be one for the timeout and a second for closing the cluster
            assertEquals(2, recordingAppender.getMessages().stream()
                    .filter(msg -> msg.equals("DEBUG - Session shouldHaveTheSessionTimeout closed\n")).count());
        }
    }

    @Test
    public void shouldEnsureSessionBindingsAreThreadSafe() throws Exception {
        final Cluster cluster = TestClientFactory.build().
                minInProcessPerConnection(16).maxInProcessPerConnection(64).create();
        final Client client = cluster.connect(name.getMethodName());

        try {
            client.submit("a=100;b=1000;c=10000;null").all().get();
            final int requests = 10000;
            final List<CompletableFuture<ResultSet>> futures = new ArrayList<>(requests);
            IntStream.range(0, requests).forEach(i -> {
                try {
                    futures.add(client.submitAsync("a+b+c"));
                } catch (Exception ex) {
                    throw new RuntimeException(ex);
                }
            });

            assertEquals(requests, futures.size());

            int counter = 0;
            for (CompletableFuture<ResultSet> f : futures) {
                final Result r = f.get().all().get(30000, TimeUnit.MILLISECONDS).get(0);
                assertEquals(11100, r.getInt());
                counter++;
            }

            assertEquals(requests, counter);
        } catch (Exception ex) {
            fail(ex.getMessage());
        } finally {
            cluster.close();
        }

    }

    @Test
    @SuppressWarnings("unchecked")
    public void shouldExecuteInSessionAndSessionlessWithoutOpeningTransactionWithSingleClient() throws Exception {
        assumeNeo4jIsPresent();

        try (final SimpleClient client = TestClientFactory.createWebSocketClient()) {

            //open a transaction, create a vertex, commit
            final RequestMessage openRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "graph.tx().open()")
                    .create();
            final List<ResponseMessage> openResponses = client.submit(openRequest);
            assertEquals(1, openResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, openResponses.get(0).getStatus().getCode());

            final RequestMessage addRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "v=graph.addVertex(\"name\",\"stephen\")")
                    .create();
            final List<ResponseMessage> addResponses = client.submit(addRequest);
            assertEquals(1, addResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, addResponses.get(0).getStatus().getCode());

            final RequestMessage commitRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "graph.tx().commit()")
                    .create();
            final List<ResponseMessage> commitResponses = client.submit(commitRequest);
            assertEquals(1, commitResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, commitResponses.get(0).getStatus().getCode());

            // Check to see if the transaction is closed.
            final RequestMessage checkRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "graph.tx().isOpen()")
                    .create();
            final List<ResponseMessage> checkResponses = client.submit(checkRequest);
            assertEquals(1, checkResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, checkResponses.get(0).getStatus().getCode());
            assertThat(((List<Boolean>) checkResponses.get(0).getResult().getData()).get(0), is(false));

            //lets run a sessionless read
            final RequestMessage sessionlessRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .addArg(Tokens.ARGS_GREMLIN, "graph.traversal().V()")
                    .create();
            final List<ResponseMessage> sessionlessResponses = client.submit(sessionlessRequest);
            assertEquals(1, sessionlessResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, sessionlessResponses.get(0).getStatus().getCode());

            // Check to see if the transaction is still closed.
            final RequestMessage checkAgainRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "graph.tx().isOpen()")
                    .create();
            final List<ResponseMessage> checkAgainstResponses = client.submit(checkAgainRequest);
            assertEquals(1, checkAgainstResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, checkAgainstResponses.get(0).getStatus().getCode());
            assertThat(((List<Boolean>) checkAgainstResponses.get(0).getResult().getData()).get(0), is(false));
        }
    }

    @Test
    @SuppressWarnings("unchecked")
    public void shouldExecuteInSessionWithTransactionManagement() throws Exception {
        assumeNeo4jIsPresent();

        try (final SimpleClient client = TestClientFactory.createWebSocketClient()) {
            final RequestMessage addRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "v=graph.addVertex(\"name\",\"stephen\")")
                    .addArg(Tokens.ARGS_MANAGE_TRANSACTION, true)
                    .create();
            final List<ResponseMessage> addResponses = client.submit(addRequest);
            assertEquals(1, addResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, addResponses.get(0).getStatus().getCode());

            // Check to see if the transaction is closed.
            final RequestMessage checkRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "graph.tx().isOpen()")
                    .create();
            final List<ResponseMessage> checkResponses = client.submit(checkRequest);
            assertEquals(1, checkResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, checkResponses.get(0).getStatus().getCode());
            assertThat(((List<Boolean>) checkResponses.get(0).getResult().getData()).get(0), is(false));

            // lets run a sessionless read and validate that the transaction was managed
            final RequestMessage sessionlessRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .addArg(Tokens.ARGS_GREMLIN, "graph.traversal().V().values('name')")
                    .create();
            final List<ResponseMessage> sessionlessResponses = client.submit(sessionlessRequest);
            assertEquals(1, sessionlessResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, sessionlessResponses.get(0).getStatus().getCode());
            assertEquals("stephen", ((List<String>) sessionlessResponses.get(0).getResult().getData()).get(0));

            // make sure the session is intact
            final RequestMessage getRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "v.values(\"name\")")
                    .addArg(Tokens.ARGS_MANAGE_TRANSACTION, true)
                    .create();
            final List<ResponseMessage> getResponses = client.submit(getRequest);
            assertEquals(1, getResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, getResponses.get(0).getStatus().getCode());
            assertEquals("stephen", ((List<String>) getResponses.get(0).getResult().getData()).get(0));

            // Check to see if the transaction is still closed.
            final RequestMessage checkAgainRequest = RequestMessage.build(Tokens.OPS_EVAL)
                    .processor("session")
                    .addArg(Tokens.ARGS_SESSION, name.getMethodName())
                    .addArg(Tokens.ARGS_GREMLIN, "graph.tx().isOpen()")
                    .create();
            final List<ResponseMessage> checkAgainstResponses = client.submit(checkAgainRequest);
            assertEquals(1, checkAgainstResponses.size());
            assertEquals(ResponseStatusCode.SUCCESS, checkAgainstResponses.get(0).getStatus().getCode());
            assertThat(((List<Boolean>) checkAgainstResponses.get(0).getResult().getData()).get(0), is(false));
        }
    }
}