summaryrefslogtreecommitdiffstats
path: root/plugins/omazureeventhubs/omazureeventhubs.c
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--plugins/omazureeventhubs/omazureeventhubs.c1354
1 files changed, 1354 insertions, 0 deletions
diff --git a/plugins/omazureeventhubs/omazureeventhubs.c b/plugins/omazureeventhubs/omazureeventhubs.c
new file mode 100644
index 0000000..a2bd8b3
--- /dev/null
+++ b/plugins/omazureeventhubs/omazureeventhubs.c
@@ -0,0 +1,1354 @@
+/* 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 <stdio.h>
+#include <stdarg.h>
+#include <stdlib.h>
+#include <string.h>
+#include <strings.h>
+#include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <pthread.h>
+#include <sys/uio.h>
+#include <sys/queue.h>
+#include <sys/types.h>
+#include <math.h>
+#ifdef HAVE_SYS_STAT_H
+# include <sys/stat.h>
+#endif
+#include <sys/time.h>
+#include <time.h>
+
+// Include Proton headers
+#include <proton/version.h>
+#include <proton/condition.h>
+#include <proton/connection.h>
+#include <proton/delivery.h>
+#include <proton/link.h>
+#include <proton/listener.h>
+#include <proton/netaddr.h>
+#include <proton/message.h>
+#include <proton/object.h>
+#include <proton/proactor.h>
+#include <proton/sasl.h>
+#include <proton/session.h>
+#include <proton/transport.h>
+#include <proton/ssl.h>
+
+// 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
+ }
+}