domo/domo-iot
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

domo-iot replace libevent by libuv

Browse Source
This commit is contained in:
NADAL Jean-Baptiste
2019-12-03 22:38:01 +01:00
parent e5bd19f12d
commit 524a9f3928
6 changed files with 49 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
src/broker/nats-broker.cpp
View File

@@ -28,7 +28,7 @@
/*------------------------------- INCLUDES ----------------------------------*/
#include <adapters/libevent.h>
#include <adapters/libuv.h>
#include "nats-broker.h"
@@ -36,7 +36,6 @@
#define kNatsServerURL "nats.nadal-fr.com"
static void onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
{
printf("Received msg: %s - %.*s\n",
@@ -71,38 +70,39 @@ NatsBroker::~NatsBroker(void)
*
* @brief Setup the Broker.
*/
int NatsBroker::setup(struct event_base *an_event_loop)
int NatsBroker::setup(uv_loop_t *an_evt_loop)
{
natsStatus the_status;
nats_Open(-1);
// One time initialization of things that we need.
natsLibevent_Init();
natsLibuv_Init();
// Libuv is not thread-safe. Almost all calls to libuv need to
// occur from the thread where the loop is running. NATS library
// may have to call into the event loop from different threads.
// This call allows natsLibuv APIs to know if they are executing
// from the event loop thread or not.
natsLibuv_SetThreadLocalLoop(an_evt_loop);
if (natsOptions_Create(&m_opts) != NATS_OK)
the_status = NATS_NO_MEMORY;
the_status = natsOptions_UseGlobalMessageDelivery(m_opts, true);
return -1;
if (natsOptions_UseGlobalMessageDelivery(m_opts, true) != NATS_OK)
return -2;
// 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);
if (natsOptions_SetEventLoop(m_opts, (void *)an_evt_loop,
natsLibuv_Attach,
natsLibuv_Read,
natsLibuv_Write,
natsLibuv_Detach) != NATS_OK)
return -3;
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;
}
if (natsOptions_SetURL(m_opts, kNatsServerURL) != NATS_OK)
return -4;
return 0;
}
@@ -110,41 +110,37 @@ int NatsBroker::setup(struct event_base *an_event_loop)
/*! ----------------------------------------------------------------------------
* @fn terminate
*
* @brief terminate the connection with the Broker server.
* @brief Terminate the Broker.
*/
int NatsBroker::terminate(void)
{
natsSubscription_Destroy(m_sub);
natsConnection_Destroy(m_conn);
natsOptions_Destroy(m_opts);
// Destroy all our objects to avoid report of memory leak
if (m_sub != NULL)
natsSubscription_Destroy(m_sub);
if (m_conn != NULL)
natsConnection_Destroy(m_conn);
if (m_opts != NULL)
natsOptions_Destroy(m_opts);
nats_Close();
return 0;
}
/*! ----------------------------------------------------------------------------
* @fn connect
*
* @brief connect to the Broker server.
* @brief Ccnnect to the Broker server.
*/
int NatsBroker::connect(void)
{
natsStatus the_status;
the_status = natsConnection_Connect(&m_conn, m_opts);
if (natsConnection_Connect(&m_conn, m_opts) != NATS_OK)
return -1;
if (the_status == NATS_OK)
the_status = natsConnection_Subscribe(&m_sub, m_conn, "foo", onMsg, NULL);
if (natsConnection_Subscribe(&m_sub, m_conn, "foo", onMsg, NULL) != NATS_OK)
return -2;
// 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;
}
if (natsSubscription_SetPendingLimits(m_sub, -1, -1)!= NATS_OK)
return -3;
return 0;
}