Block up to 1 second for final fire and forget on close

The Mono returned before from the Bean destroyMethod would not be subscribed to, and therefore nothing would happen on destroying the bean. Now we will wait up to 1 second for the fire and forget final scrape to be sent on close.

Resolves gh-35

Author: shakuzen

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

@@ -150,18 +150,18 @@ public void close() {
     }
   }
 
-  public Mono<Void> pushAndClose() {
+  public void pushAndClose() {
     PublicKey key = latestKey.get();
     if (key != null) {
       try {
-        return sendingSocket
+        sendingSocket
             .fireAndForget(scrapePayload(key))
-            .then(Mono.fromRunnable(this::close));
+            .block(Duration.ofSeconds(1));
       } catch (Exception ignored) {
         // give up, we tried...
       }
     }
-    return Mono.fromRunnable(this::close);
+    close();
   }
 
   private Payload scrapePayload(@Nullable PublicKey publicKey) throws Exception {

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

@@ -130,7 +130,7 @@ public Mono<Void> fireAndForget(Payload payload) {
     assertThat(normalScrapeLatch.await(1, TimeUnit.SECONDS)).isTrue();
 
     // trigger dying scrape
-    client.pushAndClose().block(Duration.ofSeconds(1));
+    client.pushAndClose();
     assertThat(dyingScrapeLatch.await(1, TimeUnit.SECONDS))
         .as("Dying scrape (fire-and-forget) should be successfully called")
         .isTrue();

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

@@ -48,8 +48,7 @@ public static void main(String[] args) throws InterruptedException {
       throw new IllegalStateException("Didn't receive a key within 10 seconds");
     }
 
-    CountDownLatch closeLatch = new CountDownLatch(1);
-    client.pushAndClose().subscribe(s -> closeLatch.countDown());
+    client.pushAndClose();
 
     if (!keyLatch.await(3, TimeUnit.SECONDS)) {
       throw new IllegalStateException("Not able to close within 3 seconds");

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

@@ -45,7 +45,7 @@ public static void main(String[] args) throws InterruptedException {
 
     Thread.sleep(1000);
 
-    client.pushAndClose().block(Duration.ofSeconds(1));
+    client.pushAndClose();
     counts.dispose();
   }
 }