fix: timeout use class variable instead of static

Author: klopfdreh

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

@@ -65,11 +65,36 @@ public class PrometheusRSocketClient {
   private volatile boolean requestedDisconnect = false;
   private RSocket sendingSocket;
 
-  private static Duration timeout = Duration.ofSeconds(5);
+  private Duration timeout = Duration.ofSeconds(5);
 
+  /**
+   * Creates a {@link PrometheusRSocketClient}.
+   *
+   * @param registryAndScrape the registry and scrape meter
+   * @param transport         the client transport
+   * @param retry             the retry configuration
+   * @param onKeyReceived     the callback if a key has been received
+   */
+  private PrometheusRSocketClient(MeterRegistryAndScrape<?> registryAndScrape,
+                                  ClientTransport transport,
+                                  Retry retry,
+                                  Runnable onKeyReceived) {
+    this(registryAndScrape, transport, retry, Duration.ofSeconds(5), onKeyReceived);
+  }
+
+  /**
+   * Creates a {@link PrometheusRSocketClient}.
+   *
+   * @param registryAndScrape the registry and scrape meter
+   * @param transport         the client transport
+   * @param retry             the retry configuration
+   * @param timeout           the timeout to connect and push the data
+   * @param onKeyReceived     the callback if a key has been received
+   */
   private PrometheusRSocketClient(MeterRegistryAndScrape<?> registryAndScrape,
                                   ClientTransport transport,
                                   Retry retry,
+                                  Duration timeout,
                                   Runnable onKeyReceived) {
     this.registryAndScrape = registryAndScrape;
 
@@ -118,8 +143,8 @@ public Mono<Void> fireAndForget(Payload payload) {
             .increment())
         .doOnNext(connection -> this.connection = connection)
         .flatMap(socket -> socket.onClose()
-          .map(v -> 1) // https://github.com/rsocket/rsocket-java/issues/819
-          .onErrorReturn(1))
+            .map(v -> 1) // https://github.com/rsocket/rsocket-java/issues/819
+            .onErrorReturn(1))
         .repeat(() -> !requestedDisconnect)
         .subscribe();
   }
@@ -170,6 +195,7 @@ public void pushAndCloseBlockingly() {
 
   /**
    * Pushes the data in a blocking way and closes the connection.
+   *
    * @param timeout the amount of time to wait for the data to be sent
    */
   public void pushAndCloseBlockingly(Duration timeout) {
@@ -188,8 +214,7 @@ public void pushAndCloseBlockingly(Duration timeout) {
             .doOnCancel(() -> LOGGER.warn("Pushing data to RSocket Proxy before closing the connection was cancelled!"))
             .doFinally(signalType -> latch.countDown())
             .subscribe();
-      }
-      catch (Exception exception) {
+      } catch (Exception exception) {
         latch.countDown();
         LOGGER.warn("Sending the payload failed!", exception);
       }
@@ -198,8 +223,7 @@ public void pushAndCloseBlockingly(Duration timeout) {
         if (!latch.await(timeout.toMillis(), MILLISECONDS)) {
           LOGGER.warn("Sending the payload timed out!");
         }
-      }
-      catch (InterruptedException exception) {
+      } catch (InterruptedException exception) {
         LOGGER.warn("Waiting for sending the payload was interrupted!", exception);
       }
     }
@@ -220,8 +244,7 @@ public void pushAndClose() {
             .doOnError(throwable -> LOGGER.warn("Pushing data to RSocket Proxy before closing the connection failed!", throwable))
             .doOnCancel(() -> LOGGER.warn("Pushing data to RSocket Proxy before closing the connection was cancelled!"))
             .subscribe();
-      }
-      catch (Exception exception) {
+      } catch (Exception exception) {
         LOGGER.warn("Sending the payload failed!", exception);
       }
     }
@@ -278,8 +301,10 @@ public static class Builder {
     private Retry retry = Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(10))
         .maxBackoff(Duration.ofMinutes(10));
 
+    private Duration timeout;
 
-    private Runnable onKeyReceived = () -> { };
+    private Runnable onKeyReceived = () -> {
+    };
 
     <M extends MeterRegistry> Builder(M registry, Supplier<String> scrape, ClientTransport clientTransport) {
       this.registryAndScrape = new MeterRegistryAndScrape<>(registry, scrape);
@@ -290,7 +315,6 @@ <M extends MeterRegistry> Builder(M registry, Supplier<String> scrape, ClientTra
      * Configures the retry for {@link PrometheusRSocketClient}.
      *
      * @param retry the retry configuration
-     *
      * @return the {@link Builder}
      */
     public Builder retry(Retry retry) {
@@ -305,7 +329,7 @@ public Builder retry(Retry retry) {
      * @return the {@link Builder}
      */
     public Builder timeout(Long timeout) {
-      PrometheusRSocketClient.timeout = Duration.ofSeconds(timeout);
+      this.timeout = Duration.ofSeconds(timeout);
       return this;
     }
 
@@ -331,6 +355,7 @@ public PrometheusRSocketClient connect() {
           registryAndScrape,
           clientTransport,
           retry,
+          timeout,
           () -> {
             LOGGER.info("Connected to RSocket Proxy!");
             onKeyReceived.run();
@@ -370,8 +395,7 @@ public PrometheusRSocketClient connectBlockingly(Duration timeout) {
         if (!latch.await(timeout.toMillis(), MILLISECONDS)) {
           LOGGER.warn("Creating the connection and receiving the key timed out!");
         }
-      }
-      catch (InterruptedException exception) {
+      } catch (InterruptedException exception) {
         LOGGER.warn("Waiting for receiving the key was interrupted!", exception);
       }