[DE-511] Retriable cursor (#505)

* removed MetaAware

* retriable cursor

* tests

* resilience test

* test fix

* resume cursor

* doc

* doc

Author: rashtao

core/src/main/java/com/arangodb/ArangoCursor.java

@@ -22,10 +22,12 @@
 
 import com.arangodb.entity.CursorStats;
 import com.arangodb.entity.CursorWarning;
+import com.arangodb.model.AqlQueryOptions;
 
 import java.io.Closeable;
 import java.util.Collection;
 import java.util.List;
+import java.util.NoSuchElementException;
 
 /**
  * @author Mark Vollmary
@@ -76,4 +78,27 @@ public interface ArangoCursor<T> extends ArangoIterable<T>, ArangoIterator<T>, C
      */
     boolean isPotentialDirtyRead();
 
+    /**
+     * @return The ID of the batch after the current one. The first batch has an ID of 1 and the value is incremented by
+     * 1 with every batch. Only set if the allowRetry query option is enabled.
+     * @since ArangoDB 3.11
+     */
+    String getNextBatchId();
+
+    /**
+     * Returns the next element in the iteration.
+     * <p/>
+     * If the cursor allows retries (see {@link AqlQueryOptions#allowRetry(Boolean)}), then it is safe to retry invoking
+     * this method in case of I/O exceptions (which are actually thrown as {@link com.arangodb.ArangoDBException} with
+     * cause {@link java.io.IOException}).
+     * <p/>
+     * If the cursor does not allow retries (default), then it is not safe to retry invoking this method in case of I/O
+     * exceptions, since the request to fetch the next batch is not idempotent (i.e. the cursor may advance multiple
+     * times on the server).
+     *
+     * @return the next element in the iteration
+     * @throws NoSuchElementException if the iteration has no more elements
+     */
+    @Override
+    T next();
 }

core/src/main/java/com/arangodb/ArangoDatabase.java

