fix: initialization and defaults for timeout

Author: klopfdreh

client/src/main/java/io/micrometer/prometheus/rsocket/PrometheusRSocketClient.java

@@ -65,7 +65,7 @@ public class PrometheusRSocketClient {
   private volatile boolean requestedDisconnect = false;
   private RSocket sendingSocket;
 
-  private Duration timeout = Duration.ofSeconds(5);
+  private Duration timeout;
 
   /**
    * Creates a {@link PrometheusRSocketClient}.
@@ -97,6 +97,7 @@ private PrometheusRSocketClient(MeterRegistryAndScrape<?> registryAndScrape,
                                   Duration timeout,
                                   Runnable onKeyReceived) {
     this.registryAndScrape = registryAndScrape;
+    this.timeout = timeout;
 
     RSocketConnector.create()
         .reconnect(new Retry() {
@@ -301,7 +302,7 @@ public static class Builder {
     private Retry retry = Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(10))
         .maxBackoff(Duration.ofMinutes(10));
 
-    private Duration timeout;
+    private Duration timeout = Duration.ofSeconds(5);
 
     private Runnable onKeyReceived = () -> {
     };
@@ -350,6 +351,17 @@ public Builder doOnKeyReceived(Runnable onKeyReceived) {
      * @return the {@link PrometheusRSocketClient}
      */
     public PrometheusRSocketClient connect() {
+      return connect(timeout);
+    }
+
+    /**
+     * Connects the {@link PrometheusRSocketClient}.
+     *
+     * @param timeout the timeout for the client to connect
+     *
+     * @return the {@link PrometheusRSocketClient}
+     */
+    public PrometheusRSocketClient connect(Duration timeout) {
       LOGGER.debug("Connecting to RSocket Proxy...");
       return new PrometheusRSocketClient(
           registryAndScrape,
@@ -384,6 +396,7 @@ public PrometheusRSocketClient connectBlockingly(Duration timeout) {
           registryAndScrape,
           clientTransport,
           retry,
+          timeout,
           () -> {
             LOGGER.info("Connected to RSocket Proxy!");
             onKeyReceived.run();

client/src/test/java/io/micrometer/prometheus/rsocket/PrometheusRSocketClientTests.java

@@ -207,17 +207,17 @@ public Mono<Payload> requestResponse(Payload payload) {
           keyReceived.set(true);
         });
 
-    Field timeoutField = PrometheusRSocketClient.class.getDeclaredField("timeout");
-    timeoutField.setAccessible(true);
-    Duration timeout = (Duration)timeoutField.get(null);
-    assertThat(timeout.toSeconds()).isEqualTo(10L);
-
     CompletableFuture<PrometheusRSocketClient> clientFuture = supplyAsync(clientBuilder::connectBlockingly, newSingleThreadExecutor());
     runAsync(keyReceivedLatch::countDown, newDelayedExecutor());
     assertThat(keyReceived).as("Public key should not be received (not connected)").isFalse();
     PrometheusRSocketClient client = clientFuture.get();
     assertThat(keyReceived).as("Public key should be received(connected)").isTrue();
 
+    Field timeoutField = PrometheusRSocketClient.class.getDeclaredField("timeout");
+    timeoutField.setAccessible(true);
+    Duration timeout = (Duration)timeoutField.get(client);
+    assertThat(timeout.toSeconds()).isEqualTo(10L);
+
     CompletableFuture<Void> closeFuture = runAsync(client::pushAndCloseBlockingly, newSingleThreadExecutor());
     runAsync(pushLatch::countDown, newDelayedExecutor());
     assertThat(pushed).as("Data should not be pushed").isFalse();