Simplify Response impl

By making `InternalFuture` implement this public interface. Previously
a special adapter object was needed. Now we can directly return
internal classes.

Author: lutovich

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

@@ -216,10 +216,10 @@ private void rollbackTx()
     @Override
     public Response<Void> commitAsync()
     {
-        return internalCommitAsync().asResponse();
+        return internalCommitAsync();
     }
 
-    InternalFuture<Void> internalCommitAsync()
+    private InternalFuture<Void> internalCommitAsync()
     {
         if ( state == State.COMMITTED )
         {
@@ -239,7 +239,7 @@ else if ( state == State.ROLLED_BACK )
     @Override
     public Response<Void> rollbackAsync()
     {
-        return internalRollbackAsync().asResponse();
+        return internalRollbackAsync();
     }
 
     InternalFuture<Void> internalRollbackAsync()
@@ -390,7 +390,7 @@ public StatementResult run( Statement statement )
     public Response<StatementResultCursor> runAsync( Statement statement )
     {
         ensureNotFailed();
-        return QueryRunner.runAsync( asyncConnection, statement, this ).asResponse();
+        return QueryRunner.runAsync( asyncConnection, statement, this );
     }
 
     @Override

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

@@ -26,7 +26,7 @@
 import org.neo4j.driver.ResultResourcesHandler;
 import org.neo4j.driver.internal.async.AsyncConnection;
 import org.neo4j.driver.internal.async.InternalFuture;
-import org.neo4j.driver.internal.async.InternalResponse;
+import org.neo4j.driver.internal.async.InternalPromise;
 import org.neo4j.driver.internal.async.QueryRunner;
 import org.neo4j.driver.internal.logging.DelegatingLogger;
 import org.neo4j.driver.internal.retry.RetryLogic;
@@ -160,7 +160,7 @@ public InternalFuture<StatementResultCursor> apply( AsyncConnection connection )
             {
                 return QueryRunner.runAsync( connection, statement );
             }
-        } ).asResponse();
+        } );
     }
 
     public static StatementResult run( Connection connection, Statement statement,
@@ -247,7 +247,7 @@ public InternalFuture<Void> apply( AsyncConnection connection )
                 {
                     return connection.forceRelease();
                 }
-            } ).asResponse();
+            } );
         }
         else if ( currentAsyncTransactionFuture != null )
         {
@@ -258,11 +258,11 @@ public InternalFuture<Void> apply( ExplicitTransaction tx )
                 {
                     return tx.internalRollbackAsync();
                 }
-            } ).asResponse();
+            } );
         }
         else
         {
-            return new InternalResponse<>( GlobalEventExecutor.INSTANCE.<Void>newSucceededFuture( null ) );
+            return new InternalPromise<Void>( GlobalEventExecutor.INSTANCE ).setSuccess( null );
         }
     }
 
@@ -431,7 +431,7 @@ public InternalFuture<ExplicitTransaction> apply( AsyncConnection connection )
                 } );
 
         //noinspection unchecked
-        return (Response) currentAsyncTransactionFuture.asResponse();
+        return (Response) currentAsyncTransactionFuture;
     }
 
     private void ensureNoUnrecoverableError()

driver/src/main/java/org/neo4j/driver/internal/async/Futures.java

@@ -33,7 +33,7 @@ private Futures()
 
     public static <T, U> InternalFuture<U> thenApply( InternalFuture<T> future, Function<T,U> fn )
     {
-        InternalPromise<U> result = new InternalPromise<>( future.eventLoop() );
+        InternalPromise<U> result = new InternalPromise<>( future.eventExecutor() );
         future.addListener( new ThenApplyListener<>( result, fn ) );
         return result;
     }
