More CC tests, support for getting cluster members

Author: lutovich

driver/src/test/java/org/neo4j/driver/v1/integration/CausalClusteringIT.java

@@ -1,15 +1,15 @@
 /**
  * Copyright (c) 2002-2016 "Neo Technology,"
  * Network Engine for Objects in Lund AB [http://neotechnology.com]
- *
+ * <p>
  * This file is part of Neo4j.
- *
+ * <p>
  * Licensed 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * 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.
@@ -21,20 +21,31 @@
 import org.junit.Rule;
 import org.junit.Test;
 
+import java.net.URI;
 import java.util.concurrent.TimeoutException;
 
-import org.neo4j.driver.internal.util.Consumer;
+import org.neo4j.driver.internal.logging.DevNullLogger;
 import org.neo4j.driver.v1.AccessMode;
+import org.neo4j.driver.v1.Config;
 import org.neo4j.driver.v1.Driver;
 import org.neo4j.driver.v1.GraphDatabase;
+import org.neo4j.driver.v1.Logger;
+import org.neo4j.driver.v1.Logging;
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.Session;
+import org.neo4j.driver.v1.Transaction;
+import org.neo4j.driver.v1.Values;
+import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.v1.exceptions.SessionExpiredException;
 import org.neo4j.driver.v1.util.Function;
 import org.neo4j.driver.v1.util.cc.Cluster;
 import org.neo4j.driver.v1.util.cc.ClusterMember;
 import org.neo4j.driver.v1.util.cc.ClusterRule;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
 public class CausalClusteringIT
 {
     private static final long DEFAULT_TIMEOUT_MS = 15_000;
@@ -47,26 +58,149 @@ public void shouldExecuteReadAndWritesWhenDriverSuppliedWithAddressOfLeader() th
     {
         Cluster cluster = clusterRule.getCluster();
 
-        ClusterMember leader = cluster.leaderTx( new Consumer<Session>()
+        int count = executeWriteAndReadThroughBolt( cluster.leader() );
+
+        assertEquals( 1, count );
+    }
+
+    @Test
+    public void shouldExecuteReadAndWritesWhenDriverSuppliedWithAddressOfFollower() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+
+        int count = executeWriteAndReadThroughBolt( cluster.anyFollower() );
+
+        assertEquals( 1, count );
+    }
+
+    @Test
+    public void sessionCreationShouldFailIfCallingDiscoveryProcedureOnEdgeServer() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+
+        ClusterMember readReplica = cluster.anyReadReplica();
+        try
         {
-            @Override
-            public void accept( Session session )
+            createDriver( readReplica.getRoutingUri() );
+            fail( "Should have thrown an exception using a read replica address for routing" );
+        }
+        catch ( ServiceUnavailableException ex )
+        {
+            assertEquals( "Could not perform discovery. No routing servers available.", ex.getMessage() );
+        }
+    }
+
+    // Ensure that Bookmarks work with single instances using a driver created using a bolt[not+routing] URI.
+    @Test
+    public void bookmarksShouldWorkWithDriverPinnedToSingleServer() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+        ClusterMember leader = cluster.leader();
+
+        try ( Driver driver = createDriver( leader.getBoltUri() ) )
+        {
+            String bookmark = inExpirableSession( driver, createSession(), new Function<Session,String>()
+            {
+                @Override
+                public String apply( Session session )
+                {
+                    try ( Transaction tx = session.beginTransaction() )
+                    {
+                        tx.run( "CREATE (p:Person {name: {name} })", Values.parameters( "name", "Alistair" ) );
+                        tx.success();
+                    }
+
+                    return session.lastBookmark();
+                }
+            } );
+
+            assertNotNull( bookmark );
+
+            try ( Session session = driver.session();
+                  Transaction tx = session.beginTransaction( bookmark ) )
             {
-                session.run( "CREATE CONSTRAINT ON (p:Person) ASSERT p.name is UNIQUE" );
+                Record record = tx.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                assertEquals( 1, record.get( "count" ).asInt() );
+                tx.success();
             }
-        } );
+        }
+    }
+
+    @Test
+    public void shouldUseBookmarkFromAReadSessionInAWriteSession() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+        ClusterMember leader = cluster.leader();
 
-        executeWriteAndReadThroughBolt( leader );
+        try ( Driver driver = createDriver( leader.getBoltUri() ) )
+        {
+            inExpirableSession( driver, createWritableSession(), new Function<Session,Void>()
+            {
+                @Override
+                public Void apply( Session session )
+                {
+                    session.run( "CREATE (p:Person {name: {name} })", Values.parameters( "name", "Jim" ) );
+                    return null;
+                }
+            } );
+
+            final String bookmark;
+            try ( Session session = driver.session( AccessMode.READ ) )
+            {
+                try ( Transaction tx = session.beginTransaction() )
+                {
+                    tx.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                    tx.success();
+                }
+
+                bookmark = session.lastBookmark();
+            }
+
+            assertNotNull( bookmark );
+
+            inExpirableSession( driver, createWritableSession(), new Function<Session,Void>()
+            {
+                @Override
+                public Void apply( Session session )
+                {
+                    try ( Transaction tx = session.beginTransaction( bookmark ) )
+                    {
+                        tx.run( "CREATE (p:Person {name: {name} })", Values.parameters( "name", "Alistair" ) );
+                        tx.success();
+                    }
+
+                    return null;
+                }
+            } );
+
+            try ( Session session = driver.session() )
+            {
+                Record record = session.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                assertEquals( 2, record.get( "count" ).asInt() );
+            }
+        }
     }
 
     private int executeWriteAndReadThroughBolt( ClusterMember member ) throws TimeoutException, InterruptedException
     {
-        try ( Driver driver = GraphDatabase.driver( member.getRoutingUri(), clusterRule.getDefaultAuthToken() ) )
+        try ( Driver driver = createDriver( member.getRoutingUri() ) )
         {
             return inExpirableSession( driver, createWritableSession(), executeWriteAndRead() );
         }
     }
 
+    private Function<Driver,Session> createSession()
+    {
+        return new Function<Driver,Session>()
+        {
+            @Override
+            public Session apply( Driver driver )
+            {
+                return driver.session();
+            }
+        };
+    }
+
     private Function<Driver,Session> createWritableSession()
     {
         return new Function<Driver,Session>()
@@ -113,4 +247,22 @@ private <T> T inExpirableSession( Driver driver, Function<Driver,Session> acquir
 
         throw new TimeoutException( "Transaction did not succeed in time" );
     }
+
+    private Driver createDriver( URI boltUri )
+    {
+        Logging devNullLogging = new Logging()
+        {
+            @Override
+            public Logger getLog( String name )
+            {
+                return DevNullLogger.DEV_NULL_LOGGER;
+            }
+        };
+
+        Config config = Config.build()
+                .withLogging( devNullLogging )
+                .toConfig();
+
+        return GraphDatabase.driver( boltUri, clusterRule.getDefaultAuthToken(), config );
+    }
 }