Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

added natsConn_(get|set)(Error|Closed)Callback #647

Closed
wants to merge 4 commits into from
Closed

added natsConn_(get|set)(Error|Closed)Callback #647

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c40813f
added get/set(Error|Closed)Callbacks to use in microservices
levb Apr 6, 2023
87baf2a
PR feedback
levb Apr 7, 2023
c93e964
fixed test_StanPing
levb Apr 7, 2023
ebc3de3
added test_SetConnClosedCB
levb Apr 7, 2023
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
56 changes: 56 additions & 0 deletions src/conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4432,3 +4432,59 @@ natsConn_defaultErrHandler(natsConnection *nc, natsSubscription *sub, natsStatus
}
fflush(stderr);
}

natsStatus
natsConn_getErrorCallback(natsErrHandler *cb, void **closure, natsConnection *nc)
{
if ((nc == NULL) || (cb == NULL) || (closure == NULL))
return nats_setDefaultError(NATS_INVALID_ARG);

natsConn_Lock(nc);
*cb = nc->opts->asyncErrCb;
*closure = nc->opts->asyncErrCbClosure;
natsConn_Unlock(nc);

return NATS_OK;
}

natsStatus
natsConn_setErrorCallback(natsConnection *nc, natsErrHandler cb, void *closure)
{
if ((nc == NULL) || (cb == NULL))
return nats_setDefaultError(NATS_INVALID_ARG);

natsConn_Lock(nc);
nc->opts->asyncErrCb = cb;
nc->opts->asyncErrCbClosure = closure;
natsConn_Unlock(nc);

return NATS_OK;
}

natsStatus
natsConn_getClosedCallback(natsConnectionHandler *cb, void **closure, natsConnection *nc)
{
if ((nc == NULL) || (cb == NULL) || (closure == NULL))
return nats_setDefaultError(NATS_INVALID_ARG);

natsConn_Lock(nc);
*cb = nc->opts->closedCb;
*closure = nc->opts->closedCbClosure;
natsConn_Unlock(nc);

return NATS_OK;
}

natsStatus
natsConn_setClosedCallback(natsConnection *nc, natsConnectionHandler cb, void *closure)
{
if ((nc == NULL) || (cb == NULL))
return nats_setDefaultError(NATS_INVALID_ARG);

natsConn_Lock(nc);
nc->opts->closedCb = cb;
nc->opts->closedCbClosure = closure;
natsConn_Unlock(nc);

return NATS_OK;
}
13 changes: 13 additions & 0 deletions src/conn.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,4 +160,17 @@ natsConn_close(natsConnection *nc);
void
natsConn_destroy(natsConnection *nc, bool fromPublicDestroy);

natsStatus
natsConn_setErrorCallback(natsConnection *nc, natsErrHandler cb, void *closure);

natsStatus
natsConn_getErrorCallback(natsErrHandler *cb, void **closure, natsConnection *nc);

natsStatus
natsConn_setClosedCallback(natsConnection *nc, natsConnectionHandler cb, void *closure);

natsStatus
natsConn_getClosedCallback(natsConnectionHandler *cb, void **closure, natsConnection *nc);