@@ -47,14 +47,14 @@ public static <T, U> InternalFuture<U> thenApply( Future<T> future, Bootstrap bo
 
     public static <T, U> InternalFuture<U> thenCombine( InternalFuture<T> future, Function<T,InternalFuture<U>> fn )
     {
-        InternalPromise<U> result = new InternalPromise<>( future.eventLoop() );
+        InternalPromise<U> result = new InternalPromise<>( future.eventExecutor() );
         future.addListener( new ThenCombineListener<>( result, fn ) );
         return result;
     }
 
     public static <T> InternalFuture<T> whenComplete( InternalFuture<T> future, Runnable action )
     {
-        InternalPromise<T> result = new InternalPromise<>( future.eventLoop() );
+        InternalPromise<T> result = new InternalPromise<>( future.eventExecutor() );
         future.addListener( new CompletionListener<>( result, action ) );
         return result;
     }

driver/src/main/java/org/neo4j/driver/internal/async/InternalFuture.java

@@ -18,17 +18,15 @@
  */
 package org.neo4j.driver.internal.async;
 
-import io.netty.channel.EventLoop;
+import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
 
 import org.neo4j.driver.v1.Response;
 import org.neo4j.driver.v1.util.Function;
 
-public interface InternalFuture<T> extends Future<T>
+public interface InternalFuture<T> extends Future<T>, Response<T>
 {
-    EventLoop eventLoop();
-
-    Response<T> asResponse();
+    EventExecutor eventExecutor();
 
     <U> InternalFuture<U> thenApply( Function<T,U> fn );
 

driver/src/main/java/org/neo4j/driver/internal/async/InternalPromise.java

@@ -19,44 +19,39 @@
 package org.neo4j.driver.internal.async;
 
 import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.EventLoop;
+import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.Promise;
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import org.neo4j.driver.v1.Response;
+import org.neo4j.driver.v1.ResponseListener;
 import org.neo4j.driver.v1.util.Function;
 
 public class InternalPromise<T> implements InternalFuture<T>, Promise<T>
 {
-    private final EventLoop eventLoop;
+    private final EventExecutor eventExecutor;
     private final Promise<T> delegate;
 
     public InternalPromise( Bootstrap bootstrap )
     {
         this( bootstrap.config().group().next() );
     }
 
-    public InternalPromise( EventLoop eventLoop )
+    public InternalPromise( EventExecutor eventExecutor )
     {
-        this.eventLoop = eventLoop;
-        this.delegate = eventLoop.newPromise();
+        this.eventExecutor = eventExecutor;
+        this.delegate = eventExecutor.newPromise();
     }
 
     @Override
-    public EventLoop eventLoop()
+    public EventExecutor eventExecutor()
     {
-        return eventLoop;
-    }
-
-    @Override
-    public Response<T> asResponse()
-    {
-        return new InternalResponse<>( this );
+        return eventExecutor;
     }
 
     @Override
@@ -77,6 +72,26 @@ public InternalFuture<T> whenComplete( Runnable action )
         return Futures.whenComplete( this, action );
     }
 
+    @Override
+    public void addListener( final ResponseListener<T> listener )
+    {
+        delegate.addListener( new FutureListener<T>()
+        {
+            @Override
+            public void operationComplete( Future<T> future )
+            {
+                if ( future.isSuccess() )
+                {
+                    listener.operationCompleted( future.getNow(), null );
+                }
+                else
+                {
+                    listener.operationCompleted( null, future.cause() );
+                }
+            }
+        } );
+    }
+
     @Override
     public InternalPromise<T> setSuccess( T result )
     {

driver/src/main/java/org/neo4j/driver/internal/async/InternalResponse.java

@@ -49,13 +49,13 @@ public List<String> keys()
     @Override
     public Response<ResultSummary> summaryAsync()
     {
-        return pullAllHandler.summaryAsync().asResponse();
+        return pullAllHandler.summaryAsync();
     }
 
     @Override
     public Response<Boolean> fetchAsync()
     {
-        return pullAllHandler.fetchRecordAsync().asResponse();
+        return pullAllHandler.fetchRecordAsync();
     }
 
     @Override