Merge pull request #1513 from nats-io/doc-examples

Doc examples

Author: scottf

src/examples/java/io/nats/examples/natsIoDoc/QueueGroupsMixedSubscribers.java

@@ -0,0 +1,60 @@
+package io.nats.examples.natsIoDoc;
+
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+
+public class QueueGroupsMixedSubscribers {
+    public static void main(String[] args) {
+        try (Connection nc = Nats.connect("nats://localhost:4222")) {
+
+            // NATS-DOC-START
+            // Set up 3 instances of the service
+            Dispatcher dService1 = nc.createDispatcher(msg -> {
+                byte[] response = calculateResponse(1, msg);
+                nc.publish(msg.getReplyTo(), response);
+            });
+            dService1.subscribe("api.calculate", "api-workers-queue");
+
+            Dispatcher dService2 = nc.createDispatcher(msg -> {
+                byte[] response = calculateResponse(2, msg);
+                nc.publish(msg.getReplyTo(), response);
+            });
+            dService2.subscribe("api.calculate", "api-workers-queue");
+
+            Dispatcher dService3 = nc.createDispatcher(msg -> {
+                byte[] response = calculateResponse(3, msg);
+                nc.publish(msg.getReplyTo(), response);
+            });
+            dService3.subscribe("api.calculate", "api-workers-queue");
+
+            // Make requests - messages are balanced among the subscribers in the queue
+            for (int x = 0; x < 10; x++) {
+                Message m = nc.request("api.calculate", null, Duration.ofMillis(500));
+                if (m == null) {
+                    System.out.println(x + ") No Response");
+                }
+                else {
+                    System.out.println(x + ") " + new String(m.getData()));
+                }
+            }
+            // NATS-DOC-END
+        }
+        catch (InterruptedException e) {
+            // can be thrown by connect
+            Thread.currentThread().interrupt();
+        }
+        catch (IOException e) {
+            // can be thrown by connect
+        }
+    }
+
+    private static byte[] calculateResponse(int i, Message msg) {
+        return ("Result from service instance " + i).getBytes(StandardCharsets.UTF_8);
+    }
+}

src/examples/java/io/nats/examples/natsIoDoc/QueueGroupsRequestReply.java