#endif /* CONN_H_ */
64 changes: 44 additions & 20 deletions src/nats.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -726,11 +726,14 @@ _timerThread(void *arg)
static void
_asyncCbsThread(void *arg)
{
natsLibAsyncCbs *asyncCbs = &(gLib.asyncCbs);
natsAsyncCbInfo *cb = NULL;
natsConnection *nc = NULL;
natsLibAsyncCbs *asyncCbs = &(gLib.asyncCbs);
natsAsyncCbInfo *cb = NULL;
natsConnection *nc = NULL;
natsConnectionHandler cbHandler = NULL;
natsErrHandler errHandler = NULL;
void *cbClosure = NULL;
#if defined(NATS_HAS_STREAMING)
stanConnection *sc = NULL;
stanConnection *sc = NULL;
#endif

WAIT_LIB_INITIALIZED;
Expand Down Expand Up @@ -760,41 +763,62 @@ _asyncCbsThread(void *arg)
sc = cb->sc;
#endif

// callback handlers can be updated on a live connection, so we need to
// lock.
natsMutex_Lock(nc->mu);
levb marked this conversation as resolved.
Show resolved Hide resolved
switch (cb->type)
{
case ASYNC_CLOSED:
(*(nc->opts->closedCb))(nc, nc->opts->closedCbClosure);
cbHandler = nc->opts->closedCb;
cbClosure = nc->opts->closedCbClosure;
break;
case ASYNC_DISCONNECTED:
(*(nc->opts->disconnectedCb))(nc, nc->opts->disconnectedCbClosure);
cbHandler = nc->opts->disconnectedCb;
cbClosure = nc->opts->disconnectedCbClosure;
break;
case ASYNC_RECONNECTED:
(*(nc->opts->reconnectedCb))(nc, nc->opts->reconnectedCbClosure);
cbHandler = nc->opts->reconnectedCb;
cbClosure = nc->opts->reconnectedCbClosure;
break;
case ASYNC_CONNECTED:
(*(nc->opts->connectedCb))(nc, nc->opts->connectedCbClosure);
cbHandler = nc->opts->connectedCb;
cbClosure = nc->opts->connectedCbClosure;
break;
case ASYNC_DISCOVERED_SERVERS:
(*(nc->opts->discoveredServersCb))(nc, nc->opts->discoveredServersClosure);
cbHandler = nc->opts->discoveredServersCb;
cbClosure = nc->opts->discoveredServersClosure;
break;
case ASYNC_LAME_DUCK_MODE:
(*(nc->opts->lameDuckCb))(nc, nc->opts->lameDuckClosure);
cbHandler = nc->opts->lameDuckCb;
cbClosure = nc->opts->lameDuckClosure;
break;
case ASYNC_ERROR:
{
if (cb->errTxt != NULL)
nats_setErrStatusAndTxt(cb->err, cb->errTxt);
(*(nc->opts->asyncErrCb))(nc, cb->sub, cb->err, nc->opts->asyncErrCbClosure);
break;
}
#if defined(NATS_HAS_STREAMING)
case ASYNC_STAN_CONN_LOST:
(*(sc->opts->connectionLostCB))(sc, sc->connLostErrTxt, sc->opts->connectionLostCBClosure);
errHandler = nc->opts->asyncErrCb;
cbClosure = nc->opts->asyncErrCbClosure;
break;
#endif
default:
break;
}
natsMutex_Unlock(nc->mu);

// Invoke the callback
if (cbHandler != NULL)
{
(*(cbHandler))(nc, cbClosure);
}
else if (errHandler != NULL)
{
if (cb->errTxt != NULL)
nats_setErrStatusAndTxt(cb->err, cb->errTxt);
(*(errHandler))(nc, cb->sub, cb->err, cbClosure);
break;
levb marked this conversation as resolved.
Show resolved Hide resolved
}
#if defined(NATS_HAS_STREAMING)
else if (cb->type == ASYNC_STAN_CONN_LOST)
{
(*(sc->opts->connectionLostCB))(sc, sc->connLostErrTxt, sc->opts->connectionLostCBClosure);
}
#endif

natsAsyncCb_Destroy(cb);
nats_clearLastError();
Expand Down
74 changes: 74 additions & 0 deletions test/test.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14373,6 +14373,79 @@ test_AsyncErrHandler(void)
_stopServer(serverPid);
}

static void
test_AsyncSetErrHandler(void)
{
natsStatus s;
natsConnection *nc = NULL;
natsOptions *opts = NULL;
natsSubscription *sub = NULL;
natsPid serverPid = NATS_INVALID_PID;
struct threadArg arg;

s = _createDefaultThreadArgsForCbTests(&arg);
if (s != NATS_OK)
FAIL("Unable to setup test!");

arg.status = NATS_OK;
arg.control= 7;

s = natsOptions_Create(&opts);
IFOK(s, natsOptions_SetURL(opts, NATS_DEFAULT_URL));
IFOK(s, natsOptions_SetMaxPendingMsgs(opts, 10));

if (s != NATS_OK)
FAIL("Unable to create options for test AsyncErrHandler");

serverPid = _startServer("nats://127.0.0.1:4222", NULL, true);
CHECK_SERVER_STARTED(serverPid);

s = natsConnection_Connect(&nc, opts);
IFOK(s, natsConnection_Subscribe(&sub, nc, "async_test", _recvTestString, (void*) &arg));

natsMutex_Lock(arg.m);
arg.sub = sub;
natsMutex_Unlock(arg.m);

// Start sending messages
for (int i=0;
(s == NATS_OK) && (i < (opts->maxPendingMsgs)); i++)
{
s = natsConnection_PublishString(nc, "async_test", "hello");
}

// Set the error handler in-flight
IFOK(s, natsConn_setErrorCallback(nc, _asyncErrCb, (void*) &arg));

for (int i=0;
(s == NATS_OK) && (i < 100); i++)
{
s = natsConnection_PublishString(nc, "async_test", "hello");
}
IFOK(s, natsConnection_Flush(nc));

// Wait for async err callback
natsMutex_Lock(arg.m);
while ((s != NATS_TIMEOUT) && !arg.done)
s = natsCondition_TimedWait(arg.c, arg.m, 2000);
natsMutex_Unlock(arg.m);

test("Aync fired properly, and all checks are good: ");
testCond((s == NATS_OK)
&& arg.done
&& arg.closed
&& (arg.status == NATS_OK));

natsOptions_Destroy(opts);
natsSubscription_Destroy(sub);
natsConnection_Destroy(nc);

_destroyDefaultThreadArgs(&arg);

_stopServer(serverPid);
}


static void
_responseCb(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
{
Expand Down Expand Up @@ -34440,6 +34513,7 @@ static testInfo allTests[] =
{"SyncSubscriptionPending", test_SyncSubscriptionPending},
{"SyncSubscriptionPendingDrain", test_SyncSubscriptionPendingDrain},
{"AsyncErrHandler", test_AsyncErrHandler},
{"AsyncSetErrHandler", test_AsyncSetErrHandler},
levb marked this conversation as resolved.
Show resolved Hide resolved
{"AsyncSubscriberStarvation", test_AsyncSubscriberStarvation},
{"AsyncSubscriberOnClose", test_AsyncSubscriberOnClose},
{"NextMsgCallOnAsyncSub", test_NextMsgCallOnAsyncSub},
Expand Down