@@ -310,6 +310,20 @@ public interface ArangoDatabase extends ArangoSerdeAccessor {
      */
     <T> ArangoCursor<T> cursor(String cursorId, Class<T> type);
 
+    /**
+     * Return an cursor from the given cursor-ID if still existing
+     *
+     * @param cursorId    The ID of the cursor
+     * @param type        The type of the result (POJO or {@link com.arangodb.util.RawData})
+     * @param nextBatchId The ID of the next cursor batch (set only if cursor allows retries, see
+     *                    {@link AqlQueryOptions#allowRetry(Boolean)}
+     * @return cursor of the results
+     * @see <a href= "https://www.arangodb.com/docs/stable/http/aql-query-cursor-accessing-cursors
+     * .html#read-next-batch-from-cursor">API Documentation</a>
+     * @since ArangoDB 3.11
+     */
+    <T> ArangoCursor<T> cursor(String cursorId, Class<T> type, String nextBatchId);
+
     /**
      * Explain an AQL query and return information about it
      *

core/src/main/java/com/arangodb/entity/MetaAware.java

@@ -22,15 +22,14 @@
 
 import com.arangodb.internal.cursor.entity.InternalCursorEntity;
 
-import java.util.Map;
 
 /**
  * @author Mark Vollmary
  */
 public interface ArangoCursorExecute {
 
-    InternalCursorEntity next(String id, Map<String, String> meta);
+    InternalCursorEntity next(String id, String nextBatchId);
 
-    void close(String id, Map<String, String> meta);
+    void close(String id);
 
 }

core/src/main/java/com/arangodb/internal/ArangoCursorExecute.java

@@ -161,7 +161,7 @@ public <T> ArangoCursor<T> query(
             final String query, final Class<T> type, final Map<String, Object> bindVars, final AqlQueryOptions options) {
         final InternalRequest request = queryRequest(query, bindVars, options);
         final HostHandle hostHandle = new HostHandle();
-        final InternalCursorEntity result = executor.execute(request, InternalCursorEntity.class, hostHandle);
+        final InternalCursorEntity result = executor.execute(request, internalCursorEntityDeserializer(), hostHandle);
         return createCursor(result, type, options, hostHandle);
     }
 
@@ -183,8 +183,20 @@ public <T> ArangoCursor<T> query(final String query, final Class<T> type) {
     @Override
     public <T> ArangoCursor<T> cursor(final String cursorId, final Class<T> type) {
         final HostHandle hostHandle = new HostHandle();
-        final InternalCursorEntity result = executor
-                .execute(queryNextRequest(cursorId, null, null), InternalCursorEntity.class, hostHandle);
+        final InternalCursorEntity result = executor.execute(
+                queryNextRequest(cursorId, null),
+                internalCursorEntityDeserializer(),
+                hostHandle);
+        return createCursor(result, type, null, hostHandle);
+    }
+
+    @Override
+    public <T> ArangoCursor<T> cursor(final String cursorId, final Class<T> type, final String nextBatchId) {
+        final HostHandle hostHandle = new HostHandle();
+        final InternalCursorEntity result = executor.execute(
+                queryNextByBatchIdRequest(cursorId, nextBatchId, null),
+                internalCursorEntityDeserializer(),
+                hostHandle);
         return createCursor(result, type, null, hostHandle);
     }
 
@@ -196,13 +208,15 @@ private <T> ArangoCursor<T> createCursor(
 
         final ArangoCursorExecute execute = new ArangoCursorExecute() {
             @Override
-            public InternalCursorEntity next(final String id, Map<String, String> meta) {
-                return executor.execute(queryNextRequest(id, options, meta), InternalCursorEntity.class, hostHandle);
+            public InternalCursorEntity next(final String id, final String nextBatchId) {
+                InternalRequest request = nextBatchId == null ?
+                        queryNextRequest(id, options) : queryNextByBatchIdRequest(id, nextBatchId, options);
+                return executor.execute(request, internalCursorEntityDeserializer(), hostHandle);
             }
 
             @Override
-            public void close(final String id, Map<String, String> meta) {
-                executor.execute(queryCloseRequest(id, options, meta), Void.class, hostHandle);
+            public void close(final String id) {
+                executor.execute(queryCloseRequest(id, options), Void.class, hostHandle);
             }
         };
 

core/src/main/java/com/arangodb/internal/ArangoDatabaseImpl.java

@@ -21,12 +21,9 @@
 package com.arangodb.internal;
 
 import com.arangodb.ArangoDBException;
-import com.arangodb.entity.MetaAware;
 import com.arangodb.internal.config.ArangoConfig;
 import com.arangodb.internal.net.CommunicationProtocol;
 import com.arangodb.internal.net.HostHandle;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.lang.reflect.Type;
@@ -36,8 +33,6 @@
  */
 public class ArangoExecutorSync extends ArangoExecutor {
 
-    private static final Logger LOG = LoggerFactory.getLogger(ArangoExecutorSync.class);
-
     private final CommunicationProtocol protocol;
 
     public ArangoExecutorSync(final CommunicationProtocol protocol, final ArangoConfig config) {
@@ -64,14 +59,7 @@ public <T> T execute(
 
         final InternalResponse response = protocol.execute(interceptRequest(request), hostHandle);
         interceptResponse(response);
-        T deserialize = responseDeserializer.deserialize(response);
-
-        if (deserialize instanceof MetaAware) {
-            LOG.debug("Response is MetaAware {}", deserialize.getClass().getName());
-            ((MetaAware) deserialize).setMeta(response.getMeta());
-        }
-
-        return deserialize;
+        return responseDeserializer.deserialize(response);
     }
 
     public void disconnect() {

core/src/main/java/com/arangodb/internal/ArangoExecutorSync.java

@@ -23,6 +23,7 @@
 import com.arangodb.entity.*;
 import com.arangodb.entity.arangosearch.analyzer.SearchAnalyzer;
 import com.arangodb.internal.ArangoExecutor.ResponseDeserializer;
+import com.arangodb.internal.cursor.entity.InternalCursorEntity;
 import com.arangodb.internal.util.RequestUtils;
 import com.arangodb.model.*;
 import com.arangodb.model.arangosearch.*;
@@ -156,27 +157,30 @@ protected InternalRequest queryRequest(final String query, final Map<String, Obj
         return request;
     }
 
-    protected InternalRequest queryNextRequest(final String id, final AqlQueryOptions options, Map<String, String> meta) {
-
+    protected InternalRequest queryNextRequest(final String id, final AqlQueryOptions options) {
         final InternalRequest request = request(name, RequestType.POST, PATH_API_CURSOR, id);
-        request.putHeaderParams(meta);
+        return completeQueryNextRequest(request, options);
+    }
 
-        final AqlQueryOptions opt = options != null ? options : new AqlQueryOptions();
+    protected InternalRequest queryNextByBatchIdRequest(final String id,
+                                                        final String nextBatchId,
+                                                        final AqlQueryOptions options) {
+        final InternalRequest request = request(name, RequestType.POST, PATH_API_CURSOR, id, nextBatchId);
+        return completeQueryNextRequest(request, options);
+    }
 
+    private InternalRequest completeQueryNextRequest(final InternalRequest request, final AqlQueryOptions options) {
+        final AqlQueryOptions opt = options != null ? options : new AqlQueryOptions();
         if (Boolean.TRUE.equals(opt.getAllowDirtyRead())) {
             RequestUtils.allowDirtyRead(request);
         }
         request.putHeaderParam(TRANSACTION_ID, opt.getStreamTransactionId());
         return request;
     }
 
-    protected InternalRequest queryCloseRequest(final String id, final AqlQueryOptions options, Map<String, String> meta) {
-
+    protected InternalRequest queryCloseRequest(final String id, final AqlQueryOptions options) {
         final InternalRequest request = request(name, RequestType.DELETE, PATH_API_CURSOR, id);
-        request.putHeaderParams(meta);
-
         final AqlQueryOptions opt = options != null ? options : new AqlQueryOptions();
-
         if (Boolean.TRUE.equals(opt.getAllowDirtyRead())) {
             RequestUtils.allowDirtyRead(request);
         }
@@ -243,6 +247,15 @@ protected InternalRequest deleteAqlFunctionRequest(final String name, final AqlF
         return request;
     }
 
+    protected ResponseDeserializer<InternalCursorEntity> internalCursorEntityDeserializer() {
+        return response -> {
+            InternalCursorEntity e = getSerde().deserialize(response.getBody(), InternalCursorEntity.class);
+            boolean potentialDirtyRead = Boolean.parseBoolean(response.getMeta("X-Arango-Potential-Dirty-Read"));
+            e.setPontentialDirtyRead(potentialDirtyRead);
+            return e;
+        };
+    }
+
     protected ResponseDeserializer<Integer> deleteAqlFunctionResponseDeserializer() {
         return response -> getSerde().deserialize(response.getBody(), "/deletedCount", Integer.class);
     }