update domo

src/broker/nats-broker.cpp

@@ -23,6 +23,9 @@
  *
  */
 
+// This is an independent project of an individual developer. Dear PVS-Studio, please check it.
+// PVS-Studio Static Code Analyzer for C, C++, C#, and Java: http://www.viva64.com
+
 /*------------------------------- INCLUDES ----------------------------------*/
 
 #include <adapters/libevent.h>
@@ -31,6 +34,20 @@
 
 #include <nats.h>
 
+#define kNatsServerURL "nats.nadal-fr.com"
+
+
+static void onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
+{
+    printf("Received msg: %s - %.*s\n",
+           natsMsg_GetSubject(msg),
+           natsMsg_GetDataLength(msg),
+           natsMsg_GetData(msg));
+
+    // Need to destroy the message!
+    natsMsg_Destroy(msg);
+}
+
 /*!  ----------------------------------------------------------------------------
  *  @fn NatsBroker
  *
@@ -54,7 +71,7 @@ NatsBroker::~NatsBroker(void)
  *
  *  @brief Setup the Broker.
  */
-int NatsBroker::setup(void)
+int NatsBroker::setup(struct event_base *an_event_loop)
 {
     natsStatus the_status;
 
@@ -64,7 +81,70 @@ int NatsBroker::setup(void)
     natsLibevent_Init();
 
     if (natsOptions_Create(&m_opts) != NATS_OK)
+        the_status = NATS_NO_MEMORY;
 
+    the_status = natsOptions_UseGlobalMessageDelivery(m_opts, true);
+
+    // Indicate which loop and callbacks to use once connected.
+    if (the_status == NATS_OK)
+        the_status = natsOptions_SetEventLoop(m_opts, (void *)an_event_loop,
+                                              natsLibevent_Attach,
+                                              natsLibevent_Read,
+                                              natsLibevent_Write,
+                                              natsLibevent_Detach);
+
+    the_status = natsOptions_SetURL(m_opts, kNatsServerURL);
+
+    if (the_status == NATS_OK)
+        the_status = natsConnection_Connect(&m_conn, m_opts);
+
+    if (the_status != NATS_OK)
+    {
+        nats_PrintLastErrorStack(stderr);
+        return -1;
+    }
+
+    return 0;
+}
+
+/*!  ----------------------------------------------------------------------------
+ *  @fn terminate
+ *
+ *  @brief terminate the connection with the Broker server.
+ */
+int NatsBroker::terminate(void)
+{
+    natsSubscription_Destroy(m_sub);
+    natsConnection_Destroy(m_conn);
+    natsOptions_Destroy(m_opts);
+
+    nats_Close();
+    return 0;
+}
+
+/*!  ----------------------------------------------------------------------------
+ *  @fn connect
+ *
+ *  @brief connect to the Broker server.
+ */
+int NatsBroker::connect(void)
+{
+    natsStatus the_status;
+    the_status = natsConnection_Connect(&m_conn, m_opts);
+
+   if (the_status == NATS_OK)
+        the_status = natsConnection_Subscribe(&m_sub, m_conn, "foo", onMsg, NULL);
+
+    // For maximum performance, set no limit on the number of pending messages.
+    if (the_status == NATS_OK)
+        the_status = natsSubscription_SetPendingLimits(m_sub, -1, -1);
+
+    // If there was an error, print a stack trace and exit
+    if (the_status != NATS_OK)
+    {
+        nats_PrintLastErrorStack(stderr);
+        return -1;
+    }
 
     return 0;
 }