Main 6.2

buildSrc/src/main/java/deltix/buildtools/SonatypeCentralPortalUploadRepositoryTask.java

@@ -32,9 +32,15 @@
 import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.net.URL;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Base64;
 
+import static java.nio.charset.StandardCharsets.US_ASCII;
+
 /**
  * This task performs manual steps to publish artifacts to Central Portal via OSSRH Staging API.
  */
@@ -110,16 +116,29 @@ public void run() throws IOException, InterruptedException {
             return;
         }
 
-        String userNameAndPassword = portalUsername.get() + ":" + portalPassword.get();
-        String bearer = Base64.getEncoder().encodeToString(userNameAndPassword.getBytes(StandardCharsets.US_ASCII));
+        final HttpClient httpClient = HttpClient.newBuilder()
+                .connectTimeout(Duration.ofMillis(CONNECTION_TIMEOUT))
+                .build();
+
+        final String userNameAndPassword = portalUsername.get() + ":" + portalPassword.get();
+        final String bearer = new String(
+                Base64.getEncoder().encode(userNameAndPassword.getBytes(US_ASCII)), US_ASCII);
+
+        final HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
+                .header("Authorization", "Bearer " + bearer);
+
         URI apiUri = URI.create(CENTRAL_PORTAL_OSSRH_API_URI);
 
         String repositoryKey = findOpenRepository(apiUri, bearer);
         System.out.println("Found published repository: " + repositoryKey);
 
-        int status = uploadRepository(apiUri, bearer, repositoryKey);
+        int status = uploadRepositoryToPortal(apiUri, httpClient, requestBuilder, repositoryKey);
         if (status == HttpURLConnection.HTTP_CLIENT_TIMEOUT)
-            uploadRepository(apiUri, bearer, repositoryKey);
+            uploadRepositoryToPortal(apiUri, httpClient, requestBuilder, repositoryKey);
+
+//        int status = uploadRepository(apiUri, bearer, repositoryKey);
+//        if (status == HttpURLConnection.HTTP_CLIENT_TIMEOUT)
+//            uploadRepository(apiUri, bearer, repositoryKey);
 
         dropRepository(apiUri, bearer, repositoryKey);
     }
@@ -189,6 +208,28 @@ private static int uploadRepository(URI apiUri, String bearer, String repository
         return status;
     }
 
