Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 31 additions & 23 deletions src/adapters/libevent.h
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,32 @@ keepAliveCb(evutil_socket_t fd, short flags, void * arg)
// do nothing...
}

static void
natsLibeventEvents_free(natsLibeventEvents *nle, bool processDetachedEvents)
{
if (nle->read != NULL)
event_free(nle->read);
if (nle->write != NULL)
event_free(nle->write);
if (nle->keepActive != NULL)
{
event_active(nle->keepActive, 0, 0);
event_free(nle->keepActive);
}
if (processDetachedEvents)
natsConnection_ProcessDetachedEvent(nle->nc);
free(nle);
}

// This callback is invoked from the event loop thread and will free the
// `natsLibeventEvents` object.
static void
_freeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
{
natsLibeventEvents *nle = (natsLibeventEvents*) arg;
natsLibeventEvents_free(nle, true);
}

/** \brief Attach a connection to the given event loop.
*
* This callback is invoked after `NATS` library has connected, or reconnected.
Expand All @@ -101,6 +127,7 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so
struct event_base *libeventLoop = (struct event_base*) loop;
natsLibeventEvents *nle = (natsLibeventEvents*) (*userData);
natsStatus s = NATS_OK;
bool created = false;

// This is the first attach (when reconnecting, nle will be non-NULL).
if (nle == NULL)
Expand All @@ -109,6 +136,8 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so
if (nle == NULL)
return NATS_NO_MEMORY;

// Indicate that we have created the object here (in case we get a failure).
created = true;
nle->nc = nc;
nle->loop = libeventLoop;

Expand Down Expand Up @@ -153,8 +182,8 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so

if (s == NATS_OK)
*userData = (void*) nle;
else
natsLibevent_Detach((void*) nle);
else if (created)
natsLibeventEvents_free(nle, false);

return s;
}
Expand Down Expand Up @@ -224,27 +253,6 @@ natsLibevent_Write(void *userData, bool add)
return (res == 0 ? NATS_OK : NATS_ERR);
}

// This callback is invoked from the event loop thread and will free the
// `natsLibeventEvents` object.
static void
_freeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
{
natsLibeventEvents *nle = (natsLibeventEvents*) arg;

if (nle->read != NULL)
event_free(nle->read);
if (nle->write != NULL)
event_free(nle->write);
if (nle->keepActive != NULL)
{
event_active(nle->keepActive, 0, 0);
event_free(nle->keepActive);
}
// This will release the connection that is retained by the library on the first attach.
natsConnection_Destroy(nle->nc);
free(nle);
}

/** \brief The connection is closed, it can be safely detached.
*
* When a connection is closed (not disconnected, pending a reconnect), this
Expand Down
31 changes: 22 additions & 9 deletions src/adapters/libuv.h
Original file line number Diff line number Diff line change
Expand Up @@ -237,24 +237,33 @@ uvAsyncAttach(natsLibuvEvents *nle)
}

static void
uvFinalCloseCb(uv_handle_t* handle)
natsLibuvEvents_free(natsLibuvEvents *nle, bool processDetachedEvent)
{
natsLibuvEvents *nle = (natsLibuvEvents*) handle->data;
natsLibuvEvent *event;
natsLibuvEvent *event;

while ((event = nle->head) != NULL)
{
nle->head = event->next;
free(event);
}
free(nle->scheduler);
uv_mutex_destroy(nle->lock);
free(nle->lock);
// This will release the connection that is retained by the library on the first attach.
natsConnection_Destroy(nle->nc);
if (nle->lock != NULL)
{
uv_mutex_destroy(nle->lock);
free(nle->lock);
}
if (processDetachedEvent)
natsConnection_ProcessDetachedEvent(nle->nc);
free(nle);
}

static void
uvFinalCloseCb(uv_handle_t* handle)
{
natsLibuvEvents *nle = (natsLibuvEvents*) handle->data;
natsLibuvEvents_free(nle, true);
}

static void
uvAsyncDetach(natsLibuvEvents *nle)
{
Expand Down Expand Up @@ -348,6 +357,7 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke
bool sched = false;
natsLibuvEvents *nle = (natsLibuvEvents*) (*userData);
natsStatus s = NATS_OK;
bool created = false;

sched = ((uv_key_get(&uvLoopThreadKey) != loop) ? true : false);

Expand All @@ -362,6 +372,9 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke
if (nle == NULL)
return NATS_NO_MEMORY;

// Indicate that we have created the object here (in case we get a failure).
created = true;

nle->lock = (uv_mutex_t*) malloc(sizeof(uv_mutex_t));
if (nle->lock == NULL)
s = NATS_NO_MEMORY;
Expand Down Expand Up @@ -402,8 +415,8 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke

if (s == NATS_OK)
*userData = (void*) nle;
else
natsLibuv_Detach((void*) nle);
else if (created)
natsLibuvEvents_free(nle, false);

return s;
}
Expand Down
6 changes: 6 additions & 0 deletions src/conn.c
Original file line number Diff line number Diff line change
Expand Up @@ -4367,6 +4367,12 @@
*socket = NATS_SOCK_INVALID;
}

void
natsConnection_ProcessDetachedEvent(natsConnection *nc)

Check warning on line 4371 in src/conn.c

View check run for this annotation

Codecov / codecov/patch

src/conn.c#L4371

Added line #L4371 was not covered by tests
{
natsConn_release(nc);

Check warning on line 4373 in src/conn.c

View check run for this annotation

Codecov / codecov/patch

src/conn.c#L4373

Added line #L4373 was not covered by tests
}

natsStatus
natsConnection_GetClientID(natsConnection *nc, uint64_t *cid)
{
Expand Down
13 changes: 13 additions & 0 deletions src/nats.h
Original file line number Diff line number Diff line change
Expand Up @@ -5085,6 +5085,19 @@ natsConnection_ProcessCloseEvent(natsSock *socket);
NATS_EXTERN void
natsConnection_ProcessWriteEvent(natsConnection *nc);

/** \brief Process a detach event when using external event loop.
*
* When a connection is closed, the library will invoke the adapter's
* #natsEvLoop_Detach callback. But the code in the adapter may run
* asynchronously. The adapter will invoke this function when the
* adapter has fully detached the NATS connection from the event loop,
* so that resources held by the library for this connection can be released.
*
* @param nc the pointer to the #natsConnection object.
*/
NATS_EXTERN void
natsConnection_ProcessDetachedEvent(natsConnection *nc);

/** \brief Connects to a `NATS Server` using any of the URL from the given list.
*
* Attempts to connect to a `NATS Server`.
Expand Down
Loading