import io.nats.client.*;

// Connect to default URL ("nats://localhost:4222")
Connection nc = Nats.connect();

// Simple Publisher
nc.publish("foo", "Hello World".getBytes());

// Simple Async Subscriber
nc.subscribe("foo", m -> {
    System.out.println("Received a message: %s\n", new String(m.getData()));
});

// Simple Sync Subscriber
Subscription sub = nc.subscribeSync("foo");
Message m = sub.nextMsg(timeout);

// Unsubscribing
sub = nc.subscribe("foo");
sub.unsubscribe();

// Requests
msg = nc.request("help", "help me", 10000);

// Replies
nc.subscribe("help", m -> {
    nc.publish(m.getReplyTo(), "I can help!");
});

// Close connection
Connect nc = Nats.connect("nats://localhost:4222");
...
nc.close();

	// Set up and load the keystore
	final KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
	final char[] keyPassPhrase = "password".toCharArray();
	final KeyStore ks = KeyStore.getInstance("JKS");
	ks.load(classLoader.getResourceAsStream("keystore.jks"), keyPassPhrase);
	kmf.init(ks, keyPassPhrase);

	// Set up and load the trust store
	final char[] trustPassPhrase = "password".toCharArray();
	final KeyStore tks = KeyStore.getInstance("JKS");
	tks.load(classLoader.getResourceAsStream("cacerts"), trustPassPhrase);
	final TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
	tmf.init(tks);

	// Get and initialize the SSLContext
	SSLContext c = SSLContext.getInstance(Constants.DEFAULT_SSL_PROTOCOL);
	c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

	// Create NATS options
	Options opts = new Options.Builder()
		.secure()       // Set the secure option, indicating that TLS is required
		.tlsDebug()     // Set TLS debug, which will produce additional console output
		.sslContext(c)  // Set the context for this factory
		.build();

	// Create a new SSL connection
	try (Connection connection = Nats.connect("nats://localhost:1222", opts)) {
		connection.publish("foo", "Hello".getBytes());
		connection.close();
	} catch (Exception e) {
		e.printStackTrace();
	}

// "*" matches any token, at any level of the subject.
nc.subscribe("foo.*.baz", m -> {
    System.out.printf("Msg received on [%s] : %s\n", m.getSubject(), new String(m.getData()));
});

nc.subscribe("foo.bar.*", m -> {
    System.out.printf("Msg received on [%s] : %s\n", m.getSubject(), new String(m.getData()));
});

// ">" matches any length of the tail of a subject, and can only be the last token
// E.g. 'foo.>' will match 'foo.bar', 'foo.bar.baz', 'foo.foo.bar.bax.22'
nc.subscribe("foo.>", m -> {
    System.out.printf("Msg received on [%s] : %s\n", m.getSubject(), new String(m.getData()));
});

// Matches all of the above
nc.publish("foo.bar.baz", "Hello World");

// All subscriptions with the same queue name will form a queue group.
// Each message will be delivered to only one subscriber per queue group,
// using queuing semantics. You can have as many queue groups as you wish.
// Normal subscribers will continue to work as expected.

nc.subscribe("foo", "job_workers", m -> {
    received += 1;
});

// Flush connection to server, returns when all messages have been processed.
nc.flush()
System.out.println("All clear!");

// flush can also be called with a timeout value.
try {
    flushed = nc.flush(1000);
    System.out.println("All clear!");
} catch (TimeoutException e) {
    System.out.println("Flushed timed out!");
}

// Auto-unsubscribe after MAX_WANTED messages received
final static int MAX_WANTED = 10;
...
sub = nc.subscribe("foo");
sub.autoUnsubscribe(MAX_WANTED);

// Multiple connections
nc1 = Nats.connect("nats://host1:4222");
nc2 = Nats.connect("nats://host2:4222");

nc1.subscribe("foo", m -> {
    System.out.printf("Received a message: %s\n", new String(m.getData()));
});

nc2.publish("foo", "Hello World!");

String[] servers = new String[] {
	"nats://localhost:1222",
	"nats://localhost:1223",
	"nats://localhost:1224",
};

// Setup options to include all servers in the cluster
ConnectionFactory cf = new ConnectionFactory();
cf.setServers(servers);

// Optionally set ReconnectWait and MaxReconnect attempts.
// This example means 10 seconds total per backend.
cf.setMaxReconnect(5);
cf.setReconnectWait(2000);

// Optionally disable randomization of the server pool
cf.setNoRandomize(true);

Connection nc = cf.createConnection();

// Setup callbacks to be notified on disconnects and reconnects
nc.setDisconnectedCallback(event -> {
    System.out.printf("Got disconnected from %s!\n", event.getConnection().getConnectedUrl());
});

// See who we are connected to on reconnect.
nc.setReconnectedCallback(event -> {
    System.out.printf("Got reconnected to %s!\n", event.getConnection().getConnectedUrl());
});

// Setup a callback to be notified when the Connection is closed
nc.setClosedCallback( event -> {
    System.out.printf("Connection to %s has been closed.\n", event.getConnection().getConnectedUrl());
});