Upgrade JNats and Server Runner

build.gradle

@@ -10,7 +10,7 @@ plugins {
 
 }
 
-def jarVersion = "2.0.4"
+def jarVersion = "2.1.0"
 group = 'io.nats'
 
 def isMerge = System.getenv("BUILD_EVENT") == "push"
@@ -33,14 +33,14 @@ repositories {
 
 dependencies {
 
-    implementation 'io.nats:jnats:2.17.2'
+    implementation 'io.nats:jnats:2.20.5'
     implementation("com.fasterxml.jackson.core:jackson-core:2.14.2")
     implementation("io.netty:netty-handler:4.1.97.Final")
     implementation(platform("io.vertx:vertx-stack-depchain:4.5.1"))
     implementation("io.vertx:vertx-core")
     implementation("io.netty:netty-resolver-dns-native-macos:4.1.80.Final:osx-x86_64")
     testImplementation("io.vertx:vertx-junit5")
-    testImplementation 'io.nats:jnats-server-runner:1.2.5'
+    testImplementation 'io.nats:jnats-server-runner:2.0.0'
 }
 
 test {

src/main/java/io/nats/client/impl/VertxDispatcher.java

@@ -8,7 +8,7 @@ public class VertxDispatcher extends NatsDispatcher {
 
     VertxDispatcher(NatsConnection conn, MessageHandler handler, ContextInternal context) {
         super(conn, handler);
-        vertxIncoming = new VertxDispatcherMessageQueue(this, context);
+        vertxIncoming = new VertxDispatcherMessageQueue(this, context, conn);
     }
 
     @Override

src/main/java/io/nats/client/impl/VertxDispatcherMessageQueue.java

@@ -10,8 +10,8 @@ public class VertxDispatcherMessageQueue extends MessageQueue {
     private final VertxDispatcher dispatcher;
     private final ContextInternal context;
 
-    VertxDispatcherMessageQueue(VertxDispatcher dispatcher, ContextInternal context) {
-        super(true);
+    VertxDispatcherMessageQueue(VertxDispatcher dispatcher, ContextInternal context, NatsConnection conn) {
+        super(true, conn.getOptions().getRequestCleanupInterval());
         this.dispatcher = dispatcher;
         this.context = context;
     }
@@ -65,11 +65,6 @@ void poisonTheQueue() {
         throw new IllegalStateException("poisonTheQueue not used.");
     }
 
-    @Override
-    boolean offer(NatsMessage msg) {
-        throw new IllegalStateException("offer not used.");
-    }
-
     @Override
     NatsMessage poll(Duration timeout) throws InterruptedException {
         return super.poll(timeout);

src/test/java/io/nats/vertx/NatsClientTest.java

@@ -1,5 +1,6 @@
 package io.nats.vertx;
 
+import io.nats.NatsServerRunner;
 import io.nats.client.*;
 import io.nats.client.api.StorageType;
 import io.nats.client.api.StreamConfiguration;
@@ -11,7 +12,6 @@
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
-import nats.io.NatsServerRunner;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

src/test/java/io/nats/vertx/NatsSimplePerfTest.java

@@ -1,12 +1,12 @@
 package io.nats.vertx;
 
+import io.nats.NatsServerRunner;
 import io.nats.client.*;
 import io.nats.client.api.StorageType;
 import io.nats.client.api.StreamConfiguration;
 import io.nats.client.api.StreamInfo;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
-import nats.io.NatsServerRunner;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -40,7 +40,7 @@ public void after() throws Exception {
 
     @BeforeEach
     public void setup() throws Exception {
-        natsServerRunner = new NatsServerRunner(0, false, true);
+        natsServerRunner = NatsServerRunner.builder().jetstream().build();
         Thread.sleep(1);
 
 

src/test/java/io/nats/vertx/NatsStreamTest.java

@@ -1,5 +1,6 @@
 package io.nats.vertx;
 
+import io.nats.NatsServerRunner;
 import io.nats.client.*;
 import io.nats.client.api.StorageType;
 import io.nats.client.api.StreamConfiguration;
@@ -8,7 +9,6 @@
 import io.nats.client.impl.NatsMessage;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
-import nats.io.NatsServerRunner;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -34,11 +34,18 @@ public class NatsStreamTest {
 
     @AfterEach
     public void after() throws Exception {
-        if (nc != null)
-            nc.close();
-
-        if (natsServerRunner != null)
-            natsServerRunner.close();
+        try {
+            if (nc != null) {
+                nc.close();
+            }
+        }
+        catch (Exception ignore) {}
+        try {
+            if (natsServerRunner != null) {
+                natsServerRunner.close();
+            }
+        }
+        catch (Exception ignore) {}
     }
 
     int port;

src/test/java/io/nats/vertx/TestUtils.java

@@ -1,11 +1,11 @@
 package io.nats.vertx;
 
+import io.nats.NatsServerRunner;
 import io.nats.client.*;
 import io.nats.client.support.Status;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
-import nats.io.NatsServerRunner;
 
 import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
@@ -100,7 +100,7 @@ public static NatsClient natsClient(int port, Vertx vertx, Handler<Throwable> ex
 
     public static NatsServerRunner startServer() throws Exception {
         NatsServerRunner.setDefaultOutputLevel(Level.WARNING);
-        NatsServerRunner natsServerRunner = new NatsServerRunner(0, false, true);
+        NatsServerRunner natsServerRunner = NatsServerRunner.builder().jetstream().build();
 
         int port = natsServerRunner.getPort();
         for (int i = 0; i < 100; i++) {

src/test/java/io/nats/vertx/impl/SubscriptionReadStreamImplTest.java

@@ -1,6 +1,7 @@
 package io.nats.vertx.impl;
 
 
+import io.nats.NatsServerRunner;
 import io.nats.client.*;
 import io.nats.client.api.StorageType;
 import io.nats.client.api.StreamConfiguration;
@@ -10,7 +11,6 @@
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
-import nats.io.NatsServerRunner;
 import org.junit.jupiter.api.*;
 import org.junit.jupiter.api.extension.ExtendWith;