[DE-535] configure Vertx instance

.github/workflows/test.yml

@@ -32,6 +32,9 @@ jobs:
           - docker.io/arangodb/arangodb:3.12
           - docker.io/arangodb/enterprise:3.11
           - docker.io/arangodb/enterprise:3.12
+        starter-docker-img:
+          # support activefailover
+          - docker.io/arangodb/arangodb-starter:0.18.5
         topology:
           - single
           - cluster
@@ -60,6 +63,7 @@ jobs:
           ARANGO_LICENSE_KEY: ${{ secrets.ARANGO_LICENSE_KEY }}
           STARTER_MODE: ${{matrix.topology}}
           DOCKER_IMAGE: ${{matrix.docker-img}}
+          STARTER_DOCKER_IMAGE: ${{matrix.starter-docker-img}}
       - name: Info
         run: mvn -version
       - name: Test
@@ -127,6 +131,9 @@ jobs:
       matrix:
         docker-img:
           - docker.io/arangodb/enterprise:3.11
+        starter-docker-img:
+          # support activefailover
+          - docker.io/arangodb/arangodb-starter:0.18.5
         topology:
           - single
           - cluster
@@ -150,6 +157,7 @@ jobs:
           ARANGO_LICENSE_KEY: ${{ secrets.ARANGO_LICENSE_KEY }}
           STARTER_MODE: ${{matrix.topology}}
           DOCKER_IMAGE: ${{matrix.docker-img}}
+          STARTER_DOCKER_IMAGE: ${{matrix.starter-docker-img}}
       - name: Set JWT
         run: |
           ENDPOINT=$(./docker/find_active_endpoint.sh)

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

@@ -23,6 +23,7 @@
 import com.arangodb.arch.UnstableApi;
 import com.arangodb.config.ArangoConfigProperties;
 import com.arangodb.config.HostDescription;
+import com.arangodb.config.ProtocolConfig;
 import com.arangodb.entity.*;
 import com.arangodb.internal.ArangoDBImpl;
 import com.arangodb.internal.ArangoExecutorSync;
