Merge pull request #7 from synadia-io/publish-1

Start Publisher 0.1.0

js-publish-extensions/README.md

@@ -1,8 +1,12 @@
 ![Synadia](src/main/javadoc/images/synadia-logo.png)      ![NATS](src/main/javadoc/images/large-logo.png)
 
-# JNATS Managed Async JetStream Publisher Extension
+# JNATS JetStream Publisher Extensions
 
-Utility to automatically publish async with load management and error handling. 
+The library contains 2 utilities.
+
+### PublishRetrier
+
+This class parallels the standard JetStream publish api with methods that will retry the publish
 
 **Current Release**: 0.0.1   **Current Snapshot**: 0.0.2-SNAPSHOT
 

js-publish-extensions/build.gradle

@@ -13,7 +13,7 @@ plugins {
     id 'signing'
 }
 
-def jarVersion = "0.0.11"
+def jarVersion = "0.1.0"
 group = 'io.synadia'
 
 def isMerge = System.getenv("BUILD_EVENT") == "push"
@@ -32,11 +32,17 @@ repositories {
     maven {
         url "https://oss.sonatype.org/content/repositories/releases/"
     }
+    maven {
+        url "https://oss.sonatype.org/content/repositories/snapshots"
+    }
+    maven {
+        url "https://s01.oss.sonatype.org/content/repositories/snapshots/"
+    }
 }
 
 dependencies {
-    implementation 'io.nats:jnats:2.18.1'
-    implementation 'io.synadia:jnats-retrier:0.0.1'
+    implementation 'io.nats:jnats:2.18.2-SNAPSHOT'
+    implementation 'io.synadia:retrier:0.2.0-SNAPSHOT'
 
     testImplementation 'io.nats:jnats-server-runner:1.2.8'
     testImplementation 'org.junit.jupiter:junit-jupiter:5.9.0'

js-publish-extensions/src/examples/java/io/synadia/examples/AsyncJsPublisherExample.java

@@ -0,0 +1,183 @@
+// Copyright (c) 2024 Synadia Communications Inc. All Rights Reserved.
+// See LICENSE and NOTICE file for details.
+
+package io.synadia.examples;
+
+import io.nats.client.Connection;
+import io.nats.client.JetStreamApiException;
+import io.nats.client.Nats;
+import io.nats.client.Options;
+import io.nats.client.api.StorageType;
+import io.nats.client.api.StreamConfiguration;
+import io.nats.client.impl.ErrorListenerConsoleImpl;
+import io.synadia.jnats.extension.AsyncJsPublishListener;
+import io.synadia.jnats.extension.AsyncJsPublisher;
+import io.synadia.jnats.extension.Flight;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static io.synadia.retrier.RetryConfig.DEFAULT_CONFIG;
+
+public class AsyncJsPublisherExample {
+
+    public static final int COUNT = 100_000;
+    public static final String STREAM = "managed";
+    public static final String SUBJECT = "managed_subject";
+
+    public static final boolean USE_RETRIER = false;  // set this to true to have each publish use retry logic
+    public static final boolean BUILT_IN_START = true; // set this to false in order to demonstrate the custom start
+
+    public static void main(String[] args) {
+        Options options = Options.builder()
+            .connectionListener((connection, events) -> print("Connection Event:" + events.getEvent()))
+            .errorListener(new ErrorListenerConsoleImpl())
+            .build();
+
+        try (Connection nc = Nats.connect(options)) {
+            setupStream(nc);
+
+            AsyncJsPublishListener publishListener = new ExamplePublishListener();
+
+            AsyncJsPublisher.Builder builder =
+                AsyncJsPublisher.builder(nc.jetStream())
+                    .publishListener(publishListener);
+
+            if (USE_RETRIER) {
+                builder.retryConfig(DEFAULT_CONFIG);
+            }
+
+            if (BUILT_IN_START) {
+                // the publisher is AutoCloseable
+                try (AsyncJsPublisher publisher = builder.start()) {
+                    publish(publisher, publishListener);
+                }
+            }
+            else {
+                // custom notification executor
+                ExecutorService notificationExecutorService = Executors.newFixedThreadPool(1);
+                builder.notificationExecutorService(notificationExecutorService);
+
+                AsyncJsPublisher publisher = builder.build();
+
+                // this custom start mimics what the built-in does but
+                // shows how to access the publish / flights runner Runnable(s)
+                Thread publishRunnerThread = new Thread(publisher::publishRunner);
+                publishRunnerThread.start();
+                Thread flightsRunnerThread = new Thread(publisher::flightsRunner);
+                flightsRunnerThread.start();
+
+                // same publish logic as the built-in start
+                publish(publisher, publishListener);
+
+                // if you have a custom start, you probably want some custom closing
+                // again, the example mimics what the built-in does
+                // don't forget to call the publisher close, because it does some stuff
+                publisher.close();
+                notificationExecutorService.shutdown();
+                if (!publisher.getPublishRunnerDoneLatch().await(publisher.getPollTime(), TimeUnit.MILLISECONDS)) {
+                    publishRunnerThread.interrupt();
+                }
+                if (!publisher.getFlightsRunnerDoneLatch().await(publisher.getPollTime(), TimeUnit.MILLISECONDS)) {
+                    flightsRunnerThread.interrupt();
+                }
+            }
+        }
+        catch (Exception e) {
+            //noinspection CallToPrintStackTrace
+            e.printStackTrace();
+        }
+    }
+
+    private static void publish(AsyncJsPublisher publisher, AsyncJsPublishListener publishListener) throws InterruptedException {
+        for (int x = 0; x < COUNT; x++) {
+            publisher.publishAsync(SUBJECT, ("data-" + x).getBytes());
+        }
+
+        while (publisher.preFlightSize() > 0) {
+            System.out.println(publishListener);
+            //noinspection BusyWait
+            Thread.sleep(1000);
+        }
+
+        while (publisher.inFlightSize() > 0) {
+            System.out.println(publishListener);
+            //noinspection BusyWait
+            Thread.sleep(1000);
+        }
+
+        System.out.println(publishListener);
+    }
+
+    private static void setupStream(Connection nc) {
+        try {
+            nc.jetStreamManagement().deleteStream(STREAM);
+        }
+        catch (Exception ignore) {}
+        try {
+            System.out.println("Creating Stream @ " + System.currentTimeMillis());
+            nc.jetStreamManagement().addStream(StreamConfiguration.builder()
+                .name(STREAM)
+                .subjects(SUBJECT)
+                .storageType(StorageType.File)
+                .build());
+        }
+        catch (IOException | JetStreamApiException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    static class ExamplePublishListener implements AsyncJsPublishListener {
+        public AtomicLong published = new AtomicLong();
+        public AtomicLong acked = new AtomicLong();
+        public AtomicLong exceptioned = new AtomicLong();
+        public AtomicLong timedOut = new AtomicLong();
+
+        @Override
+        public String toString() {
+            return "published=" + published +
+                ", acked=" + acked +
+                ", exceptioned=" + exceptioned +
+                ", timed out=" + timedOut;
+        }
+
+        @Override
+        public void published(Flight flight) {
+            published.incrementAndGet();
+        }
+
+        @Override
+        public void acked(Flight flight) {
+            acked.incrementAndGet();
+        }
+
+        @Override
+        public void completedExceptionally(Flight flight) {
+            try {
+                exceptioned.incrementAndGet();
+                flight.publishAckFuture.get();
+            }
+            catch (Exception e) {
+                print("completedExceptionally", new String(flight.getBody()), e.toString());
+            }
+        }
+
+        @Override
+        public void timeout(Flight flight) {
+            try {
+                timedOut.incrementAndGet();
+                flight.publishAckFuture.get();
+            }
+            catch (Exception e) {
+                print("timeout", new String(flight.getBody()), e.toString());
+            }
+        }
+    }
+
+    private static void print(String... strings) {
+        System.out.println(String.join(" | ", strings));
+    }
+}

js-publish-extensions/src/examples/java/io/synadia/examples/RetrierPublishAsyncExample.java → js-publish-extensions/src/examples/java/io/synadia/examples/PublishRetrierAsyncExample.java

@@ -9,17 +9,17 @@
 import io.nats.client.api.PublishAck;
 import io.nats.client.api.StorageType;
 import io.nats.client.api.StreamConfiguration;
-import io.synadia.jnats.extension.Retrier;
-import io.synadia.jnats.extension.RetryConfig;
+import io.synadia.jnats.extension.PublishRetrier;
+import io.synadia.retrier.RetryConfig;
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
-public class RetrierPublishAsyncExample {
+public class PublishRetrierAsyncExample {
 
-    public static String STREAM = "retrierA";
-    public static String SUBJECT = "retrierA-subject";
+    public static String STREAM = "pr-async-stream";
+    public static String SUBJECT = "pr-async-subject";
 
     public static void main(String[] args) {
         try (Connection nc = Nats.connect()) {
@@ -53,7 +53,7 @@ public static void main(String[] args) {
             long now = System.currentTimeMillis();
 
             System.out.println("Publishing @ " + now);
-            CompletableFuture<PublishAck> cfpa = Retrier.publishAsync(config, nc.jetStream(), SUBJECT, null);
+            CompletableFuture<PublishAck> cfpa = PublishRetrier.publishAsync(config, nc.jetStream(), SUBJECT, null);
             PublishAck pa = cfpa.get(30, TimeUnit.SECONDS);
             long done = System.currentTimeMillis();