Listen for Connection Events

    The actual API for these listeners is language dependent, but the following examples show a few of the more common use cases. See the API documentation for the client library you are using for more specific instructions.

    Connection events may include the connection being closed, disconnected or reconnected. Reconnecting involves a disconnect and connect, but depending on the library implementation may also include multiple disconnects as the client tries to find a server, or the server is rebooted.

    {% tabs %} {% tab title=”Go” %}

    {% endtab %}

    {% tab title=”Java” %}

    2.     public void connectionEvent(Connection natsConnection, Events event) {
    3.         System.out.println("Connection event - "+event);
    4.     }
    5. }
    7. public class SetConnectionListener {
    8.     public static void main(String[] args) {
    10.         try {
    11.             Options options = new Options.Builder().
    12.                                         server("nats://demo.nats.io:4222").
    13.                                         connectionListener(new MyConnectionListener()). // Set the listener
    14.                                         build();
    15.             Connection nc = Nats.connect(options);
    17.             // Do something with the connection
    19.             nc.close();
    20.         } catch (Exception e) {
    21.             e.printStackTrace();
    22.         }
    23.     }
    24. }

    {% endtab %}

    {% tab title=”JavaScript” %}

    1. const nc = await connect({ servers: ["demo.nats.io"] });
    2. nc.closed().then(() => {
    3.   t.log("the connection closed!");
    4. });
    6. (async () => {
    7.     for await (const s of nc.status()) {
    8.       switch (s.type) {
    9.         case Status.Disconnect:
    10.           t.log(`client disconnected - ${s.data}`);
    11.           break;
    12.         case Status.LDM:
    13.           t.log("client has been requested to reconnect");
    14.           break;
    15.         case Status.Update:
    16.           t.log(`client received a cluster update - ${s.data}`);
    17.           break;
    18.         case Status.Reconnect:
    19.           t.log(`client reconnected - ${s.data}`);
    20.           break;
    21.         case Status.Error:
    22.           t.log("client got a permissions error");
    23.           break;
    24.         case DebugEvents.Reconnecting:
    25.           t.log("client is attempting to reconnect");
    26.           break;
    27.         case DebugEvents.StaleConnection:
    28.           t.log("client has a stale connection");
    29.           break;
    30.         default:
    31.           t.log(`got an unknown status ${s.type}`);
    32.       }
    33.     }
    34. })().then();

    {% endtab %}

    {% tab title=”Python” %}

    1. # Asyncio NATS client can be defined a number of event callbacks
    2. async def disconnected_cb():
    3.     print("Got disconnected!")
    5. async def reconnected_cb():
    6.     # See who we are connected to on reconnect.
    7.     print("Got reconnected to {url}".format(url=nc.connected_url.netloc))
    9. async def error_cb(e):
    10.     print("There was an error: {}".format(e))
    12. async def closed_cb():
    13.     print("Connection is closed")
    15. # Setup callbacks to be notified on disconnects and reconnects
    16. options["disconnected_cb"] = disconnected_cb
    17. options["reconnected_cb"] = reconnected_cb
    19. # Setup callbacks to be notified when there is an error
    20. # or connection is closed.
    21. options["error_cb"] = error_cb
    22. options["closed_cb"] = closed_cb
    24. await nc.connect(**options)

    {% endtab %}

    {% tab title=”Ruby” %}

    1. # There is not a single listener for connection events in the Ruby NATS Client.
    2. # Instead, you can set individual event handlers using:
    4. NATS.on_disconnect do
    5. end
    7. NATS.on_reconnect do
    8. end
    10. NATS.on_close do
    11. end
    13. NATS.on_error do
    14. end

    {% endtab %}

    1. static void
    2. disconnectedCB(natsConnection *conn, void *closure)
    3. {
    4.     // Do something
    5.     printf("Connection disconnected\n");
    6. }
    8. static void
    9. reconnectedCB(natsConnection *conn, void *closure)
    10. {
    11.     printf("Connection reconnected\n");
    14. static void
    15. closedCB(natsConnection *conn, void *closure)
    16. {
    17.     // Do something
    18.     printf("Connection closed\n");
    19. }
    21. (...)
    23. natsConnection      *conn      = NULL;
    24. natsOptions         *opts      = NULL;
    25. natsStatus          s          = NATS_OK;
    27. s = natsOptions_Create(&opts);
    28. if (s == NATS_OK)
    29.     s = natsOptions_SetDisconnectedCB(opts, disconnectedCB, NULL);
    30. if (s == NATS_OK)
    31.     s = natsOptions_SetReconnectedCB(opts, reconnectedCB, NULL);
    32. if (s == NATS_OK)
    33.     s = natsOptions_SetClosedCB(opts, closedCB, NULL);
    34. if (s == NATS_OK)
    35.     s = natsConnection_Connect(&conn, opts);
    37. (...)
    39. // Destroy objects that were created
    40. natsConnection_Destroy(conn);
    41. natsOptions_Destroy(opts);

    {% endtab %} {% endtabs %}

    When working with a cluster, servers may be added or changed. Some of the clients allow you to listen for this notification:

    {% tabs %} {% tab title=”Go” %}

    {% endtab %}

    {% tab title=”Java” %}

    1. class ServersAddedListener implements ConnectionListener {
    2.     public void connectionEvent(Connection nc, Events event) {
    3.         if (event == Events.DISCOVERED_SERVERS) {
    4.             for (String server : nc.getServers()) {
    5.                 System.out.println("Known server: "+server);
    6.             }
    7.         }
    8.     }
    9. }
    11. public class ListenForNewServers {
    12.     public static void main(String[] args) {
    14.         try {
    15.             Options options = new Options.Builder().
    16.                                         server("nats://demo.nats.io:4222").
    17.                                         connectionListener(new ServersAddedListener()). // Set the listener
    18.                                         build();
    19.             Connection nc = Nats.connect(options);
    21.             // Do something with the connection
    23.             nc.close();
    24.         } catch (Exception e) {
    25.             e.printStackTrace();
    26.         }
    27.     }
    28. }

    {% endtab %}

    {% tab title=”JavaScript” %}

    1. const nc = await connect({ servers: ["demo.nats.io:4222"] });
    2. (async () => {
    3.   for await (const s of nc.status()) {
    4.     switch (s.type) {
    5.       case Status.Update:
    6.         t.log(`servers added - ${s.data.added}`);
    7.         t.log(`servers deleted - ${s.data.deleted}`);
    8.         break;
    9.       default:
    10.     }
    11.   }
    12. })().then();

    {% endtab %}

    {% tab title=”Python” %}

    1. # Asyncio NATS client does not support discovered servers handler right now

    {% endtab %}

    {% tab title=”Ruby” %}

    1. # The Ruby NATS client does not support discovered servers handler right now

    {% endtab %}

    1. static void
    2. discoveredServersCB(natsConnection *conn, void *closure)
    3. {
    4.     natsStatus  s         = NATS_OK;
    5.     char        **servers = NULL;
    6.     int         count     = 0;
    8.     s = natsConnection_GetDiscoveredServers(conn, &servers, &count);
    9.     if (s == NATS_OK)
    10.     {
    11.         int i;
    13.         // Do something...
    14.         for (i=0; i<count; i++)
    15.             printf("Discovered server: %s\n", servers[i]);
    17.         // Free allocated memory
    18.         for (i=0; i<count; i++)
    19.             free(servers[i]);
    20.         free(servers);
    21.     }
    22. }
    24. (...)
    26. natsConnection      *conn      = NULL;
    27. natsOptions         *opts      = NULL;
    28. natsStatus          s          = NATS_OK;
    30. s = natsOptions_Create(&opts);
    31. if (s == NATS_OK)
    32.     s = natsOptions_SetDiscoveredServersCB(opts, discoveredServersCB, NULL);
    33. if (s == NATS_OK)
    34.     s = natsConnection_Connect(&conn, opts);
    38. // Destroy objects that were created
    39. natsConnection_Destroy(conn);
    40. natsOptions_Destroy(opts);

    {% endtab %} {% endtabs %}

    Listen for Errors

    The client library may separate server-to-client errors from events. Many server events are not handled by application code and result in the connection being closed. Listening for the errors can be very useful for debugging problems.

    {% tabs %} {% tab title=”Go” %}

    {% endtab %}

    {% tab title=”Java” %}

    1. class MyErrorListener implements ErrorListener {
    2.     public void errorOccurred(Connection conn, String error)
    3.     {
    4.         System.out.println("The server notificed the client with: "+error);
    5.     }
    7.     public void exceptionOccurred(Connection conn, Exception exp) {
    8.         System.out.println("The connection handled an exception: "+exp.getLocalizedMessage());
    9.     }
    11.     public void slowConsumerDetected(Connection conn, Consumer consumer) {
    12.         System.out.println("A slow consumer was detected.");
    13.     }
    14. }
    16. public class SetErrorListener {
    17.     public static void main(String[] args) {
    19.         try {
    20.             Options options = new Options.Builder().
    21.                                         server("nats://demo.nats.io:4222").
    22.                                         errorListener(new MyErrorListener()). // Set the listener
    23.                                         build();
    24.             Connection nc = Nats.connect(options);
    26.             // Do something with the connection
    28.             nc.close();
    29.         } catch (Exception e) {
    30.             e.printStackTrace();
    31.         }
    32.     }
    33. }

    {% endtab %}

    {% tab title=”JavaScript” %}

    1. const nc = await connect({ servers: ["demo.nats.io"] });
    3. // if the client gets closed with an error you can trap that
    4. // condition in the closed handler like this:
    5. nc.closed().then((err) => {
    6.   if (err) {
    7.     t.log(`the connection closed with an error ${err.message}`);
    8.   } else {
    9.     t.log(`the connection closed.`);
    10.   }
    11. });
    13. // if you have a status listener, it will too get notified
    14. (async () => {
    15.   for await (const s of nc.status()) {
    16.     switch (s.type) {
    17.       case Status.Error:
    18.         // typically if you get this the nats connection will close
    19.         t.log("client got an async error from the server");
    20.         break;
    21.       default:
    22.         t.log(`got an unknown status ${s.type}`);
    23.     }
    24.   }
    25. })().then();

    {% endtab %}

    {% tab title=”Python” %}

    1. nc = NATS()
    3. async def error_cb(e):
    4.    print("Error: ", e)
    6. await nc.connect(
    7.    servers=["nats://demo.nats.io:4222"],
    8.    reconnect_time_wait=10,
    9.    error_cb=error_cb,
    10.    )
    12. # Do something with the connection.

    {% endtab %}

    {% tab title=”Ruby” %}

    1. require 'nats/client'
    3. NATS.start(servers:["nats://demo.nats.io:4222"]) do |nc|
    4.    nc.on_error do |e|
    5.     puts "Error: #{e}"
    6.   end
    8.   nc.close
    9. end

    {% endtab %}

    1. static void
    2. errorCB(natsConnection *conn, natsSubscription *sub, natsStatus s, void *closure)
    3. {
    4.     // Do something
    5.     printf("Error: %d - %s\n", s, natsStatus_GetText(s));
    6. }
    8. (...)
    10. natsConnection      *conn      = NULL;
    11. natsOptions         *opts      = NULL;
    12. natsStatus          s          = NATS_OK;
    14. s = natsOptions_Create(&opts);
    15. if (s == NATS_OK)
    16.     s = natsOptions_SetErrorHandler(opts, errorCB, NULL);
    17. if (s == NATS_OK)
    18.     s = natsConnection_Connect(&conn, opts);
    20. (...)
    22. // Destroy objects that were created
    23. natsConnection_Destroy(conn);
    24. natsOptions_Destroy(opts);

    {% endtab %} {% endtabs %}