Caches, Flush and Ping

For performance reasons, most if not all, of the client libraries will buffer outgoing data so that bigger chunks can be written to the network at one time. This may be as simple as a byte buffer that stores a few messages before being pushed to the network.

These buffers do not hold messages forever, generally they are designed to hold messages in high throughput scenarios, while still providing good latency in low throughput situations.

It is the libraries job to make sure messages flow in a high performance manner. But there may be times when an application needs to know that a message has “hit the wire.” In this case, applications can use a flush call to tell the library to move data through the system.

Go

  1. nc, err := nats.Connect("demo.nats.io")
  2. if err != nil {
  3. log.Fatal(err)
  4. }
  5. defer nc.Close()
  6. // Just to not collide using the demo server with other users.
  7. subject := nats.NewInbox()
  8. if err := nc.Publish(subject, []byte("All is Well")); err != nil {
  9. log.Fatal(err)
  10. }
  11. // Sends a PING and wait for a PONG from the server, up to the given timeout.
  12. // This gives guarantee that the server has processed the above message.
  13. if err := nc.FlushTimeout(time.Second); err != nil {
  14. log.Fatal(err)
  15. }

Java

  1. Connection nc = Nats.connect("nats://demo.nats.io:4222");
  2. nc.publish("updates", "All is Well".getBytes(StandardCharsets.UTF_8));
  3. nc.flush(Duration.ofSeconds(1)); // Flush the message queue
  4. nc.close();

JavaScript

  1. let nc = NATS.connect({url: "nats://demo.nats.io:4222"});
  2. let start = Date.now();
  3. nc.flush(() => {
  4. t.log('round trip completed in', Date.now() - start, 'ms');
  5. });
  6. nc.publish('foo');
  7. // function in flush is optional
  8. nc.flush();

Python

  1. nc = NATS()
  2. await nc.connect(servers=["nats://demo.nats.io:4222"])
  3. await nc.publish("updates", b'All is Well')
  4. # Sends a PING and wait for a PONG from the server, up to the given timeout.
  5. # This gives guarantee that the server has processed above message.
  6. await nc.flush(timeout=1)

Ruby

  1. require 'nats/client'
  2. require 'fiber'
  3. NATS.start(servers:["nats://127.0.0.1:4222"]) do |nc|
  4. nc.subscribe("updates") do |msg|
  5. puts msg
  6. end
  7. nc.publish("updates", "All is Well")
  8. nc.flush do
  9. # Sends a PING and wait for a PONG from the server, up to the given timeout.
  10. # This gives guarantee that the server has processed above message at this point.
  11. end
  12. end

TypeScript

  1. let nc = await connect({
  2. url: "nats://demo.nats.io:4222"
  3. });
  4. // you can use flush to trigger a function in your
  5. // application once the round-trip to the server finishes
  6. let start = Date.now();
  7. nc.flush(() => {
  8. t.log('round trip completed in', Date.now() - start, 'ms');
  9. });
  10. nc.publish('foo');
  11. // another way, simply wait for the promise to resolve
  12. await nc.flush();
  13. nc.close();

C

  1. natsConnection *conn = NULL;
  2. natsStatus s = NATS_OK;
  3. s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
  4. // Send a request and wait for up to 1 second
  5. if (s == NATS_OK)
  6. s = natsConnection_PublishString(conn, "foo", "All is Well");
  7. // Sends a PING and wait for a PONG from the server, up to the given timeout.
  8. // This gives guarantee that the server has processed the above message.
  9. if (s == NATS_OK)
  10. s = natsConnection_FlushTimeout(conn, 1000);
  11. (...)
  12. // Destroy objects that were created
  13. natsConnection_Destroy(conn);

Flush and Ping/Pong

Many of the client libraries use the PING/PONG interaction built into the NATS protocol to ensure that flush pushed all of the buffered messages to the server. When an application calls flush, most libraries will put a PING on the outgoing queue of messages, and wait for the server to respond with a PONG before saying that the flush was successful.

Even though the client may use PING/PONG for flush, pings sent this way do not count towards max outgoing pings.