+    private static int uploadRepositoryToPortal(
+            final URI apiUri,
+            final HttpClient httpClient,
+            final HttpRequest.Builder requestBuilder,
+            final String repositoryKey) throws IOException, InterruptedException {
+
+        HttpRequest request = requestBuilder
+                .copy()
+                .POST(HttpRequest.BodyPublishers.noBody())
+                .uri(apiUri.resolve("/manual/upload/repository/" + repositoryKey + "?publishing_type=automatic"))
+                .build();
+        HttpResponse<String> response = httpClient.send(
+                request, (HttpResponse.ResponseInfo responseInfo) -> HttpResponse.BodySubscribers.ofString(US_ASCII));
+
+        return response.statusCode();
+
+//        if (200 != response.statusCode()) {
+//            throw new IllegalStateException("Failed to upload repository: repository_key=" + repositoryKey +
+//                    ", status=" + response.statusCode() + ", response=" + response.body());
+//        }
+    }
+
     private static void dropRepository(URI apiUri, String bearer, String repositoryKey) throws IOException {
         String endpoint = apiUri.resolve("/manual/drop/repository/" + repositoryKey).toString();
         HttpURLConnection conn = (HttpURLConnection) new URL(endpoint).openConnection();

java/build.gradle

@@ -97,8 +97,8 @@ configure(leafProjects) {
     compileTestJava.options.compilerArgs += '--add-opens=java.base/java.lang=ALL-UNNAMED'
     compileTestJava.options.compilerArgs += '--add-opens=java.rmi/sun.rmi.transport=ALL-UNNAMED'
 
-    sourceCompatibility = 11
-    targetCompatibility = 11
+    sourceCompatibility = JavaVersion.VERSION_11
+    targetCompatibility = JavaVersion.VERSION_11
 
     configurations.all {
         resolutionStrategy {
@@ -301,7 +301,8 @@ configure(leafProjects) {
             dependency 'org.slf4j:slf4j-log4j12:1.7.10'
 
             dependency 'com.sun.xml.bind:jaxb-impl:2.3.0'
-            dependency 'junit:junit:4.13.1'
+
+            dependency "junit:junit:4.13.2"
 
             dependency 'com.nimbusds:nimbus-jose-jwt:9.40'
 
@@ -351,8 +352,17 @@ configure(leafProjects) {
         compileOnly 'com.google.code.findbugs:annotations'
 
         testCompileOnly 'com.google.code.findbugs:jsr305'
-        testCompile 'junit:junit:4.13.1'
-        testCompile 'org.mockito:mockito-core:1.10.19'
+
+        testImplementation "org.mockito:mockito-core:5.14.2"
+        testImplementation 'org.mockito:mockito-junit-jupiter:5.14.2'
+        testImplementation 'org.mockito:mockito-inline:4.11.0'
+
+        testImplementation 'junit:junit'
+        testImplementation 'org.junit.jupiter:junit-jupiter-api:5.12.2'
+
+        testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.12.2'
+        testRuntimeOnly 'org.junit.vintage:junit-vintage-engine:5.12.2'
+        testRuntimeOnly 'org.junit.platform:junit-platform-launcher:1.12.2'
     }
 
     spotbugs {

java/timebase/api/build.gradle

@@ -33,4 +33,7 @@ dependencies {
     // TODO: Move to client and server?
     implementation ('io.aeron:aeron-client')
     implementation ('io.aeron:aeron-driver')
+
+    // For simulation of connection loss scenarios
+    testImplementation 'com.github.netcrusherorg:netcrusher-core:0.10'
 }

java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelOutputStream.java

@@ -19,6 +19,7 @@
 import com.epam.deltix.util.concurrent.UncheckedInterruptedException;
 import com.epam.deltix.util.lang.Util;
 import net.jcip.annotations.GuardedBy;
+import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.IOException;
@@ -28,7 +29,7 @@
  * Date: Mar 25, 2010
  */
 public class ChannelOutputStream extends VSOutputStream {
-    //@ApiStatus.Experimental // Temporary option for testing performance effect of flushing single packet
+    @ApiStatus.Experimental // Temporary option for testing performance effect of flushing single packet
     private static final boolean SINGLE_SEND_ON_PARTIAL_FLUSH = Boolean.getBoolean("TimeBase.network.channel.singleSendOnPartialFlush");
 
     private final int                       maxCapacity;
@@ -87,12 +88,12 @@ public synchronized void    enableFlushing() throws IOException {
             // However, if we are above 75% capacity, we should flush all data
             // and block till all accumulated data is sent.
             // Otherwise, if the consumer too slow, the buffer will start to grow indefinitely.
-            // See https://gitlab.deltixhub.com/Deltix/QuantServer/QuantServer/-/issues/1298
             int buffer75percent = halfCapacity + (halfCapacity >> 1);
             boolean partialOk = size < buffer75percent;
             try {
                 flushInternal(partialOk, false);
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 throw new UncheckedInterruptedException(e);
             }
         }
@@ -269,6 +270,7 @@ else if (newSize <= buffer.length) {
                     send (b, off, len);
                 }
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 throw new UncheckedInterruptedException (e);
             }
         }
@@ -290,6 +292,7 @@ else if (size >= maxCapacity)
 
             buffer [size++] = (byte) b;
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new UncheckedInterruptedException (e);
         }
     }
@@ -367,4 +370,4 @@ public String toString() {
         return "ChannelOutputStream@" + Integer.toHexString(hashCode()) +
                 " for channel=" + channel;
     }
-}
+}

java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ConnectionStateListener.java

@@ -17,18 +17,31 @@
 package com.epam.deltix.util.vsocket;
 
 abstract class ConnectionStateListener {
-
+    /**
+     * Triggered when connection loss causes dispatcher to give up on recovery.
+     * <p>
+     * Triggered only once per dispatcher lifecycle.
+     * <p>
+     * Not triggered if dispatcher is stopped normally with {@link VSDispatcher#close()}.
+     */
     abstract void onDisconnected();
 
-    abstract void onReconnected();
+    /**
+     * Triggered when the first connection is established.
+     */
+    abstract void onConnected();
 
     /**
-     * @return true if transport is already known to be unrecoverable
+     * Triggered when transport is stopped (e.g. connection lost) but may be recoverable.
+     *
+     * @return true if transport is already known to be unrecoverable (and recovery should be stopped right away)
      */
-    abstract boolean onTransportStopped(VSocketRecoveryInfo recoveryInfo);
+    abstract boolean onTransportRecoveryStart(VSocketRecoveryInfo recoveryInfo);
 
     /**
+     * Triggered when transport recovery have to stop (because of timeout or dispatcher shutdown).
+     *
      * @return true if transport was permanently lost (can't be recovered anymore)
      */
-    abstract boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo);
-}
+    abstract boolean onTransportRecoveryStop(VSocketRecoveryInfo recoveryInfo);
+}

java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannelImpl.java

@@ -315,6 +315,7 @@ public void                 close (boolean terminate) {
             } catch (ConnectionAbortedException x) {
                 LOGGER.log (Level.FINE, "Error sending disconnect.", x);
             } catch (InterruptedException x) {
+                Thread.currentThread().interrupt();
                 LOGGER.log (Level.FINE, "Sending disconnect interrupted.", x);
             } catch (Exception x) {
                 LOGGER.log (Level.WARNING, "Error sending disconnect", x);