@@ -264,9 +265,9 @@ public interface ArangoDB extends ArangoSerdeAccessor {
      *
      * @param user        The name of the user
      * @param permissions The permissions the user grant
-     * @since ArangoDB 3.2.0
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/users/#set-a-users-database-access-level">API
      * Documentation</a>
+     * @since ArangoDB 3.2.0
      */
     void grantDefaultDatabaseAccess(String user, Permissions permissions);
 
@@ -276,9 +277,9 @@ public interface ArangoDB extends ArangoSerdeAccessor {
      *
      * @param user        The name of the user
      * @param permissions The permissions the user grant
-     * @since ArangoDB 3.2.0
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/users/#set-a-users-collection-access-level">API
      * Documentation</a>
+     * @since ArangoDB 3.2.0
      */
     void grantDefaultCollectionAccess(String user, Permissions permissions);
 
@@ -313,19 +314,19 @@ public interface ArangoDB extends ArangoSerdeAccessor {
      * Returns the server's current loglevel settings.
      *
      * @return the server's current loglevel settings
-     * @since ArangoDB 3.1.0
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/monitoring/logs/#get-the-server-log-levels">API
      * Documentation</a>
+     * @since ArangoDB 3.1.0
      */
     LogLevelEntity getLogLevel();
 
     /**
      * Returns the server's current loglevel settings.
      *
      * @return the server's current loglevel settings
-     * @since ArangoDB 3.10
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/monitoring/logs/#get-the-server-log-levels">API
      * Documentation</a>
+     * @since ArangoDB 3.10
      */
     LogLevelEntity getLogLevel(LogLevelOptions options);
 
@@ -334,9 +335,9 @@ public interface ArangoDB extends ArangoSerdeAccessor {
      *
      * @param entity loglevel settings
      * @return the server's current loglevel settings
-     * @since ArangoDB 3.1.0
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/monitoring/logs/#set-the-server-log-levels">API
      * Documentation</a>
+     * @since ArangoDB 3.1.0
      */
     LogLevelEntity setLogLevel(LogLevelEntity entity);
 
@@ -345,17 +346,17 @@ public interface ArangoDB extends ArangoSerdeAccessor {
      *
      * @param entity loglevel settings
      * @return the server's current loglevel settings
-     * @since ArangoDB 3.10
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/monitoring/logs/#set-the-server-log-levels">API
      * Documentation</a>
+     * @since ArangoDB 3.10
      */
     LogLevelEntity setLogLevel(LogLevelEntity entity, LogLevelOptions options);
 
     /**
      * @return the list of available rules and their respective flags
-     * @since ArangoDB 3.10
      * @see <a href="https://docs.arangodb.com/stable/develop/http-api/queries/aql-queries/#list-all-aql-optimizer-rules">API
      * Documentation</a>
+     * @since ArangoDB 3.10
      */
     Collection<QueryOptimizerRule> getQueryOptimizerRules();
 
@@ -381,7 +382,7 @@ public ArangoDB build() {
             ProtocolProvider protocolProvider = protocolProvider(config.getProtocol());
             config.setProtocolModule(protocolProvider.protocolModule());
 
-            ConnectionFactory connectionFactory = protocolProvider.createConnectionFactory();
+            ConnectionFactory connectionFactory = protocolProvider.createConnectionFactory(config.getProtocolConfig());
             Collection<Host> hostList = createHostList(connectionFactory);
             HostResolver hostResolver = createHostResolver(hostList, connectionFactory);
             HostHandler hostHandler = createHostHandler(hostResolver);
@@ -680,6 +681,15 @@ public Builder compressionLevel(Integer level) {
             return this;
         }
 
+        /**
+         * Configuration specific for {@link com.arangodb.internal.net.ProtocolProvider}.
+         * @return {@link ArangoDB.Builder}
+         */
+        public Builder protocolConfig(ProtocolConfig protocolConfig) {
+            config.setProtocolConfig(protocolConfig);
+            return this;
+        }
+
         @UnstableApi
         protected ProtocolProvider protocolProvider(Protocol protocol) {
             ServiceLoader<ProtocolProvider> loader = ServiceLoader.load(ProtocolProvider.class);

core/src/main/java/com/arangodb/config/ArangoConfigProperties.java

@@ -110,4 +110,8 @@ default Optional<Integer> getCompressionLevel() {
         return Optional.empty();
     }
 
+    default Optional<Boolean> getReuseVertx() {
+        return Optional.empty();
+    }
+
 }

core/src/main/java/com/arangodb/config/ProtocolConfig.java

@@ -0,0 +1,7 @@
+package com.arangodb.config;
+
+/**
+ * Configuration specific for {@link com.arangodb.internal.net.ProtocolProvider}.
+ */
+public interface ProtocolConfig {
+}

core/src/main/java/com/arangodb/internal/config/ArangoConfig.java

@@ -7,6 +7,7 @@
 import com.arangodb.arch.UsedInApi;
 import com.arangodb.config.ArangoConfigProperties;
 import com.arangodb.config.HostDescription;
+import com.arangodb.config.ProtocolConfig;
 import com.arangodb.entity.LoadBalancingStrategy;
 import com.arangodb.internal.ArangoDefaults;
 import com.arangodb.internal.serde.ContentTypeFactory;
@@ -49,6 +50,7 @@ public class ArangoConfig {
     private Compression compression;
     private Integer compressionThreshold;
     private Integer compressionLevel;
+    private ProtocolConfig protocolConfig;
 
     private static final Logger LOG = LoggerFactory.getLogger(ArangoConfig.class);
 
@@ -329,4 +331,12 @@ public Integer getCompressionLevel() {
     public void setCompressionLevel(Integer compressionLevel) {
         this.compressionLevel = compressionLevel;
     }
+
+    public ProtocolConfig getProtocolConfig() {
+        return protocolConfig;
+    }
+
+    public void setProtocolConfig(ProtocolConfig protocolConfig) {
+        this.protocolConfig = protocolConfig;
+    }
 }

core/src/main/java/com/arangodb/internal/net/ProtocolProvider.java

@@ -3,6 +3,7 @@
 
 import com.arangodb.Protocol;
 import com.arangodb.arch.UsedInApi;
+import com.arangodb.config.ProtocolConfig;
 import com.arangodb.internal.config.ArangoConfig;
 import com.fasterxml.jackson.databind.Module;
 
@@ -11,7 +12,17 @@ public interface ProtocolProvider {
 
     boolean supportsProtocol(Protocol protocol);
 
-    ConnectionFactory createConnectionFactory();
+    /**
+     * @deprecated use {@link #createConnectionFactory(ProtocolConfig)} instead
+     */
+    @Deprecated
+    default ConnectionFactory createConnectionFactory() {
+        throw new UnsupportedOperationException();
+    }
+
+    default ConnectionFactory createConnectionFactory(ProtocolConfig config) {
+        return createConnectionFactory();
+    }
 
     CommunicationProtocol createProtocol(ArangoConfig config, HostHandler hostHandler);
 

docker/start_db.sh

@@ -3,6 +3,7 @@
 # Configuration environment variables:
 #   STARTER_MODE:             (single|cluster|activefailover), default single
 #   DOCKER_IMAGE:             ArangoDB docker image, default docker.io/arangodb/arangodb:latest
+#   STARTER_DOCKER_IMAGE:     ArangoDB Starter docker image, default docker.io/arangodb/arangodb-starter:latest
 #   SSL:                      (true|false), default false
 #   ARANGO_LICENSE_KEY:       only required for ArangoDB Enterprise
 
@@ -11,10 +12,10 @@
 
 STARTER_MODE=${STARTER_MODE:=single}
 DOCKER_IMAGE=${DOCKER_IMAGE:=docker.io/arangodb/arangodb:latest}
+STARTER_DOCKER_IMAGE=${STARTER_DOCKER_IMAGE:=docker.io/arangodb/arangodb-starter:latest}
 SSL=${SSL:=false}
 COMPRESSION=${COMPRESSION:=false}
 
-STARTER_DOCKER_IMAGE=docker.io/arangodb/arangodb-starter:latest
 GW=172.28.0.1
 docker network create arangodb --subnet 172.28.0.0/16
 

driver/src/test/java/com/arangodb/ArangoConfigTest.java

@@ -1,5 +1,6 @@
 package com.arangodb;
 
+import com.arangodb.http.HttpProtocolConfig;
 import com.arangodb.internal.ArangoDefaults;
 import com.arangodb.internal.config.ArangoConfig;
 import org.junit.jupiter.api.Test;
@@ -8,7 +9,7 @@
 
 public class ArangoConfigTest {
     @Test
-    void defaultValues() {
+    void ArangoConfigDefaultValues() {
         ArangoConfig cfg = new ArangoConfig();
         assertThat(cfg.getHosts()).isEqualTo(ArangoDefaults.DEFAULT_HOSTS);
         assertThat(cfg.getProtocol()).isEqualTo(Protocol.HTTP2_JSON);
@@ -31,5 +32,13 @@ void defaultValues() {
         assertThat(cfg.getCompression()).isEqualTo(ArangoDefaults.DEFAULT_COMPRESSION);
         assertThat(cfg.getCompressionThreshold()).isEqualTo(ArangoDefaults.DEFAULT_COMPRESSION_THRESHOLD);
         assertThat(cfg.getCompressionLevel()).isEqualTo(ArangoDefaults.DEFAULT_COMPRESSION_LEVEL);
+        assertThat(cfg.getProtocolConfig()).isNull();
     }
+
+    @Test
+    void HttpProtocolConfigDefaultValues() {
+        HttpProtocolConfig cfg = HttpProtocolConfig.builder().build();
+        assertThat(cfg.getVertx()).isNull();
+    }
+
 }

driver/src/test/java/com/arangodb/ArangoDBAsyncTest.java

@@ -234,6 +234,7 @@ void getAccessibleDatabasesFor(ArangoDBAsync arangoDB) throws ExecutionException
     @ParameterizedTest
     @MethodSource("asyncArangos")
     void createUser(ArangoDBAsync arangoDB) throws ExecutionException, InterruptedException {
+        assumeTrue(isSingleServer());
         String username = "user-" + UUID.randomUUID();
         final UserEntity result = arangoDB.createUser(username, PW, null).get();
         assertThat(result.getUser()).isEqualTo(username);

driver/src/test/java/perf/SimpleSyncPerfTest.java

@@ -21,13 +21,16 @@
 package perf;
 
 import com.arangodb.ArangoDB;
+import com.arangodb.BaseJunit5;
 import com.arangodb.Protocol;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
 import java.util.Date;
 
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
 /**
  * @author Michele Rastelli
  */
@@ -44,6 +47,8 @@ private void doGetVersion(ArangoDB arangoDB) {
     @ParameterizedTest
     @EnumSource(Protocol.class)
     void getVersion(Protocol protocol) throws InterruptedException {
+        assumeTrue(!protocol.equals(Protocol.VST) || BaseJunit5.isLessThanVersion(3, 12));
+
         ArangoDB arangoDB = new ArangoDB.Builder()
                 .host("172.28.0.1", 8529)
                 .password("test")

http/src/main/java/com/arangodb/http/HttpConnection.java

@@ -75,19 +75,19 @@ public class HttpConnection implements Connection {
     private static final String CONTENT_TYPE_VPACK = "application/x-velocypack";
     private static final String USER_AGENT = getUserAgent();
     private static final AtomicInteger THREAD_COUNT = new AtomicInteger();
-    private String auth;
+    private volatile String auth;
     private final int compressionThreshold;
     private final Encoder encoder;
     private final WebClient client;
     private final Integer timeout;
     private final MultiMap commonHeaders = MultiMap.caseInsensitiveMultiMap();
-    private final Vertx vertx;
+    private final Vertx vertxToClose;
 
     private static String getUserAgent() {
         return "JavaDriver/" + PackageVersion.VERSION + " (JVM/" + System.getProperty("java.specification.version") + ")";
     }
 
-    HttpConnection(final ArangoConfig config, final HostDescription host) {
+    HttpConnection(final ArangoConfig config, final HostDescription host, final Vertx existingVertx) {
         super();
         Protocol protocol = config.getProtocol();
         ContentType contentType = ContentTypeFactory.of(protocol);
@@ -108,14 +108,25 @@ private static String getUserAgent() {
         }
         commonHeaders.add("x-arango-driver", USER_AGENT);
         timeout = config.getTimeout();
-        vertx = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true).setEventLoopPoolSize(1));
-        vertx.runOnContext(e -> {
-            Thread.currentThread().setName("adb-http-" + THREAD_COUNT.getAndIncrement());
-            auth = new UsernamePasswordCredentials(
-                    config.getUser(), Optional.ofNullable(config.getPassword()).orElse("")
-            ).toHttpAuthorization();
-            LOGGER.debug("Created Vert.x context");
-        });
+        auth = new UsernamePasswordCredentials(
+                config.getUser(), Optional.ofNullable(config.getPassword()).orElse("")
+        ).toHttpAuthorization();
+
+        Vertx vertxToUse;
+        if (existingVertx != null) {
+            // reuse existing Vert.x
+            vertxToUse = existingVertx;
+            // Vert.x will not be closed when connection is closed
+            vertxToClose = null;
+            LOGGER.debug("Reusing existing Vert.x instance");
+        } else {
+            // create a new Vert.x instance
+            LOGGER.debug("Creating new Vert.x instance");
+            vertxToUse = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true).setEventLoopPoolSize(1));
+            vertxToUse.runOnContext(e -> Thread.currentThread().setName("adb-http-" + THREAD_COUNT.getAndIncrement()));
+            // Vert.x be closed when connection is closed
+            vertxToClose = vertxToUse;
+        }
 
         int intTtl = Optional.ofNullable(config.getConnectionTtl())
                 .map(ttl -> Math.toIntExact(ttl / 1000))
@@ -193,7 +204,7 @@ public SslContextFactory sslContextFactory() {
                     });
         }
 
-        client = WebClient.create(vertx, webClientOptions);
+        client = WebClient.create(vertxToUse, webClientOptions);
     }
 
     private static String buildUrl(final InternalRequest request) {
@@ -229,7 +240,10 @@ private static void addHeader(final InternalRequest request, final HttpRequest<?
     @Override
     public void close() {
         client.close();
-        vertx.close();
+        if (vertxToClose != null) {
+            LOGGER.debug("Closing Vert.x instance");
+            vertxToClose.close();
+        }
     }
 
     private HttpMethod requestTypeToHttpMethod(RequestType requestType) {
@@ -254,7 +268,7 @@ private HttpMethod requestTypeToHttpMethod(RequestType requestType) {
     @UnstableApi
     public CompletableFuture<InternalResponse> executeAsync(@UnstableApi final InternalRequest request) {
         CompletableFuture<InternalResponse> rfuture = new CompletableFuture<>();
-        vertx.runOnContext(e -> doExecute(request, rfuture));
+        doExecute(request, rfuture);
         return rfuture;
     }
 
@@ -308,7 +322,7 @@ private InternalResponse buildResponse(final HttpResponse<Buffer> httpResponse)
     @Override
     public void setJwt(String jwt) {
         if (jwt != null) {
-            vertx.runOnContext(e -> auth = new TokenCredentials(jwt).toHttpAuthorization());
+            auth = new TokenCredentials(jwt).toHttpAuthorization();
         }
     }