Merge pull request #1328 from ajax-surovskyi-y/auth_violation_reproduce_program

Accepting - no production code, just testing code. Can always be removed later.

Author: scottf

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

@@ -55,7 +55,7 @@ class NatsConnection implements Connection {
     private boolean disconnecting; // you can only disconnect in one thread
     private boolean closing; // respect a close call regardless
     private Exception exceptionDuringConnectChange; // exception occurred in another thread while dis/connecting
-    private final ReentrantLock closeSocketLock;
+    final ReentrantLock closeSocketLock;
 
     private Status status;
     private final ReentrantLock statusLock;

src/test/java/io/nats/client/impl/AuthViolationDuringReconnect.java

@@ -0,0 +1,150 @@
+package io.nats.client.impl;
+
+import io.nats.client.*;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+/* Program to reproduce #1320 */
+public class AuthViolationDuringReconnect {
+    private static final ConcurrentHashMap.KeySetView<String, Boolean> subscriptions = ConcurrentHashMap.newKeySet();
+    private static final ScheduledExecutorService serverRestarter = Executors.newSingleThreadScheduledExecutor();
+    private static final ExecutorService unsubThreadpool = Executors.newFixedThreadPool(2);
+    private static final AtomicReference<NatsTestServer> ts = new AtomicReference<>();
+    private static final ErrorListener AUTHORIZATION_VIOLATION_LISTENER = new ErrorListener() {
+        @Override
+        public void errorOccurred(Connection conn, String error) {
+            if (error.contains("Authorization Violation")) {
+                System.out.println("Authorization Violation, Stopping server");
+                try {
+                    Thread.sleep(1000);
+                    ts.get().shutdown();
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+                System.exit(-1);
+            }
+        }
+    };
+    private static volatile CountDownLatch latch;
+
+    public static void main(String[] args) throws IOException, InterruptedException {
+        int port = NatsTestServer.nextPort();
+        ts.set(new NatsTestServer(new String[]{"--auth", "1234", "-m", "8222"}, port, false));
+
+        ReconnectedHandler reconnectedHandler = new ReconnectedHandler();
+        NatsConnection nc = new MockPausingNatsConnection(buildOptions(port, reconnectedHandler));
+        nc.connect(true);
+        Dispatcher d = nc.createDispatcher();
+
+        reconnectedHandler.setConsumer((ignored) -> subscribe(d));
+        subscribe(d);
+
+        serverRestarter.scheduleWithFixedDelay(() -> restartServer(ts, port), 2000, 3000, TimeUnit.MILLISECONDS);
+
+        new Thread(waitCloseSocket(nc)).start();
+    }
+
+    private static Runnable waitCloseSocket(NatsConnection nc) {
+        return () -> {
+            try {
+                Thread.sleep(200);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            while (true) {
+                if (nc.closeSocketLock.isLocked()) {
+                    System.out.printf("Unsubscribing all subscriptions due to disconnection %d \n", subscriptions.size());
+                    latch.countDown();
+                    while (nc.closeSocketLock.isLocked()) {
+                    }
+                }
+            }
+        };
+    }
+
+    private static void restartServer(AtomicReference<NatsTestServer> ts, int port) {
+        try {
+            System.out.println("Restarting server");
+            ts.get().shutdown();
+            ts.set(new NatsTestServer(new String[]{"--auth", "1234", "-m", "8222"}, port, false));
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static void subscribe(Dispatcher d) {
+        latch = new CountDownLatch(1);
+        for (int i = 0; i < 1_000; i++) {
+            String subject = "test_" + i;
+            subscriptions.add(subject);
+            d.subscribe(subject);
+            unsubThreadpool.execute(() -> {
+                try {
+                    latch.await();
+                    d.unsubscribe(subject);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            });
+        }
+    }
+
+    private static Options buildOptions(int port, ReconnectedHandler reconnectedHandler) {
+        Options.Builder natsOptions = new Options.Builder()
+                .servers(new String[]{"nats://incorrect:1111", "nats://localhost:" + port})
+                .noRandomize()
+                .token(new char[]{'1', '2', '3', '4'})
+                .maxReconnects(-1)
+                .reconnectWait(Duration.ofMillis(2000))
+                .connectionTimeout(Duration.ofMillis(500))
+                .connectionListener(reconnectedHandler)
+                .errorListener(AUTHORIZATION_VIOLATION_LISTENER);
+
+        return natsOptions.build();
+    }
+
+    private static class ReconnectedHandler implements ConnectionListener {
+
+        private java.util.function.Consumer<Void> consumer;
+
+        public void setConsumer(java.util.function.Consumer<Void> consumer) {
+            this.consumer = consumer;
+        }
+
+        @Override
+        public void connectionEvent(Connection conn, Events type) {
+            if (type == Events.RECONNECTED) {
+                consumer.accept(null);
+            }
+        }
+    }
+
+    static class MockPausingNatsConnection extends NatsConnection {
+        MockPausingNatsConnection(Options options) {
+            super(options);
+        }
+
+        @Override
+        void closeSocketImpl(boolean forceClose) {
+            try {
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            super.closeSocketImpl(forceClose);
+        }
+
+        @Override
+        void sendUnsub(NatsSubscription sub, int after) {
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            super.sendUnsub(sub, after);
+        }
+    }
+}