Merge branch 'main' into feature/system-props

Author: bevzzz

pom.xml

@@ -170,9 +170,8 @@
     </dependency>
     <dependency>
         <groupId>org.slf4j</groupId>
-        <artifactId>slf4j-nop</artifactId>
+        <artifactId>slf4j-api</artifactId>
         <version>${slf4j.version}</version>
-        <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.mock-server</groupId>
@@ -237,13 +236,11 @@
           <artifactId>maven-surefire-plugin</artifactId>
           <version>3.5.4</version>
           <configuration>
-            <trimStackTrace>false</trimStackTrace>
-            <parallel>classes</parallel>
-            <forkCount>4</forkCount>
-            <reuseForks>true</reuseForks>
-            <threadCount>1</threadCount>
-            <perCoreThreadCount>true</perCoreThreadCount>
-            <runOrder>balanced</runOrder>
+            <systemPropertyVariables>
+              <java.util.logging.config.file>
+                ${project.basedir}/src/test/resources/logging.properties
+              </java.util.logging.config.file>
+            </systemPropertyVariables>
           </configuration>
         </plugin>
         <plugin>
@@ -546,5 +543,5 @@
         <artifactId>central-publishing-maven-plugin</artifactId>
       </plugin>
     </plugins>
-  </build>
+    </build>
 </project>

src/main/java/io/weaviate/client6/v1/internal/TransportOptions.java

@@ -64,4 +64,20 @@ public TrustManagerFactory trustManagerFactory() {
   public Proxy proxy() {
     return this.proxy;
   }
+  
+  /**
+   * isWeaviateDomain returns true if the host matches weaviate.io,
+   * semi.technology, or weaviate.cloud domain.
+   */
+  public static boolean isWeaviateDomain(String host) {
+    var lower = host.toLowerCase();
+    return lower.contains("weaviate.io") ||
+        lower.contains("semi.technology") ||
+        lower.contains("weaviate.cloud");
+  }
+
+  public static boolean isGoogleCloudDomain(String host) {
+    var lower = host.toLowerCase();
+    return lower.contains("gcp");
+  }
 }

src/main/java/io/weaviate/client6/v1/internal/grpc/DefaultGrpcTransport.java

@@ -1,5 +1,14 @@
 package io.weaviate.client6.v1.internal.grpc;
 
+import static java.util.Objects.requireNonNull;
+
+import java.util.OptionalInt;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+import javax.net.ssl.SSLException;
+
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -11,11 +20,14 @@
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
 import io.grpc.stub.AbstractStub;
 import io.grpc.stub.MetadataUtils;
+import io.grpc.stub.StreamObserver;
 import io.weaviate.client6.v1.api.WeaviateApiException;
 import io.weaviate.client6.v1.internal.Proxy;
 import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateGrpc;
 import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateGrpc.WeaviateBlockingStub;
 import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateGrpc.WeaviateFutureStub;
+import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateProtoBatch.BatchStreamReply;
+import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateProtoBatch.BatchStreamRequest;
 
 import javax.net.ssl.SSLException;
 import java.net.InetSocketAddress;
@@ -24,51 +36,48 @@
 import java.util.concurrent.TimeUnit;
 
 public final class DefaultGrpcTransport implements GrpcTransport {
+  /**
+   * ListenableFuture callbacks are executed
+   * in the same thread they are called from.
+   */
+  private static final Executor FUTURE_CALLBACK_EXECUTOR = Runnable::run;
+
+  private final GrpcChannelOptions transportOptions;
   private final ManagedChannel channel;
 
   private final WeaviateBlockingStub blockingStub;
   private final WeaviateFutureStub futureStub;
 
-  private final GrpcChannelOptions transportOptions;
-
   private TokenCallCredentials callCredentials;
 
   public DefaultGrpcTransport(GrpcChannelOptions transportOptions) {
-    this.transportOptions = transportOptions;
-    this.channel = buildChannel(transportOptions);
-
-    var blockingStub = WeaviateGrpc.newBlockingStub(channel)
-        .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(transportOptions.headers()));
-
-    var futureStub = WeaviateGrpc.newFutureStub(channel)
-        .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(transportOptions.headers()));
-
-    if (transportOptions.maxMessageSize() != null) {
-      var max = transportOptions.maxMessageSize();
-      blockingStub = blockingStub.withMaxInboundMessageSize(max).withMaxOutboundMessageSize(max);
-      futureStub = futureStub.withMaxInboundMessageSize(max).withMaxOutboundMessageSize(max);
-    }
+    requireNonNull(transportOptions, "transportOptions is null");
 
+    this.transportOptions = transportOptions;
     if (transportOptions.tokenProvider() != null) {
       this.callCredentials = new TokenCallCredentials(transportOptions.tokenProvider());
-      blockingStub = blockingStub.withCallCredentials(callCredentials);
-      futureStub = futureStub.withCallCredentials(callCredentials);
     }
 