@@ -0,0 +1,60 @@
+package io.nats.examples.natsIoDoc;
+
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+public class QueueGroupsRequestReply {
+    public static void main(String[] args) {
+        try (Connection nc = Nats.connect("nats://localhost:4222")) {
+
+            // NATS-DOC-START
+            // Audit logger - receives all messages
+            Dispatcher dAudit = nc.createDispatcher(msg -> {
+                System.out.printf("[AUDIT] %s: %s\n", msg.getSubject(), new String(msg.getData(), StandardCharsets.UTF_8));
+            });
+            dAudit.subscribe("orders.>");
+
+            Dispatcher dMetrics = nc.createDispatcher(msg -> {
+                System.out.printf("[METRICS] %s: %s\n", msg.getSubject(), new String(msg.getData(), StandardCharsets.UTF_8));
+            });
+            dMetrics.subscribe("orders.>");
+
+            Dispatcher dNewOrderWorker1 = nc.createDispatcher(msg -> {
+                System.out.printf("[WORKER 1] %s: %s\n", msg.getSubject(), new String(msg.getData(), StandardCharsets.UTF_8));
+            });
+            dNewOrderWorker1.subscribe("orders.new", "new-orders-queue");
+
+            Dispatcher dNewOrderWorker2 = nc.createDispatcher(msg -> {
+                System.out.printf("[WORKER 2] %s: %s\n", msg.getSubject(), new String(msg.getData(), StandardCharsets.UTF_8));
+            });
+            dNewOrderWorker2.subscribe("orders.new", "new-orders-queue");
+
+            // Publish order
+            nc.publish("orders.new", "Order 123".getBytes(StandardCharsets.UTF_8));
+            nc.publish("orders.new", "Order 124".getBytes(StandardCharsets.UTF_8));
+            // Audit and metrics see it, one worker processes it
+            // NATS-DOC-END
+
+            Thread.sleep(100);
+        }
+        catch (InterruptedException e) {
+            // can be thrown by connect
+            Thread.currentThread().interrupt();
+        }
+        catch (IOException e) {
+            // can be thrown by connect
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private static String processNewOrder(int i, Message msg) {
+        return "Order processed by instance " + i;
+    }
+}

src/examples/java/io/nats/examples/natsIoDoc/RequestReplyBasic.java

@@ -0,0 +1,53 @@
+package io.nats.examples.natsIoDoc;
+
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Date;
+import java.util.concurrent.*;
+
+public class RequestReplyBasic {
+    public static void main(String[] args) {
+        try (Connection nc = Nats.connect("nats://localhost:4222")) {
+
+            // NATS-DOC-START
+            // Set up the time service
+            Dispatcher dTime = nc.createDispatcher(msg -> {
+                nc.publish(msg.getReplyTo(), ("" + System.currentTimeMillis()).getBytes(StandardCharsets.UTF_8));
+            });
+            dTime.subscribe("time");
+
+            // Make a request expecting a future
+            CompletableFuture<Message> responseFuture = nc.request("time", null);
+            try {
+                Message m = responseFuture.get(500, TimeUnit.MILLISECONDS);
+                System.out.println("1) Time is " + new Date(Long.parseLong(new String(m.getData()))));
+            }
+            catch (CancellationException | ExecutionException | TimeoutException e) {
+                System.out.println("1) No Response");
+            }
+
+            // Make a request with a timeout and direct response
+            Message m = nc.request("time", null, Duration.ofMillis(500));
+            if (m == null) {
+                System.out.println("2) No Response");
+            }
+            else {
+                System.out.println("2) Time is " + new Date(Long.parseLong(new String(m.getData()))));
+            }
+            // NATS-DOC-END
+        }
+        catch (InterruptedException e) {
+            // can be thrown by connect
+            Thread.currentThread().interrupt();
+        }
+        catch (IOException e) {
+            // can be thrown by connect
+        }
+    }
+}

src/examples/java/io/nats/examples/natsIoDoc/RequestReplyCalculator.java

@@ -0,0 +1,63 @@
+package io.nats.examples.natsIoDoc;
+
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.*;
+
+public class RequestReplyCalculator {
+    public static void main(String[] args) {
+        try (Connection nc = Nats.connect("nats://localhost:4222")) {
+
+            // NATS-DOC-START
+            // Set up the calculator add service
+            Dispatcher dCalcAdd = nc.createDispatcher(msg -> {
+                // data is in the form "x y"
+                try {
+                    String[] parts = new String(msg.getData()).split(" ");
+                    if (parts.length == 2) {
+                        int x = Integer.parseInt(parts[0]);
+                        int y = Integer.parseInt(parts[1]);
+                        nc.publish(msg.getReplyTo(), ("" + (x + y)).getBytes(StandardCharsets.UTF_8));
+                    }
+                }
+                catch (Exception e) {
+                    // you could make some other reply here
+                }
+            });
+            dCalcAdd.subscribe("calc.add");
+
+            // Make a request expecting a future
+            CompletableFuture<Message> responseFuture = nc.request("calc.add", "5 3".getBytes(StandardCharsets.UTF_8));
+            try {
+                Message m = responseFuture.get(500, TimeUnit.MILLISECONDS);
+                System.out.printf("5 + 3 = %s\n", new String(m.getData()));
+            }
+            catch (CancellationException | ExecutionException | TimeoutException e) {
+                System.out.println("1) No Response");
+            }
+
+            // Make a request with a timeout and direct response
+            Message m = nc.request("calc.add", "10 7".getBytes(StandardCharsets.UTF_8), Duration.ofMillis(500));
+            if (m == null) {
+                System.out.println("2) No Response");
+            }
+            else {
+                System.out.printf("10 + 7 = %s\n", new String(m.getData()));
+            }
+            // NATS-DOC-END
+        }
+        catch (InterruptedException e) {
+            // can be thrown by connect
+            Thread.currentThread().interrupt();
+        }
+        catch (IOException e) {
+            // can be thrown by connect
+        }
+    }
+}

src/examples/java/io/nats/examples/natsIoDoc/RequestReplyHeaders.java

@@ -0,0 +1,65 @@
+package io.nats.examples.natsIoDoc;
+
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+import io.nats.client.impl.Headers;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.*;
+
+public class RequestReplyHeaders {
+    public static void main(String[] args) {
+        try (Connection nc = Nats.connect("nats://localhost:4222")) {
+
+            // NATS-DOC-START
+            // Set up the header echo service
+            Dispatcher dService = nc.createDispatcher(msg -> {
+                Headers hIncoming = msg.getHeaders();
+                Headers hResponse = new Headers();
+                for (String keys : hIncoming.keySet()) {
+                    hResponse.put(keys, hIncoming.get(keys));
+                }
+                nc.publish(msg.getReplyTo(), hResponse, null);
+            });
+            dService.subscribe("header.echo");
+
+            // Make a request expecting a future
+            Headers headers1 = new Headers();
+            headers1.put("X-Request-ID", "1");
+            headers1.put("X-Priority", "high");
+            CompletableFuture<Message> responseFuture = nc.request("header.echo", headers1, null);
+            try {
+                Message m = responseFuture.get(500, TimeUnit.MILLISECONDS);
+                Headers hIncoming = m.getHeaders();
+                System.out.println("Response Headers: " + hIncoming);
+            }
+            catch (CancellationException | ExecutionException | TimeoutException e) {
+                System.out.println("1) No Response");
+            }
+
+            // Make a request with a timeout and direct response
+            Headers headers2 = new Headers();
+            headers2.put("X-Request-ID", "2");
+            headers2.put("X-Priority", "med");
+            Message m = nc.request("header.echo", headers2, null, Duration.ofMillis(500));
+            if (m == null) {
+                System.out.println("2) No Response");
+            }
+            else {
+                Headers hIncoming = m.getHeaders();
+                System.out.println("Response Headers: " + hIncoming);
+            }
+            // NATS-DOC-END
+        }
+        catch (InterruptedException e) {
+            // can be thrown by connect
+            Thread.currentThread().interrupt();
+        }
+        catch (IOException e) {
+            // can be thrown by connect
+        }
+    }
+}

src/examples/java/io/nats/examples/natsIoDoc/RequestReplyMultipleResponders.java

@@ -0,0 +1,63 @@
+package io.nats.examples.natsIoDoc;
+
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.*;
+
+public class RequestReplyMultipleResponders {
+    public static void main(String[] args) {
+        try (Connection nc = Nats.connect("nats://localhost:4222")) {
+
+            // NATS-DOC-START
+            // Set up 2 instances of the service
+            Dispatcher dService1 = nc.createDispatcher(msg -> {
+                byte[] response = calculateResponse(1, msg);
+                nc.publish(msg.getReplyTo(), response);
+            });
+            dService1.subscribe("service");
+
+            Dispatcher dService2 = nc.createDispatcher(msg -> {
+                byte[] response = calculateResponse(2, msg);
+                nc.publish(msg.getReplyTo(), response);
+            });
+            dService2.subscribe("service");
+
+            // Make a request expecting a future
+            CompletableFuture<Message> responseFuture = nc.request("service", null);
+            try {
+                Message m = responseFuture.get(500, TimeUnit.MILLISECONDS);
+                System.out.println("1) " + new String(m.getData()));
+            }
+            catch (CancellationException | ExecutionException | TimeoutException e) {
+                System.out.println("1) No Response");
+            }
+
+            // Make a request with a timeout and direct response
+            Message m = nc.request("service", null, Duration.ofMillis(500));
+            if (m == null) {
+                System.out.println("2) No Response");
+            }
+            else {
+                System.out.println("2) " + new String(m.getData()));
+            }
+            // NATS-DOC-END
+        }
+        catch (InterruptedException e) {
+            // can be thrown by connect
+            Thread.currentThread().interrupt();
+        }
+        catch (IOException e) {
+            // can be thrown by connect
+        }
+    }
+
+    private static byte[] calculateResponse(int i, Message msg) {
+        return ("Result from service instance " + i).getBytes(StandardCharsets.UTF_8);
+    }
+}