Multiple bookmarks

Previously it was possible to only supply a single bookmark when creating
a new session. However there is a use-case to be able to supply multiple
bookmarks when multiple worker threads execute write queries and then
reader should be able to observe all those writes. To achieve this driver
now exposed session creation methods that take an iterable of bookmarks.

It was chosen to add additional `Driver#session()` overloads with
`Iterable<String>` instead of extending single-bookmark methods with
varargs because:
 * this maintains backwards compatibility for pre-existing methods
 * addition of varargs would make API look strange because call or
   `Driver#session()` and call of `Driver#session(String... bookmarks)`
   without params is essentially the same thing
 * invocation of vararg methods with single `null` argument can be confusing
 * varargs are more suitable for manual params listing

Driver will now send:
```
{
    bookmark: "max",
    bookmarks: ["one", "two", "max"]
}
```
instead of simple:
```
{
    bookmark: "max"
}
```
this is done to maintain backwards compatibility with databases that
only support single bookmark. It forces driver to parse and compare
bookmarks which violates the fact that bookmarks are opaque. This is
done only to maintain backwards compatibility and should not be copied.
Code doing this will eventually be removed.

Author: lutovich

driver/src/main/java/org/neo4j/driver/internal/Bookmark.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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
+ *
+ * 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.neo4j.driver.internal;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.neo4j.driver.v1.Value;
+
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.singleton;
+import static org.neo4j.driver.v1.Values.value;
+
+public final class Bookmark
+{
+    private static final String BOOKMARK_KEY = "bookmark";
+    private static final String BOOKMARKS_KEY = "bookmarks";
+    private static final String BOOKMARK_PREFIX = "neo4j:bookmark:v1:tx";
+
+    private static final long UNKNOWN_BOOKMARK_VALUE = -1;
+
+    private static final Bookmark EMPTY = new Bookmark( Collections.<String>emptySet() );
+
+    private final Iterable<String> values;
+    private final String maxValue;
+
+    private Bookmark( Iterable<String> values )
+    {
+        this.values = values;
+        this.maxValue = maxBookmark( values );
+    }
+
+    public static Bookmark empty()
+    {
+        return EMPTY;
+    }
+
+    public static Bookmark from( String value )
+    {
+        if ( value == null )
+        {
+            return empty();
+        }
+        return from( singleton( value ) );
+    }
+
+    public static Bookmark from( Iterable<String> values )
+    {
+        if ( values == null )
+        {
+            return empty();
+        }
+        return new Bookmark( values );
+    }
+
+    public boolean isEmpty()
+    {
+        return maxValue == null;
+    }
+
+    public String asString()
+    {
+        return maxValue;
+    }
+
+    public Map<String,Value> asParameters()
+    {
+        if ( isEmpty() )
+        {
+            return emptyMap();
+        }
+
+        // Driver sends {bookmark: "max", bookmarks: ["one", "two", "max"]} instead of simple
+        // {bookmarks: ["one", "two", "max"]} for backwards compatibility reasons. Old servers can only accept single
+        // bookmark that is why driver has to parse and compare given list of bookmarks. This functionality will
+        // eventually be removed.
+        Map<String,Value> parameters = new HashMap<>( 4 );
+        parameters.put( BOOKMARK_KEY, value( maxValue ) );
+        parameters.put( BOOKMARKS_KEY, value( values ) );
+        return parameters;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "Bookmark{values=" + values + "}";
+    }
+
+    private static String maxBookmark( Iterable<String> bookmarks )
+    {
+        if ( bookmarks == null )
+        {
+            return null;
+        }
+
+        Iterator<String> iterator = bookmarks.iterator();
+
+        if ( !iterator.hasNext() )
+        {
+            return null;
+        }
+
+        String maxBookmark = iterator.next();
+        long maxValue = bookmarkValue( maxBookmark );
+
+        while ( iterator.hasNext() )
+        {
+            String bookmark = iterator.next();
+            long value = bookmarkValue( bookmark );
+
+            if ( value > maxValue )
+            {
+                maxBookmark = bookmark;
+                maxValue = value;
+            }
+        }
+
+        return maxBookmark;
+    }
+
+    private static long bookmarkValue( String value )
+    {
+        if ( value.startsWith( BOOKMARK_PREFIX ) )
+        {
+            try
+            {
+                return Long.parseLong( value.substring( BOOKMARK_PREFIX.length() ) );
+            }
+            catch ( NumberFormatException e )
+            {
+                return UNKNOWN_BOOKMARK_VALUE;
+            }
+        }
+        return UNKNOWN_BOOKMARK_VALUE;
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/BookmarkCollector.java

@@ -30,7 +30,7 @@ class BookmarkCollector extends NoOperationCollector
     }
 
     @Override
-    public void bookmark( String bookmark )
+    public void bookmark( Bookmark bookmark )
     {
         transaction.setBookmark( bookmark );
     }

driver/src/main/java/org/neo4j/driver/internal/ExplicitTransaction.java

@@ -34,8 +34,6 @@
 import org.neo4j.driver.v1.exceptions.Neo4jException;
 import org.neo4j.driver.v1.types.TypeSystem;
 
-import static java.util.Collections.emptyMap;
-import static java.util.Collections.singletonMap;
 import static org.neo4j.driver.v1.Values.ofValue;
 import static org.neo4j.driver.v1.Values.value;
 
@@ -68,19 +66,19 @@ private enum State
     private final SessionResourcesHandler resourcesHandler;
     private final Connection conn;
 
-    private String bookmark = null;
+    private Bookmark bookmark = Bookmark.empty();
     private State state = State.ACTIVE;
 
     public ExplicitTransaction( Connection conn, SessionResourcesHandler resourcesHandler )
     {
-        this( conn, resourcesHandler, null );
+        this( conn, resourcesHandler, Bookmark.empty() );
     }
 
-    ExplicitTransaction( Connection conn, SessionResourcesHandler resourcesHandler, String bookmark )
+    ExplicitTransaction( Connection conn, SessionResourcesHandler resourcesHandler, Bookmark initialBookmark )
     {
         this.conn = conn;
         this.resourcesHandler = resourcesHandler;
-        runBeginStatement( conn, bookmark );
+        runBeginStatement( conn, initialBookmark );
     }
 
     @Override
@@ -238,32 +236,28 @@ public synchronized void markToClose()
         state = State.FAILED;
     }
 