-    this.blockingStub = blockingStub;
-    this.futureStub = futureStub;
+    this.channel = buildChannel(transportOptions);
+    this.blockingStub = configure(WeaviateGrpc.newBlockingStub(channel));
+    this.futureStub = configure(WeaviateGrpc.newFutureStub(channel));
   }
 
   private <StubT extends AbstractStub<StubT>> StubT applyTimeout(StubT stub, Rpc<?, ?, ?, ?> rpc) {
     if (transportOptions.timeout() == null) {
       return stub;
     }
-    var timeout = rpc.isInsert()
+    int timeout = rpc.isInsert()
         ? transportOptions.timeout().insertSeconds()
         : transportOptions.timeout().querySeconds();
     return stub.withDeadlineAfter(timeout, TimeUnit.SECONDS);
   }
 
+  @Override
+  public OptionalInt maxMessageSizeBytes() {
+    return transportOptions.maxMessageSize();
+  }
+
   @Override
   public <RequestT, RequestM, ReplyM, ResponseT> ResponseT performRequest(RequestT request,
       Rpc<RequestT, RequestM, ResponseT, ReplyM> rpc) {
@@ -98,7 +107,9 @@ public <RequestT, RequestM, ReplyM, ResponseT> CompletableFuture<ResponseT> perf
    * reusing the thread in which the original future is completed.
    */
   private static final <T> CompletableFuture<T> toCompletableFuture(ListenableFuture<T> listenable) {
-    var completable = new CompletableFuture<T>();
+    requireNonNull(listenable, "listenable is null");
+
+    CompletableFuture<T> completable = new CompletableFuture<>();
     Futures.addCallback(listenable, new FutureCallback<T>() {
 
       @Override
@@ -115,13 +126,14 @@ public void onFailure(Throwable t) {
         completable.completeExceptionally(t);
       }
 
-    }, Runnable::run);
+    }, FUTURE_CALLBACK_EXECUTOR);
     return completable;
   }
 
   private static ManagedChannel buildChannel(GrpcChannelOptions transportOptions) {
-    var channel = NettyChannelBuilder.forAddress(transportOptions.host(), transportOptions.port());
+    requireNonNull(transportOptions, "transportOptions is null");
 
+    NettyChannelBuilder channel = NettyChannelBuilder.forAddress(transportOptions.host(), transportOptions.port());
     if (transportOptions.isSecure()) {
       channel.useTransportSecurity();
     } else {
@@ -163,15 +175,39 @@ private static ManagedChannel buildChannel(GrpcChannelOptions transportOptions)
     }
 
     channel.intercept(MetadataUtils.newAttachHeadersInterceptor(transportOptions.headers()));
-
     return channel.build();
   }
 
+  @Override
+  public StreamObserver<BatchStreamRequest> createStream(StreamObserver<BatchStreamReply> recv) {
+    return configure(WeaviateGrpc.newStub(channel)).batchStream(recv);
+  }
+
+  /** Apply common configuration to a stub. */
+  private <S extends AbstractStub<S>> S configure(S stub) {
+    requireNonNull(stub, "stub is null");
+
+    stub = stub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(transportOptions.headers()));
+    if (transportOptions.maxMessageSize().isPresent()) {
+      int max = transportOptions.maxMessageSize().getAsInt();
+      stub = stub.withMaxInboundMessageSize(max).withMaxOutboundMessageSize(max);
+    }
+    if (callCredentials != null) {
+      stub = stub.withCallCredentials(callCredentials);
+    }
+    return stub;
+  }
+
   @Override
   public void close() throws Exception {
     channel.shutdown();
     if (callCredentials != null) {
       callCredentials.close();
     }
   }
+
+  @Override
+  public String host() {
+    return transportOptions.host();
+  }
 }

src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcChannelOptions.java

@@ -1,6 +1,7 @@
 package io.weaviate.client6.v1.internal.grpc;
 
 import java.util.Map;
+import java.util.OptionalInt;
 
 import javax.net.ssl.TrustManagerFactory;
 
@@ -11,25 +12,25 @@
 import io.weaviate.client6.v1.internal.TransportOptions;
 
 public class GrpcChannelOptions extends TransportOptions<Metadata> {
-  private final Integer maxMessageSize;
+  private final OptionalInt maxMessageSize;
 
   public GrpcChannelOptions(String scheme, String host, int port, Map<String, String> headers,
       TokenProvider tokenProvider, TrustManagerFactory tmf, Timeout timeout, Proxy proxy) {
     this(scheme, host, port, buildMetadata(headers), tokenProvider, tmf, null, timeout, proxy);
   }
 
   private GrpcChannelOptions(String scheme, String host, int port, Metadata headers,
-      TokenProvider tokenProvider, TrustManagerFactory tmf, Integer maxMessageSize, Timeout timeout, Proxy proxy) {
+      TokenProvider tokenProvider, TrustManagerFactory tmf, OptionalInt maxMessageSize, Timeout timeout, Proxy proxy) {
     super(scheme, host, port, headers, tokenProvider, tmf, timeout, proxy);
     this.maxMessageSize = maxMessageSize;
   }
 
   public GrpcChannelOptions withMaxMessageSize(int maxMessageSize) {
-    return new GrpcChannelOptions(scheme, host, port, headers, tokenProvider, trustManagerFactory, maxMessageSize,
-        timeout, proxy);
+    return new GrpcChannelOptions(scheme, host, port, headers, tokenProvider, trustManagerFactory,
+        OptionalInt.of(maxMessageSize), timeout, proxy);
   }
 
-  public Integer maxMessageSize() {
+  public OptionalInt maxMessageSize() {
     return maxMessageSize;
   }