// Create a connection factory
StreamingConnectionFactory cf = new StreamingConnectionFactory("test-cluster", "bar");

// A StreamingConnection is a logical connection to the NATS streaming
// server.  This API creates an underlying core NATS connection for
// convenience and simplicity.  In most cases one would create a secure
// core NATS connection and pass it in via
// StreamingConnectionFactory.setNatsConnection(Connection nc)
StreamingConnection sc = cf.createConnection();

// This simple synchronous publish API blocks until an acknowledgement
// is returned from the server.  If no exception is thrown, the message
// has been stored in NATS streaming.
sc.publish("foo", "Hello World".getBytes());

// Use a countdown latch to wait for our subscriber to receive the
// message we published above.
final CountDownLatch doneSignal = new CountDownLatch(1);

// Simple Async Subscriber that retrieves all available messages.
Subscription sub = sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", new String(m.getData()));
        doneSignal.countDown();
    }
}, new SubscriptionOptions.Builder().deliverAllAvailable().build());

doneSignal.await();

// Unsubscribe to clean up
sub.unsubscribe();

// Close the logical connection to NATS streaming
sc.close();

// Subscribe starting with most recently published value
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().startWithLastReceived().build());

// Receive all stored values in order
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().deliverAllAvailable().build());

// Receive messages starting at a specific sequence number
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().startAtSequence(22).build());

// Subscribe starting at a specific time
Instant startTime = Instant.now().minus(30, ChronoUnit.MINUTES);
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().startAtTime(startTime).build());

// Subscribe starting a specific amount of time in the past (e.g. 30 seconds ago)
Duration ago = Duration.ofSeconds(90);
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().startAtTimeDelta(ago).build());

StreamingConnection sc = new StreamingConnectionFactory("test-cluster", "client-123").createConnection();

// Subscribe with a durable name
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().durableName("my-durable").build());

// The client receives message sequence 1-40, then disconnects.
sc.close();

// Meanwhile more messages are published to subject "foo"

// Here the client reconnects with same clientID "client-123"
sc = new StreamingConnectionFactory("test-cluster", "client-123").createConnection();

// client re-subscribes to "foo" with same durable name "my-durable"
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());
    }
}, new SubscriptionOptions.Builder().durableName("my-durable").build());

// client receives messages 41-current

// The ack handler will be invoked when a publish acknowledgement is received
AckHandler ackHandler = new AckHandler() {
    public void onAck(String guid, Exception err) {
        if (err != null) {
            System.err.printf("Error publishing msg id %s: %s\n", guid, err.getMessage());
        } else {
            System.out.printf("Received ack for msg id %s\n", guid);
        }
    }
};

// This returns immediately.  The result of the publish can be handled in the ack handler.
String guid = sc.publish("foo", "Hello World".getBytes(), ackHandler);

// Subscribe with manual ack mode, and set AckWait to 60 seconds
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received a message: %s\n", m.getData());

        // You must manually ack when manualAcks() are set.
        try {
            m.ack();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}, new SubscriptionOptions.Builder().setManualAcks(true), setAckWait(Duration.ofSeconds(60)));

StreamingConnectionFactory cf = new StreamingConnectionFactory("test-cluster", "client-123");
cf.setMaxPubAcksInFlight(25);

StreamingConnection sc = cf.createConnection();

AckHandler ah = new AckHandler() {
    public void onAck(String guid, Exception e) {
        // process the ack
    }
};

for (int i = 1; i < 1000; i++) {
    // If the server is unable to keep up with the publisher, the number of oustanding acks will eventually
    // reach the max and this call will block

    String guid = sc.publish("foo", "Hello World".getBytes(), ah);
    // track the guid in application code to resend, log, etc. if an error is identified in the ack handler.

// Subscribe with manual ack mode and a max in-flight limit of 25
sc.subscribe("foo", new MessageHandler() {
    public void onMessage(Message m) {
        System.out.printf("Received message : %s\n", m.getData());

        // You must manually ack when manualAcks() are set.  If acks fail or take too long,
        // message delivery will suspend one the number of unacknowledged messages reaches 25
        // due to the max in flight value.
        try {
            m.ack();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}, new SubscriptionOptions.Builder().manualAcks().maxInFlight(25).build());