-    public String bookmark()
+    public Bookmark bookmark()
     {
         return bookmark;
     }
 
-    void setBookmark( String bookmark )
+    void setBookmark( Bookmark bookmark )
     {
-        this.bookmark = bookmark;
+        if ( bookmark != null && !bookmark.isEmpty() )
+        {
+            this.bookmark = bookmark;
+        }
     }
 
-    private static void runBeginStatement( Connection connection, String bookmark )
+    private static void runBeginStatement( Connection connection, Bookmark bookmark )
     {
-        Map<String,Value> parameters;
-        if ( bookmark != null )
-        {
-            parameters = singletonMap( "bookmark", value( bookmark ) );
-        }
-        else
-        {
-            parameters = emptyMap();
-        }
+        Bookmark initialBookmark = bookmark == null ? Bookmark.empty() : bookmark;
+        Map<String,Value> parameters = initialBookmark.asParameters();
 
         connection.run( "BEGIN", parameters, Collector.NO_OP );
         connection.pullAll( Collector.NO_OP );
 
-        if ( bookmark != null )
+        if ( !initialBookmark.isEmpty() )
         {
             connection.sync();
         }

driver/src/main/java/org/neo4j/driver/internal/InternalDriver.java

@@ -18,6 +18,7 @@
  */
 package org.neo4j.driver.internal;
 
+import java.util.Collection;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.neo4j.driver.internal.security.SecurityPlan;
@@ -62,7 +63,7 @@ public final Session session()
     @Override
     public final Session session( AccessMode mode )
     {
-        return session( mode, null );
+        return newSession( mode, Bookmark.empty() );
     }
 
     @Override
@@ -73,6 +74,23 @@ public final Session session( String bookmark )
 
     @Override
     public final Session session( AccessMode mode, String bookmark )
+    {
+        return newSession( mode, Bookmark.from( bookmark ) );
+    }
+
+    @Override
+    public Session session( Iterable<String> bookmarks )
+    {
+        return session( AccessMode.WRITE, bookmarks );
+    }
+
+    @Override
+    public Session session( AccessMode mode, Iterable<String> bookmarks )
+    {
+        return newSession( mode, Bookmark.from( bookmarks ) );
+    }
+
+    private Session newSession( AccessMode mode, Bookmark bookmark )
     {
         assertOpen();
         Session session = sessionFactory.newInstance( mode, bookmark );

driver/src/main/java/org/neo4j/driver/internal/InternalStatementResult.java

@@ -138,7 +138,7 @@ public void notifications( List<Notification> notifications )
             }
 
             @Override
-            public void bookmark( String bookmark )
+            public void bookmark( Bookmark bookmark )
             {
                 if ( transaction != null )
                 {

driver/src/main/java/org/neo4j/driver/internal/NetworkSession.java

@@ -53,7 +53,7 @@ public class NetworkSession implements Session, SessionResourcesHandler
     private final RetryLogic retryLogic;
     protected final Logger logger;
 
-    private String bookmark;
+    private Bookmark bookmark = Bookmark.empty();
     private PooledConnection currentConnection;
     private ExplicitTransaction currentTransaction;
 
@@ -179,7 +179,7 @@ public synchronized Transaction beginTransaction()
     @Override
     public synchronized Transaction beginTransaction( String bookmark )
     {
-        setBookmark( bookmark );
+        setBookmark( Bookmark.from( bookmark ) );
         return beginTransaction();
     }
 
@@ -195,12 +195,9 @@ public <T> T writeTransaction( TransactionWork<T> work )
         return transaction( AccessMode.WRITE, work );
     }
 
-    // Internal method for setting the bookmark explicitly, mainly for testing.
-    // This method does not prevent setting the bookmark to null since that
-    // is a valid requirement for some test scenarios.
-    void setBookmark( String bookmark )
+    void setBookmark( Bookmark bookmark )
     {
-        if( bookmark != null )
+        if ( bookmark != null && !bookmark.isEmpty() )
         {
             this.bookmark = bookmark;
         }
@@ -209,7 +206,7 @@ void setBookmark( String bookmark )
     @Override
     public String lastBookmark()
     {
-        return bookmark;
+        return bookmark == null ? null : bookmark.asString();
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/SessionFactory.java

@@ -23,5 +23,5 @@
 
 public interface SessionFactory extends AutoCloseable
 {
-    Session newInstance( AccessMode mode, String bookmark );
+    Session newInstance( AccessMode mode, Bookmark bookmark );
 }

driver/src/main/java/org/neo4j/driver/internal/SessionFactoryImpl.java

@@ -27,10 +27,10 @@
 
 public class SessionFactoryImpl implements SessionFactory
 {
-    protected final ConnectionProvider connectionProvider;
-    protected final RetryLogic retryLogic;
-    protected final Logging logging;
-    protected final boolean leakedSessionsLoggingEnabled;
+    private final ConnectionProvider connectionProvider;
+    private final RetryLogic retryLogic;
+    private final Logging logging;
+    private final boolean leakedSessionsLoggingEnabled;
 
     SessionFactoryImpl( ConnectionProvider connectionProvider, RetryLogic retryLogic, Config config )
     {
@@ -41,23 +41,23 @@ public class SessionFactoryImpl implements SessionFactory
     }
 
     @Override
-    public Session newInstance( AccessMode mode, String bookmark )
+    public final Session newInstance( AccessMode mode, Bookmark bookmark )
     {
-        NetworkSession session;
-        if ( leakedSessionsLoggingEnabled )
-        {
-            session = new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, logging );
-        }
-        else
-        {
-            session = new NetworkSession( connectionProvider, mode, retryLogic, logging );
-        }
+        NetworkSession session = createSession( connectionProvider, retryLogic, mode, logging );
         session.setBookmark( bookmark );
         return session;
     }
 
+    protected NetworkSession createSession( ConnectionProvider connectionProvider, RetryLogic retryLogic,
+            AccessMode mode, Logging logging )
+    {
+        return leakedSessionsLoggingEnabled ?
+               new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, logging ) :
+               new NetworkSession( connectionProvider, mode, retryLogic, logging );
+    }
+
     @Override
-    public void close() throws Exception
+    public final void close() throws Exception
     {
         connectionProvider.close();
     }
@@ -69,7 +69,7 @@ public void close() throws Exception
      *
      * @return the connection provider used by this factory.
      */
-    public ConnectionProvider getConnectionProvider()
+    public final ConnectionProvider getConnectionProvider()
     {
         return connectionProvider;
     }