/* omazureeventhubs.c * This output plugin make rsyslog talk to Azure EventHubs. * * Copyright 2014-2017 by Adiscon GmbH. * * This file is part of rsyslog. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * -or- * see COPYING.ASL20 in the source distribution * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #include "config.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #ifdef HAVE_SYS_STAT_H # include #endif #include #include // Include Proton headers #include #include #include #include #include #include #include #include #include #include #include #include #include #include // Include rsyslog headers #include "rsyslog.h" #include "syslogd-types.h" #include "srUtils.h" #include "template.h" #include "module-template.h" #include "errmsg.h" #include "atomic.h" #include "statsobj.h" #include "unicode-helper.h" #include "datetime.h" #include "glbl.h" MODULE_TYPE_OUTPUT MODULE_TYPE_NOKEEP MODULE_CNFNAME("omazureeventhubs") /* internal structures */ DEF_OMOD_STATIC_DATA DEFobjCurrIf(glbl) DEFobjCurrIf(datetime) DEFobjCurrIf(strm) DEFobjCurrIf(statsobj) statsobj_t *azureStats; STATSCOUNTER_DEF(ctrMessageSubmit, mutCtrMessageSubmit); STATSCOUNTER_DEF(ctrAzureFail, mutCtrAzureFail); STATSCOUNTER_DEF(ctrCacheMiss, mutCtrCacheMiss); STATSCOUNTER_DEF(ctrCacheEvict, mutCtrCacheEvict); STATSCOUNTER_DEF(ctrCacheSkip, mutCtrCacheSkip); STATSCOUNTER_DEF(ctrAzureAck, mutCtrAzureAck); STATSCOUNTER_DEF(ctrAzureMsgTooLarge, mutCtrAzureMsgTooLarge); STATSCOUNTER_DEF(ctrAzureQueueFull, mutCtrAzureQueueFull); STATSCOUNTER_DEF(ctrAzureOtherErrors, mutCtrAzureOtherErrors); STATSCOUNTER_DEF(ctrAzureRespTimedOut, mutCtrAzureRespTimedOut); STATSCOUNTER_DEF(ctrAzureRespTransport, mutCtrAzureRespTransport); STATSCOUNTER_DEF(ctrAzureRespBrokerDown, mutCtrAzureRespBrokerDown); STATSCOUNTER_DEF(ctrAzureRespAuth, mutCtrAzureRespAuth); STATSCOUNTER_DEF(ctrAzureRespSSL, mutCtrAzureRespSSL); STATSCOUNTER_DEF(ctrAzureRespOther, mutCtrAzureRespOther); #define MAX_ERRMSG 1024 /* max size of error messages that we support */ #define MAX_DEFAULTMSGS 1024 /* Initial max size of the proton message helper array */ #define SETUP_PROTON_NONE 0 #define SETUP_PROTON_AUTOCLOSE 1 /* flags for writeAzure: shall we resubmit a failed message? */ #define RESUBMIT 1 #define NO_RESUBMIT 0 /* flags for transaction Handling */ enum proton_submission_status { PROTON_UNSUBMITTED = 0, // Message not submitted yet PROTON_SUBMITTED, // Message submitted to proton sender instance PROTON_ACCEPTED, // Message accepted from remote target PROTON_REJECTED, // Message rejected from remote target (zero credit?) }; // event_property NEEDED? struct event_property { const char *key; const char *val; }; static pn_timestamp_t time_now(void); /* Struct for Proton Messages Listitems */ struct s_protonmsg_entry { uchar* payload; size_t payload_len; uchar* MsgID; size_t MsgID_len; uchar* address; char status; }; typedef struct s_protonmsg_entry protonmsg_entry; /* Struct for module InstanceData */ typedef struct _instanceData { uchar *amqp_address; uchar *azurehost; uchar *azureport; uchar *azure_key_name; uchar *azure_key; uchar *container; uchar *tplName; /* assigned output template */ int nEventProperties; struct event_property *eventProperties; uchar *statsName; statsobj_t *stats; STATSCOUNTER_DEF(ctrMessageSubmit, mutCtrMessageSubmit); STATSCOUNTER_DEF(ctrAzureFail, mutCtrAzureFail); STATSCOUNTER_DEF(ctrAzureAck, mutCtrAzureAck); STATSCOUNTER_DEF(ctrAzureOtherErrors, mutCtrAzureOtherErrors); } instanceData; /* Struct for module workerInstanceData */ typedef struct wrkrInstanceData { instanceData *pData; protonmsg_entry **aProtonMsgs; /* dynamically sized array for transactional outputs */ unsigned int nProtonMsgs; /* current used proton msgs */ unsigned int nMaxProtonMsgs; /* current max */ int bIsConnecting; /* 1 if connecting, 0 if disconnected */ int bIsConnected; /* 1 if connected, 0 if disconnected */ int bIsSuspended; /* when broker fail, we need to suspend the action */ pthread_rwlock_t pnLock; // PROTON Handles pn_proactor_t *pnProactor; pn_transport_t *pnTransport; pn_connection_t *pnConn; pn_link_t* pnSender; pn_rwbytes_t pnMessageBuffer; /* Buffer for messages */ int pnStatus; // Message Counters for sender link in worker instance unsigned int iMsgSeq; unsigned int iMaxMsgSeq; /* The following structure controls the proton handling threads. Passes necessary pointers * needed for their access. */ sbool bThreadRunning; pthread_t tid; /* the worker's thread ID */ } wrkrInstanceData_t; #define INST_STATSCOUNTER_INC(inst, ctr, mut) \ do { \ if (inst->stats) { STATSCOUNTER_INC(ctr, mut); } \ } while(0); // QPID Proton Handler functions static rsRetVal proton_run_thread(wrkrInstanceData_t *pWrkrData); static rsRetVal proton_shutdown_thread(wrkrInstanceData_t *pWrkrData); static void * proton_thread(void *myInfo); static void handleProtonDelivery(wrkrInstanceData_t *const pWrkrData); static void handleProton(wrkrInstanceData_t *const pWrkrData, pn_event_t *event); static rsRetVal writeProton(wrkrInstanceData_t *__restrict__ const pWrkrData, const actWrkrIParams_t *__restrict__ const pParam, const int iMsg); /* tables for interfacing with the v6 config system */ /* action (instance) parameters */ static struct cnfparamdescr actpdescr[] = { { "azurehost", eCmdHdlrString, CNFPARAM_REQUIRED }, { "azureport", eCmdHdlrString, CNFPARAM_REQUIRED }, { "azure_key_name", eCmdHdlrString, CNFPARAM_REQUIRED }, { "azure_key", eCmdHdlrString, CNFPARAM_REQUIRED }, { "amqp_address", eCmdHdlrString, 0 }, { "container", eCmdHdlrString, 0 }, { "eventproperties", eCmdHdlrArray, 0 }, { "template", eCmdHdlrGetWord, 0 }, { "statsname", eCmdHdlrGetWord, 0 } }; static struct cnfparamblk actpblk = { CNFPARAMBLK_VERSION, sizeof(actpdescr)/sizeof(struct cnfparamdescr), actpdescr }; BEGINinitConfVars /* (re)set config variables to default values */ CODESTARTinitConfVars ENDinitConfVars static void ATTR_NONNULL(1) protonmsg_entry_destruct(protonmsg_entry *const __restrict__ fmsgEntry) { free(fmsgEntry->MsgID); free(fmsgEntry->payload); free(fmsgEntry->address); free(fmsgEntry); } /* note: we need the length of message as we need to deal with * non-NUL terminated strings under some circumstances. */ static protonmsg_entry * ATTR_NONNULL(1,3) protonmsg_entry_construct( const char *const MsgID, const size_t msgidlen, const char *const msg, const size_t msglen, const char *const address) { protonmsg_entry *etry = NULL; if((etry = malloc(sizeof(struct s_protonmsg_entry))) == NULL) { return NULL; } etry->status = PROTON_UNSUBMITTED; // Unsubmitted default */ etry->MsgID_len = msgidlen; if((etry->MsgID = (uchar*)malloc(msgidlen+1)) == NULL) { free(etry); return NULL; } memcpy(etry->MsgID, MsgID, msgidlen); etry->MsgID[msgidlen] = '\0'; etry->payload_len = msglen; if((etry->payload = (uchar*)malloc(msglen+1)) == NULL) { free(etry->MsgID); free(etry); return NULL; } memcpy(etry->payload, msg, msglen); etry->payload[msglen] = '\0'; if((etry->address = (uchar*)strdup(address)) == NULL) { free(etry->MsgID); free(etry->payload); free(etry); return NULL; } return etry; } /* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */ static pn_message_t* proton_encode_message(wrkrInstanceData_t *const pWrkrData, protonmsg_entry* pMsgEntry) { instanceData *const pData = (instanceData *const) pWrkrData->pData; /* Construct a message with the map */ pn_message_t* message = pn_message(); // Optionally include Address? // pn_message_set_address(message, (char *) pWrkrData->amqp_address); // Send in BINARY MODE ( as Stream ) pn_message_set_content_type(message, (char*) "application/octect-stream"); pn_message_set_creation_time(message, time_now()); pn_message_set_inferred(message, true); // Set message ID pn_message_set_id(message, (pn_atom_t){ .type=PN_STRING, .u.as_bytes.start = (char*)pMsgEntry->MsgID, .u.as_bytes.size = pMsgEntry->MsgID_len }); if (pData->nEventProperties > 0) { // Add Event properties pn_data_t *props = pn_message_properties(message); pn_data_put_map(props); pn_data_enter(props); for(int i = 0 ; i < pData->nEventProperties ; ++i) { DBGPRINTF("proton_encode_message: add eventproperty %s:%s\n", pData->eventProperties[i].key, pData->eventProperties[i].val); pn_data_put_string(props, pn_bytes(strlen(pData->eventProperties[i].key), pData->eventProperties[i].key)); pn_data_put_string(props, pn_bytes(strlen(pData->eventProperties[i].val), pData->eventProperties[i].val)); } pn_data_exit(props); } // Set message BODY pn_data_t* body = pn_message_body(message); pn_data_enter(body); pn_data_put_binary(body, pn_bytes(pMsgEntry->payload_len, (char*)pMsgEntry->payload)); pn_data_exit(body); DBGPRINTF("proton_encode_message: created message id '%s': '%.*s'\n", (char*)pMsgEntry->MsgID, (pMsgEntry->payload_len > 0 ? (int)pMsgEntry->payload_len-1 : 0), (char*)pMsgEntry->payload); return message; } static rsRetVal closeProton(wrkrInstanceData_t *const __restrict__ pWrkrData) { DEFiRet; instanceData *const pData = (instanceData *const) pWrkrData->pData; #ifndef NDEBUG DBGPRINTF("closeProton[%p]: ENTER\n", pWrkrData); #endif if (pWrkrData->pnSender) { pn_link_close(pWrkrData->pnSender); DBGPRINTF("closeProton[%p]: pn_link_close\n", pWrkrData); pn_session_close(pn_link_session(pWrkrData->pnSender)); DBGPRINTF("closeProton[%p]: pn_session_close\n", pWrkrData); } if (pWrkrData->pnConn) { DBGPRINTF("closeProton[%p]: pn_connection_close connection\n", pWrkrData); pn_connection_close(pWrkrData->pnConn); } pWrkrData->bIsConnecting = 0; pWrkrData->bIsConnected = 0; pWrkrData->pnStatus = PN_EVENT_NONE; pWrkrData->pnSender = NULL; pWrkrData->pnConn = NULL; pWrkrData->iMsgSeq = 0; pWrkrData->iMaxMsgSeq = 0; // Mark all remaining entries as REJECTED if(pWrkrData->aProtonMsgs != NULL) { for(unsigned int i = 0 ; i < pWrkrData->nProtonMsgs ; ++i) { if (pWrkrData->aProtonMsgs[i] != NULL && ( pWrkrData->aProtonMsgs[i]->status == PROTON_UNSUBMITTED || pWrkrData->aProtonMsgs[i]->status == PROTON_SUBMITTED) ) { pWrkrData->aProtonMsgs[i]->status = PROTON_REJECTED; DBGPRINTF("closeProton[%p]: Setting ProtonMsg %s to PROTON_REJECTED \n", pWrkrData, pWrkrData->aProtonMsgs[i]->MsgID); // Increment Stats Counter STATSCOUNTER_INC(ctrAzureFail, mutCtrAzureFail); INST_STATSCOUNTER_INC(pData, pData->ctrAzureFail, pData->mutCtrAzureFail); } } } FINALIZE; finalize_it: RETiRet; } static rsRetVal openProton(wrkrInstanceData_t *const __restrict__ pWrkrData) { DEFiRet; instanceData *const pData = (instanceData *const) pWrkrData->pData; int pnErr = PN_OK; char szAddr[PN_MAX_ADDR]; pn_ssl_t* pnSsl; #ifndef NDEBUG DBGPRINTF("openProton[%p]: ENTER\n", pWrkrData); #endif if(pWrkrData->bIsConnecting == 1 || pWrkrData->bIsConnected == 1) FINALIZE; pWrkrData->pnStatus = PN_EVENT_NONE; pn_proactor_addr(szAddr, sizeof(szAddr), (const char *) pData->azurehost, (const char *) pData->azureport); // Configure a transport for SSL. The transport will be freed by the proactor. pWrkrData->pnTransport = pn_transport(); DBGPRINTF("openProton[%p]: create transport to '%s:%s'\n", pWrkrData, pData->azurehost, pData->azureport); pnSsl = pn_ssl(pWrkrData->pnTransport); if (pnSsl != NULL) { pn_ssl_domain_t* pnDomain = pn_ssl_domain(PN_SSL_MODE_CLIENT); if (pData->azure_key_name != NULL && pData->azure_key != NULL) { pnErr = pn_ssl_init(pnSsl, pnDomain, NULL); if (pnErr) { DBGPRINTF("openProton[%p]: pn_ssl_init failed for '%s:%s' with error %d: %s\n", pWrkrData, pData->azurehost, pData->azureport, pnErr, pn_code(pnErr)); } pn_sasl_allowed_mechs(pn_sasl(pWrkrData->pnTransport), "PLAIN"); } else { pnErr = pn_ssl_domain_set_peer_authentication(pnDomain, PN_SSL_ANONYMOUS_PEER, NULL); if (!pnErr) { pnErr = pn_ssl_init(pnSsl, pnDomain, NULL); } else { DBGPRINTF( "openProton[%p]: pn_ssl_domain_set_peer_authentication failed with '%d'\n", pWrkrData, pnErr); } } pn_ssl_domain_free(pnDomain); } else { LogError(0, RS_RET_ERR, "openProton[%p]: openProton pn_ssl_init NULL", pWrkrData); } // Handle ERROR Output if (pnErr) { LogError(0, RS_RET_IO_ERROR, "openProton[%p]: creating transport to '%s:%s' " "failed with error %d: %s\n", pWrkrData, pData->azurehost, pData->azureport, pnErr, pn_code(pnErr)); ABORT_FINALIZE(RS_RET_IO_ERROR); } // Connect to Azure Event Hubs pn_proactor_connect2(pWrkrData->pnProactor, NULL, pWrkrData->pnTransport, szAddr); // Successfully connecting pWrkrData->bIsConnecting = 1; pWrkrData->bIsSuspended = 0; finalize_it: if(iRet != RS_RET_OK) { closeProton(pWrkrData); // Make sure to free ressources } RETiRet; } static sbool proton_check_condition( pn_event_t *event, wrkrInstanceData_t *const __restrict__ pWrkrData, pn_condition_t *cond, const char * pszReason) { if (pn_condition_is_set(cond)) { DBGPRINTF("proton_check_condition: %s %s: %s: %s", pszReason, pn_event_type_name(pn_event_type(event)), pn_condition_get_name(cond), pn_condition_get_description(cond)); LogError(0, RS_RET_ERR, "omazureeventhubs: %s %s: %s: %s", pszReason, pn_event_type_name(pn_event_type(event)), pn_condition_get_name(cond), pn_condition_get_description(cond)); // Connection can be closed closeProton(pWrkrData); // Set Worker to suspended state! pWrkrData->bIsSuspended = 1; return 0; } else { return 1; } } static rsRetVal setupProtonHandle(wrkrInstanceData_t *const __restrict__ pWrkrData, int autoclose) { DEFiRet; DBGPRINTF("omazureeventhubs[%p]: setupProtonHandle ENTER\n", pWrkrData); pthread_rwlock_wrlock(&pWrkrData->pnLock); if (autoclose == SETUP_PROTON_AUTOCLOSE && (pWrkrData->bIsConnected == 1)) { DBGPRINTF("omazureeventhubs[%p]: setupProtonHandle closeProton\n", pWrkrData); closeProton(pWrkrData); } CHKiRet(openProton(pWrkrData)); finalize_it: if (iRet != RS_RET_OK) { /* Parameter Error's cannot be resumed, so we need to disable the action */ if (iRet == RS_RET_PARAM_ERROR) { iRet = RS_RET_DISABLE_ACTION; LogError(0, iRet, "omazureeventhubs: action will be disabled due invalid " "configuration parameters\n"); } } pthread_rwlock_unlock(&pWrkrData->pnLock); RETiRet; } static rsRetVal writeProton(wrkrInstanceData_t *__restrict__ const pWrkrData, const actWrkrIParams_t *__restrict__ const pParam, const int iMsg) { DEFiRet; instanceData *const pData = (instanceData *const) pWrkrData->pData; protonmsg_entry* fmsgEntry; // Create Unqiue Message ID char szMsgID[64]; sprintf(szMsgID, "%d", pWrkrData->iMsgSeq); const char* pszParamStr = (const char*)actParam(pParam, 1 /*pData->iNumTpls*/, iMsg, 0).param; size_t tzParamStrLen = actParam(pParam, 1 /*pData->iNumTpls*/, iMsg, 0).lenStr; DBGPRINTF("omazureeventhubs[%p]: writeProton for msg %d (seq %d) msg:'%.*s%s'\n", pWrkrData, iMsg, pWrkrData->iMsgSeq, (int)(tzParamStrLen > 0 ? (tzParamStrLen > 64 ? 64 : tzParamStrLen-1) : 0), pszParamStr, (tzParamStrLen > 64 ? "..." : "")); // Increment Message sequence number pWrkrData->iMsgSeq++; // Add message to LIST for sending CHKmalloc(fmsgEntry = protonmsg_entry_construct( szMsgID, sizeof(szMsgID), pszParamStr, tzParamStrLen, (const char*)pData->amqp_address)); // Add to helper Array pWrkrData->aProtonMsgs[iMsg] = fmsgEntry; finalize_it: RETiRet; } BEGINcreateInstance CODESTARTcreateInstance DBGPRINTF("createInstance[%p]: ENTER\n", pData); pData->amqp_address = NULL; pData->azurehost = NULL; pData->azureport = NULL; pData->azure_key_name = NULL; pData->azure_key = NULL; pData->container = NULL; ENDcreateInstance BEGINcreateWrkrInstance CODESTARTcreateWrkrInstance DBGPRINTF("createWrkrInstance[%p]: ENTER\n", pWrkrData); pWrkrData->bIsConnecting = 0; pWrkrData->bIsConnected = 0; pWrkrData->bIsSuspended = 0; // Create Proton proActor in Worker Instance pWrkrData->pnProactor = pn_proactor(); pWrkrData->pnConn = NULL; pWrkrData->pnTransport = NULL; pWrkrData->pnSender = NULL; pWrkrData->iMsgSeq = 0; pWrkrData->iMaxMsgSeq = 0; pWrkrData->pnMessageBuffer.start = NULL; pWrkrData->nProtonMsgs = 0; pWrkrData->nMaxProtonMsgs = MAX_DEFAULTMSGS; CHKmalloc(pWrkrData->aProtonMsgs = calloc(MAX_DEFAULTMSGS, sizeof(struct s_protonmsg_entry))); CHKiRet(pthread_rwlock_init(&pWrkrData->pnLock, NULL)); pWrkrData->bThreadRunning = 0; pWrkrData->tid = 0; // Run Proton Worker Thread now proton_run_thread(pWrkrData); finalize_it: DBGPRINTF("createWrkrInstance[%p] returned %d\n", pWrkrData, iRet); ENDcreateWrkrInstance BEGINisCompatibleWithFeature CODESTARTisCompatibleWithFeature ENDisCompatibleWithFeature BEGINfreeInstance CODESTARTfreeInstance DBGPRINTF("freeInstance[%p]: ENTER\n", pData); if (pData->stats) { statsobj.Destruct(&pData->stats); } /* Free other mem */ free(pData->amqp_address); free(pData->azurehost); free(pData->azureport); free(pData->azure_key_name); free(pData->azure_key); free(pData->container); free(pData->tplName); free(pData->statsName); for(int i = 0 ; i < pData->nEventProperties ; ++i) { free((void*) pData->eventProperties[i].key); free((void*) pData->eventProperties[i].val); } free(pData->eventProperties); ENDfreeInstance BEGINfreeWrkrInstance CODESTARTfreeWrkrInstance DBGPRINTF("freeWrkrInstance[%p]: ENTER\n", pWrkrData); /* Closing azure first! */ pthread_rwlock_wrlock(&pWrkrData->pnLock); // Close Proton Connection closeProton(pWrkrData); // Stop Proton Handle Thread proton_shutdown_thread(pWrkrData); // Free Proton Ressources if (pWrkrData->pnProactor != NULL) { DBGPRINTF("freeWrkrInstance[%p]: FREE proactor\n", pWrkrData); pn_proactor_free(pWrkrData->pnProactor); pWrkrData->pnProactor = NULL; } free(pWrkrData->pnMessageBuffer.start); pthread_rwlock_unlock(&pWrkrData->pnLock); // Free our proton helper array if(pWrkrData->aProtonMsgs != NULL) { for(unsigned int i = 0 ; i < pWrkrData->nProtonMsgs ; ++i) { if (pWrkrData->aProtonMsgs[i] != NULL) { protonmsg_entry_destruct(pWrkrData->aProtonMsgs[i]); } } free(pWrkrData->aProtonMsgs); } pthread_rwlock_destroy(&pWrkrData->pnLock); ENDfreeWrkrInstance BEGINdbgPrintInstInfo CODESTARTdbgPrintInstInfo ENDdbgPrintInstInfo BEGINtryResume CODESTARTtryResume #ifndef NDEBUG DBGPRINTF("omazureeventhubs[%p]: tryResume ENTER\n", pWrkrData); #endif if (pWrkrData->bIsConnecting == 0 && pWrkrData->bIsConnected == 0) { DBGPRINTF("omazureeventhubs[%p]: tryResume setupProtonHandle\n", pWrkrData); CHKiRet(setupProtonHandle(pWrkrData, SETUP_PROTON_AUTOCLOSE)); } finalize_it: DBGPRINTF("omazureeventhubs[%p]: tryResume returned %d\n", pWrkrData, iRet); ENDtryResume BEGINbeginTransaction CODESTARTbeginTransaction /* we have nothing to do to begin a transaction */ DBGPRINTF("omazureeventhubs[%p]: beginTransaction ENTER\n", pWrkrData); if (pWrkrData->bIsConnecting == 0 && pWrkrData->bIsConnected == 0) { CHKiRet(setupProtonHandle(pWrkrData, SETUP_PROTON_NONE)); } finalize_it: RETiRet; ENDbeginTransaction /* * New Transaction action interface */ BEGINcommitTransaction // instanceData *__restrict__ const pData = pWrkrData->pData; unsigned i; unsigned iNeedSubmission; sbool bDone = 0; protonmsg_entry* pMsgEntry = NULL; CODESTARTcommitTransaction #ifndef NDEBUG DBGPRINTF("omazureeventhubs[%p]: commitTransaction [%d msgs] ENTER\n", pWrkrData, nParams); #endif // Handle/Expand our proton helper array if (nParams > pWrkrData->nMaxProtonMsgs) { // Free old Array if(pWrkrData->aProtonMsgs != NULL) { free(pWrkrData->aProtonMsgs); } // Expand our proton helper array DBGPRINTF("omazureeventhubs[%p]: commitTransaction expand helper array from %d to %d\n", pWrkrData, pWrkrData->nMaxProtonMsgs, nParams); pWrkrData->nMaxProtonMsgs = nParams; // Set new MAX CHKmalloc(pWrkrData->aProtonMsgs = calloc(pWrkrData->nMaxProtonMsgs, sizeof(struct s_protonmsg_entry))); } // Copy count of New Messages and increase MaxMsgSeq pWrkrData->nProtonMsgs = nParams; pWrkrData->iMaxMsgSeq += nParams; do { iNeedSubmission = 0; // Put unsubmitted messages into Proton for(i = 0 ; i < nParams ; ++i) { // Get reference to Proton Array Helper pMsgEntry = ((protonmsg_entry*)pWrkrData->aProtonMsgs[i]); if ( pMsgEntry == NULL) { // Send Message to Proton writeProton(pWrkrData, pParams, i); } else if ( pMsgEntry->status == PROTON_REJECTED) { // Reset Message Entry, it will be retried pMsgEntry->status = PROTON_UNSUBMITTED; } } bDone = 1; // Wait 100 microseconds srSleep(0, 100000); // Verify if messages have been submitted successfully for(i = 0 ; i < nParams ; ++i) { // Get reference to Proton Array Helper pMsgEntry = ((protonmsg_entry*)pWrkrData->aProtonMsgs[i]); if (pMsgEntry != NULL) { if ( pMsgEntry->status == PROTON_UNSUBMITTED) { iNeedSubmission++; // we need to retry check later bDone = 0; } else if ( pMsgEntry->status == PROTON_SUBMITTED) { // we need to retry check later bDone = 0; } } } if (iNeedSubmission > 0) { if ( pWrkrData->bIsConnected == 1) { int credits = pn_link_credit(pWrkrData->pnSender); if (pn_link_credit(pWrkrData->pnSender) > 0) { DBGPRINTF("omazureeventhubs[%p]: trigger pn_connection_wake\n", pWrkrData); pn_connection_wake(pWrkrData->pnConn); } else { DBGPRINTF("omazureeventhubs[%p]: warning pn_link_credit returned %d\n", pWrkrData, credits); } } else { DBGPRINTF("omazureeventhubs[%p]: commitTransaction Suspended=%s Connecting=%s\n", pWrkrData, pWrkrData->bIsSuspended == 1 ? "YES" : "NO", pWrkrData->bIsConnecting == 1 ? "YES" : "NO"); if (pWrkrData->bIsSuspended == 1 && pWrkrData->bIsConnecting == 0) { ABORT_FINALIZE(RS_RET_SUSPENDED); } } } } while (bDone == 0); finalize_it: // Free Proton Message Helpers if (pWrkrData->aProtonMsgs != NULL) { for(i = 0 ; i < nParams ; ++i) { if (pWrkrData->aProtonMsgs[i] != NULL) { // Destroy protonmsg_entry_destruct(pWrkrData->aProtonMsgs[i]); pWrkrData->aProtonMsgs[i] = NULL; } } } /* TODO: Suspend Action if broker problems were reported in error callback */ if (pWrkrData->bIsSuspended == 1) { DBGPRINTF("omazureeventhubs[%p]: commitTransaction failed to send messages, suspending action\n", pWrkrData); iRet = RS_RET_SUSPENDED; } if(iRet != RS_RET_OK) { DBGPRINTF("omazureeventhubs[%p]: commitTransaction failed with status %d\n", pWrkrData, iRet); } DBGPRINTF("omazureeventhubs[%p]: commitTransaction [%d msgs] EXIT\n", pWrkrData, nParams); ENDcommitTransaction static void setInstParamDefaults(instanceData *pData) { DBGPRINTF("setInstParamDefaults[%p]: ENTER\n", pData); pData->amqp_address = NULL; pData->azurehost = NULL; pData->azureport = NULL; pData->azure_key_name = NULL; pData->azure_key = NULL; pData->container = NULL; pData->nEventProperties = 0; pData->eventProperties = NULL; } static rsRetVal processEventProperty(char *const param, const char **const name, const char **const paramval) { DEFiRet; char *val = strstr(param, "="); if(val == NULL) { LogError(0, RS_RET_PARAM_ERROR, "missing equal sign in " "parameter '%s'", param); ABORT_FINALIZE(RS_RET_PARAM_ERROR); } *val = '\0'; /* terminates name */ ++val; /* now points to begin of value */ CHKmalloc(*name = strdup(param)); CHKmalloc(*paramval = strdup(val)); finalize_it: RETiRet; } BEGINnewActInst struct cnfparamvals *pvals; int i; int iNumTpls; DBGPRINTF("newActInst: ENTER\n"); CODESTARTnewActInst if((pvals = nvlstGetParams(lst, &actpblk, NULL)) == NULL) { ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); } CHKiRet(createInstance(&pData)); setInstParamDefaults(pData); for(i = 0 ; i < actpblk.nParams ; ++i) { if(!pvals[i].bUsed) continue; if(!strcmp(actpblk.descr[i].name, "amqp_address")) { pData->amqp_address = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "azurehost")) { pData->azurehost = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "azureport")) { pData->azureport = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "azure_key_name")) { pData->azure_key_name = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "azure_key")) { pData->azure_key = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "container")) { pData->container = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "eventproperties")) { pData->nEventProperties = pvals[i].val.d.ar->nmemb; CHKmalloc(pData->eventProperties = malloc(sizeof(struct event_property) * pvals[i].val.d.ar->nmemb )); for(int j = 0 ; j < pvals[i].val.d.ar->nmemb ; ++j) { char *cstr = es_str2cstr(pvals[i].val.d.ar->arr[j], NULL); CHKiRet(processEventProperty(cstr, &pData->eventProperties[j].key, &pData->eventProperties[j].val)); free(cstr); } } else if(!strcmp(actpblk.descr[i].name, "template")) { pData->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "statsname")) { pData->statsName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else { LogError(0, RS_RET_INTERNAL_ERROR, "omazureeventhubs: program error, non-handled param '%s'\n", actpblk.descr[i].name); } } if(pData->azure_key_name == NULL || pData->azure_key == NULL) { LogError(0, RS_RET_CONFIG_ERROR, "omazureeventhubs: azure_key_name and azure_key are requires to access azure eventhubs" " - action definition invalid"); ABORT_FINALIZE(RS_RET_CONFIG_ERROR); } if(pData->container == NULL) { LogError(0, RS_RET_CONFIG_ERROR, "omazureeventhubs: Event Hubs \"container\" parameter (which is instance) not specified " " - action definition invalid"); ABORT_FINALIZE(RS_RET_CONFIG_ERROR); } if(pData->amqp_address == NULL) { if(pData->azurehost == NULL) { LogMsg(0, NO_ERRCODE, LOG_INFO, "omazureeventhubs: \"azurehost\" parameter not specified " "(youreventhubinstance.servicebus.windows.net- action definition invalid!"); ABORT_FINALIZE(RS_RET_CONFIG_ERROR); } if(pData->azureport== NULL) { // Set default CHKmalloc(pData->azureport = (uchar *) strdup("5671")); } // Create amqps URL from parameters char szAddress[1024]; sprintf(szAddress, "amqps://%s:%s@%s:%s/%s", pData->azure_key_name, pData->azure_key, pData->azurehost, pData->azureport, pData->container); CHKmalloc(pData->amqp_address = (uchar*) strdup(szAddress)); } iNumTpls = 1; CODE_STD_STRING_REQUESTnewActInst(iNumTpls); CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)strdup((pData->tplName == NULL) ? "RSYSLOG_FileFormat" : (char*)pData->tplName), OMSR_NO_RQD_TPL_OPTS)); if (pData->statsName) { CHKiRet(statsobj.Construct(&pData->stats)); CHKiRet(statsobj.SetName(pData->stats, (uchar *)pData->statsName)); CHKiRet(statsobj.SetOrigin(pData->stats, (uchar *)"omazureeventhubs")); /* Track following stats */ STATSCOUNTER_INIT(pData->ctrMessageSubmit, pData->mutCtrMessageSubmit); CHKiRet(statsobj.AddCounter(pData->stats, (uchar *)"submitted", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pData->ctrMessageSubmit)); STATSCOUNTER_INIT(pData->ctrAzureFail, pData->mutCtrAzureFail); CHKiRet(statsobj.AddCounter(pData->stats, (uchar *)"failures", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pData->ctrAzureFail)); STATSCOUNTER_INIT(pData->ctrAzureAck, pData->mutCtrAzureAck); CHKiRet(statsobj.AddCounter(pData->stats, (uchar *)"accepted", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pData->ctrAzureAck)); STATSCOUNTER_INIT(pData->ctrAzureOtherErrors, pData->mutCtrAzureOtherErrors); CHKiRet(statsobj.AddCounter(pData->stats, (uchar *)"othererrors", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pData->ctrAzureOtherErrors)); CHKiRet(statsobj.ConstructFinalize(pData->stats)); } CODE_STD_FINALIZERnewActInst cnfparamvalsDestruct(pvals, &actpblk); ENDnewActInst BEGINmodExit CODESTARTmodExit DBGPRINTF("modExit: ENTER\n"); statsobj.Destruct(&azureStats); CHKiRet(objRelease(statsobj, CORE_COMPONENT)); DESTROY_ATOMIC_HELPER_MUT(mutClock); objRelease(glbl, CORE_COMPONENT); finalize_it: ENDmodExit NO_LEGACY_CONF_parseSelectorAct BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMODTX_QUERIES CODEqueryEtryPt_STD_OMOD8_QUERIES CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES ENDqueryEtryPt BEGINmodInit() CODESTARTmodInit // uchar *pTmp; DBGPRINTF("modInit: ENTER\n"); INITLegCnfVars *ipIFVersProvided = CURR_MOD_IF_VERSION; CODEmodInit_QueryRegCFSLineHdlr /* request objects we use */ CHKiRet(objUse(glbl, CORE_COMPONENT)); CHKiRet(objUse(datetime, CORE_COMPONENT)); CHKiRet(objUse(strm, CORE_COMPONENT)); CHKiRet(objUse(statsobj, CORE_COMPONENT)); // INIT_ATOMIC_HELPER_MUT(mutClock); DBGPRINTF("omazureeventhubs %s using qpid-proton library %d.%d.%d\n", VERSION, PN_VERSION_MAJOR, PN_VERSION_MINOR, PN_VERSION_POINT); CHKiRet(statsobj.Construct(&azureStats)); CHKiRet(statsobj.SetName(azureStats, (uchar *)"omazureeventhubs")); CHKiRet(statsobj.SetOrigin(azureStats, (uchar*)"omazureeventhubs")); STATSCOUNTER_INIT(ctrMessageSubmit, mutCtrMessageSubmit); CHKiRet(statsobj.AddCounter(azureStats, (uchar *)"submitted", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &ctrMessageSubmit)); STATSCOUNTER_INIT(ctrAzureFail, mutCtrAzureFail); CHKiRet(statsobj.AddCounter(azureStats, (uchar *)"failures", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &ctrAzureFail)); STATSCOUNTER_INIT(ctrAzureAck, mutCtrAzureAck); CHKiRet(statsobj.AddCounter(azureStats, (uchar *)"accepted", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &ctrAzureAck)); STATSCOUNTER_INIT(ctrAzureOtherErrors, mutCtrAzureOtherErrors); CHKiRet(statsobj.AddCounter(azureStats, (uchar *)"failures_other", ctrType_IntCtr, CTR_FLAG_RESETTABLE, &ctrAzureOtherErrors)); CHKiRet(statsobj.ConstructFinalize(azureStats)); ENDmodInit pn_timestamp_t time_now(void) { struct timeval now; if (gettimeofday(&now, NULL)) return 0; return ((pn_timestamp_t)now.tv_sec) * 1000 + (now.tv_usec / 1000); } /* * Start PROTON Handling Thread */ static rsRetVal proton_run_thread(wrkrInstanceData_t *pWrkrData) { DEFiRet; int iErr = 0; if ( !pWrkrData->bThreadRunning) { DBGPRINTF("omazureeventhubs[%p]: proton_run_thread\n", pWrkrData); do { iErr = pthread_create(&pWrkrData->tid, NULL, proton_thread, pWrkrData); if (!iErr) { pWrkrData->bThreadRunning = 1; DBGPRINTF("omazureeventhubs[%p]: proton_run_thread (tid %lx) created\n", pWrkrData, pWrkrData->tid); FINALIZE; } } while (iErr == EAGAIN); } else { DBGPRINTF("omazureeventhubs[%p]: proton_run_thread (tid %ld) already running\n", pWrkrData, pWrkrData->tid); } finalize_it: if (iRet != RS_RET_OK) { LogError(0, RS_RET_SYS_ERR, "omazureeventhubs: proton_run_thread thread create failed with error: %d", iErr); } RETiRet; } /* Stop PROTON Handling Thread */ static rsRetVal proton_shutdown_thread(wrkrInstanceData_t *pWrkrData) { DEFiRet; if ( pWrkrData->bThreadRunning) { DBGPRINTF("omazureeventhubs[%p]: STOPPING Thread\n", pWrkrData); int r = pthread_cancel(pWrkrData->tid); if(r == 0) { pthread_join(pWrkrData->tid, NULL); } DBGPRINTF("omazureeventhubs[%p]: STOPPED Thread\n", pWrkrData); pWrkrData->bThreadRunning = 0; } FINALIZE; finalize_it: RETiRet; } /* * Workerthread function for a single ProActor Handler */ static void * proton_thread(void __attribute__((unused)) *pVoidWrkrData) { wrkrInstanceData_t *const pWrkrData = (wrkrInstanceData_t *const) pVoidWrkrData; instanceData *const pData = (instanceData *const) pWrkrData->pData; DBGPRINTF("proton_thread[%p]: started protocol workerthread(%lx) for %s:%s/%s\n", pWrkrData, pthread_self(), pData->azurehost, pData->azureport, pData->container); do { if ( pWrkrData->pnProactor != NULL) { // Process Protocol events pn_event_batch_t *events = pn_proactor_wait(pWrkrData->pnProactor); pn_event_t *event; while ((event = pn_event_batch_next(events))) { handleProton(pWrkrData, event); } pn_proactor_done(pWrkrData->pnProactor, events); } else { // Wait 10 microseconds srSleep(0, 10000); } } while(glbl.GetGlobalInputTermState() == 0); DBGPRINTF("proton_thread[%p]: stopped protocol workerthread\n", pWrkrData); return NULL; } static void handleProtonDelivery(wrkrInstanceData_t *const pWrkrData) { instanceData *const pData = (instanceData *const) pWrkrData->pData; /* Process messages from ARRAY */ for(unsigned int i = 0 ; i < pWrkrData->nProtonMsgs ; ++i) { protonmsg_entry* pMsgEntry = (protonmsg_entry*) pWrkrData->aProtonMsgs[i]; // Process Unsubmitted messages only if (pMsgEntry != NULL) { if (pMsgEntry->status == PROTON_UNSUBMITTED) { int iCreditBalance = pn_link_credit(pWrkrData->pnSender); if (iCreditBalance > 0) { DBGPRINTF( "handleProtonDelivery: PN_LINK_FLOW deliver '%s' @ %p:%s:%s/%s, msg:'%.*s'\n", pMsgEntry->MsgID, pWrkrData, pData->azurehost, pData->azureport, pData->container, (pMsgEntry->payload_len > 0 ? (int)pMsgEntry->payload_len-1 : 0), pMsgEntry->payload); /* Use sent counter as unique delivery tag. */ pn_delivery(pWrkrData->pnSender, pn_dtag((const char *)pMsgEntry->MsgID, pMsgEntry->MsgID_len)); /* Construct a message with the map */ pn_message_t* message = proton_encode_message(pWrkrData, pMsgEntry); /* * pn_message_send does performs the following steps: * - call pn_message_encode2() to encode the message to a buffer * - call pn_link_send() to send the encoded message bytes * - call pn_link_advance() to indicate the message is complete */ if (pn_message_send( message, pWrkrData->pnSender, &pWrkrData->pnMessageBuffer) < 0) { LogMsg(0, NO_ERRCODE, LOG_INFO, "handleProtonDelivery: PN_LINK_FLOW deliver SEND ERROR %s\n", pn_error_text(pn_message_error(message))); pn_message_free(message); break; } else { DBGPRINTF("handleProtonDelivery: PN_LINK_FLOW deliver SUCCESS\n"); pn_message_free(message); } STATSCOUNTER_INC(ctrMessageSubmit, mutCtrMessageSubmit); INST_STATSCOUNTER_INC( pData, pData->ctrMessageSubmit, pData->mutCtrMessageSubmit); pMsgEntry->status = PROTON_SUBMITTED; } else { DBGPRINTF("handleProtonDelivery: sender credit balance reached %d. " "extend credit for %d\n", iCreditBalance, pWrkrData->nProtonMsgs); pn_link_flow(pWrkrData->pnSender, pWrkrData->nProtonMsgs); // TODO: MAKE CONFIGUREABLE // Wait 10 microseconds // srSleep(0, 10000); break; } } } else { // Abort further processing if pMsgEntry is NULL break; } } } /* Handles PROTON Communication in this Function * */ #pragma GCC diagnostic ignored "-Wswitch" static void handleProton(wrkrInstanceData_t *const pWrkrData, pn_event_t *event) { instanceData *const pData = (instanceData *const) pWrkrData->pData; /* Handle Proton Events */ switch (pn_event_type(event)) { case PN_CONNECTION_BOUND: { DBGPRINTF("handleProton: PN_CONNECTION_BOUND to %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); break; } case PN_SESSION_INIT : { DBGPRINTF("handleProton: PN_SESSION_INIT to %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); break; } case PN_LINK_INIT: { DBGPRINTF("handleProton: PN_LINK_INIT to %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); break; } case PN_LINK_REMOTE_OPEN: { DBGPRINTF("handleProton: PN_LINK_REMOTE_OPEN to %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); pWrkrData->bIsConnected = 1; pWrkrData->bIsConnecting = 0; break; } case PN_CONNECTION_INIT: { DBGPRINTF("handleProton: PN_CONNECTION_INIT to %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); pWrkrData->pnStatus = PN_CONNECTION_INIT; // Get Connection pWrkrData->pnConn = pn_event_connection(event); // Set AMQP Properties pn_connection_set_container(pWrkrData->pnConn, (const char *) pData->container); pn_connection_set_hostname(pWrkrData->pnConn, (const char *) pData->azurehost); pn_connection_set_user(pWrkrData->pnConn, (const char *) pData->azure_key_name); pn_connection_set_password(pWrkrData->pnConn, (const char *) pData->azure_key); pn_connection_open(pWrkrData->pnConn); // Open Connection pn_session_t* pnSession = pn_session(pWrkrData->pnConn); // Create Session pn_session_open(pnSession); // Open Session pWrkrData->pnSender = pn_sender(pnSession, (char *)pData->azure_key_name); // Create Link DBGPRINTF("handleProton: PN_CONNECTION_INIT with amqp address: %s\n", pData->amqp_address); pn_terminus_set_address(pn_link_target(pWrkrData->pnSender), (const char *) pData->amqp_address); pn_link_open(pWrkrData->pnSender); pn_link_flow(pWrkrData->pnSender, pWrkrData->nProtonMsgs); break; } case PN_CONNECTION_REMOTE_OPEN: { DBGPRINTF("handleProton: PN_CONNECTION_REMOTE_OPEN to %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); pWrkrData->pnStatus = PN_CONNECTION_REMOTE_OPEN; pn_ssl_t *ssl = pn_ssl(pn_event_transport(event)); if (ssl) { char name[1024]; pn_ssl_get_protocol_name(ssl, name, sizeof(name)); { const char *subject = pn_ssl_get_remote_subject(ssl); if (subject) { DBGPRINTF( "handleProton: handleProton secure connection: to %s using %s\n", subject, name); } else { DBGPRINTF( "handleProton: handleProton anonymous connection: using %s\n", name); } fflush(stdout); } } break; } case PN_CONNECTION_WAKE: { DBGPRINTF("handleProton: PN_CONNECTION_WAKE (%d) to %p:%s:%s/%s\n", pWrkrData->nProtonMsgs, pWrkrData, pData->azurehost, pData->azureport, pData->container); /* Process messages */ handleProtonDelivery(pWrkrData); break; } case PN_LINK_FLOW: { pWrkrData->pnStatus = PN_LINK_FLOW; /* The peer has given us some credit, now we can send messages */ pWrkrData->pnSender = pn_event_link(event); /* Process messages */ handleProtonDelivery(pWrkrData); break; } case PN_DELIVERY: { pWrkrData->pnStatus = PN_DELIVERY; /* We received acknowledgement from the peer that a message was delivered. */ pn_delivery_t* pDeliveryStatus = pn_event_delivery(event); pn_delivery_tag_t pnTag = pn_delivery_tag(pDeliveryStatus); if (pnTag.start != NULL) { // Convert Tag into Number! unsigned int iTagNum = (unsigned int) atoi((pnTag.start != NULL ? pnTag.start : "")); // Calc QueueNumber from Tagnum unsigned int iQueueNum = pWrkrData->nProtonMsgs - (pWrkrData->iMaxMsgSeq - iTagNum); // Get proton Msg Helper (checks for out of bound array access) protonmsg_entry* pMsgEntry = NULL; if (pWrkrData->nMaxProtonMsgs > iQueueNum) { pMsgEntry = (protonmsg_entry*) pWrkrData->aProtonMsgs[iQueueNum]; } // Process if found if (pMsgEntry != NULL) { if (pn_delivery_remote_state(pDeliveryStatus) == PN_ACCEPTED) { DBGPRINTF( "handleProton: PN_DELIVERY SUCCESS for MSG '%s(Q %d, MAX %d)' @ %p:%s:%s/%s\n", (pnTag.start != NULL ? (char*) pnTag.start : "NULL"), iQueueNum, pWrkrData->nMaxProtonMsgs, pWrkrData, pData->azurehost, pData->azureport, pData->container); pMsgEntry->status = PROTON_ACCEPTED; // Increment Stats Counter STATSCOUNTER_INC(ctrAzureAck, mutCtrAzureAck); INST_STATSCOUNTER_INC(pData, pData->ctrAzureAck, pData->mutCtrAzureAck); } else if (pn_delivery_remote_state(pDeliveryStatus) == PN_REJECTED) { LogError(0, RS_RET_ERR, "omazureeventhubs: PN_DELIVERY REJECTED for MSG '%s'" " - @ %p:%s:%s/%s\n", (pnTag.start != NULL ? (char*) pnTag.start : "NULL"), pWrkrData, pData->azurehost, pData->azureport, pData->container); pMsgEntry->status = PROTON_REJECTED; // Increment Stats Counter STATSCOUNTER_INC(ctrAzureFail, mutCtrAzureFail); INST_STATSCOUNTER_INC(pData, pData->ctrAzureFail, pData->mutCtrAzureFail); } } else { DBGPRINTF("handleProton: PN_DELIVERY MISSING MSG '%d(Q %d,MAX %d)' - @ %p:%s:%s/%s\n", iTagNum, iQueueNum, pWrkrData->nMaxProtonMsgs, pWrkrData, pData->azurehost, pData->azureport, pData->container); STATSCOUNTER_INC(ctrAzureOtherErrors, mutCtrAzureOtherErrors); INST_STATSCOUNTER_INC(pData, pData->ctrAzureOtherErrors, pData->mutCtrAzureOtherErrors); } } else { LogError(0, RS_RET_ERR,"handleProton: PN_DELIVERY HELPER ARRAY is NULL - @ %p:%s:%s/%s\n", pWrkrData, pData->azurehost, pData->azureport, pData->container); STATSCOUNTER_INC(ctrAzureOtherErrors, mutCtrAzureOtherErrors); INST_STATSCOUNTER_INC(pData, pData->ctrAzureOtherErrors, pData->mutCtrAzureOtherErrors); } break; } case PN_TRANSPORT_CLOSED: DBGPRINTF("handleProton: transport closed for %p:%s\n", pWrkrData, pData->azurehost); proton_check_condition(event, pWrkrData, pn_transport_condition(pn_event_transport(event)), "transport closed"); break; case PN_CONNECTION_REMOTE_CLOSE: DBGPRINTF("handleProton: connection closed for %p:%s\n", pWrkrData, pData->azurehost); proton_check_condition(event, pWrkrData, pn_connection_remote_condition(pn_event_connection(event)), "connection closed"); break; case PN_SESSION_REMOTE_CLOSE: DBGPRINTF("handleProton: remote session closed for %p:%s\n", pWrkrData, pData->azurehost); proton_check_condition(event, pWrkrData, pn_session_remote_condition(pn_event_session(event)), "remote session closed"); break; case PN_LINK_REMOTE_CLOSE: case PN_LINK_REMOTE_DETACH: DBGPRINTF("handleProton: remote link closed for %p:%s\n", pWrkrData, pData->azurehost); proton_check_condition(event, pWrkrData, pn_link_remote_condition(pn_event_link(event)), "remote link closed"); break; case PN_PROACTOR_INACTIVE: DBGPRINTF("handleProton: INAKTIVE for %p:%s\n", pWrkrData, pData->azurehost); break; /* Workarround compiler warning: * error: enumeration value '...' not handled in switch [-Werror=switch-enum] */ #ifdef __GNU_C default: DBGPRINTF("handleProton: UNHANDELED EVENT %d for %p:%s\n", pn_event_type(event), pWrkrData, pData->azurehost); break; #endif } }