/*************************************************************************//**
*
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
*
* Unless otherwise specified, all software contained herein is
* Licensed under the Apache License, Version 2.0 (the "License");
* How long we're prepared to wait for the API service to respond in
* seconds.
*****************************************************************************/
-static const int EVEL_API_TIMEOUT = 5;
+static const int EVEL_API_TIMEOUT = 10;
/*****************************************************************************/
/* Prototypes of locally scoped functions. */
static bool evel_token_equals_string(const MEMORY_CHUNK * const chunk,
const jsmntok_t * const json_token,
const char * check_string);
+static void * event_multi_handler(void * arg __attribute__ ((unused)));
/**************************************************************************//**
* Buffers for error strings from libcurl.
*****************************************************************************/
static char curl_err_string[CURL_ERROR_SIZE] = "<NULL>";
+static char curl_err_string2[CURL_ERROR_SIZE] = "<NULL>";
/**************************************************************************//**
* Handle for the API into libcurl.
*****************************************************************************/
static CURL * curl_handle = NULL;
+static CURL * curl_handle2 = NULL;
+static CURLM * multi_handle = NULL;
+int curr_global_handles = 0;
+int activmode = -1;
/**************************************************************************//**
* Special headers that we send.
*****************************************************************************/
static struct curl_slist * hdr_chunk = NULL;
+static struct curl_slist * hdr_chunk2 = NULL;
/**************************************************************************//**
* Message queue for sending events to the API.
static char * evel_throt_api_url;
static char * evel_batch_api_url;
+static char * evel_bevent_api_url;
+static char * evel_bthrot_api_url;
+static char * evel_bbatch_api_url;
+
+
/**************************************************************************//**
* Initialize the event handler.
*
* logs.
*****************************************************************************/
EVEL_ERR_CODES event_handler_initialize(const char * const event_api_url,
+ const char * const bakup_api_url,
const char * const throt_api_url,
const char * const source_ip,
+ const char * const source_ip_bakup,
int ring_buf_size,
int secure,
+ int activitymode,
const char * const cert_file_path,
const char * const key_file_path,
const char * const ca_info,
long verify_host,
const char * const username,
const char * const password,
+ const char * const username2,
+ const char * const password2,
int verbosity)
{
int rc = EVEL_SUCCESS;
assert(throt_api_url != NULL);
assert(username != NULL);
assert(password != NULL);
+ if( bakup_api_url != NULL )
+ {
+ assert(username2 != NULL);
+ assert(password2 != NULL);
+ }
+
/***************************************************************************/
/* Store the API URLs. */
assert(evel_batch_api_url != NULL);
evel_throt_api_url = strdup(throt_api_url);
assert(evel_throt_api_url != NULL);
+ curr_global_handles = 1;
+
+ if( bakup_api_url != NULL )
+ {
+ evel_bevent_api_url = strdup(bakup_api_url);
+ assert(evel_bevent_api_url != NULL);
+ sprintf(batch_api_url,"%s/eventBatch",event_api_url);
+ evel_bbatch_api_url = strdup(batch_api_url);
+ assert(evel_bbatch_api_url != NULL);
+ evel_bthrot_api_url = strdup(throt_api_url);
+ assert(evel_bthrot_api_url != NULL);
+ curr_global_handles = 2;
+ }
curl_version_info_data *d = curl_version_info(CURLVERSION_NOW);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_handle2 = curl_easy_init();
+ if (curl_handle2 == NULL)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to get backup libCURL handle");
+ goto exit_label;
+ }
+ }
/***************************************************************************/
/* Prime the library to give friendly error codes. */
/***************************************************************************/
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_ERRORBUFFER,
+ curl_err_string2);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL2 to provide friendly errors. "
+ "Error code=%d", curl_rc);
+ goto exit_label;
+ }
+ }
+
+
/***************************************************************************/
/* If running in verbose mode generate more output. */
/***************************************************************************/
"Error code=%d", curl_rc);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2, CURLOPT_VERBOSE, 1L);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL to be verbose. "
+ "Error code=%d", curl_rc);
+ goto exit_label;
+ }
+ }
+
}
+
+
/***************************************************************************/
/* Set the URL for the API. */
/***************************************************************************/
}
EVEL_INFO("Initializing CURL to send events to: %s", event_api_url);
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_URL,
+ bakup_api_url);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
+
/***************************************************************************/
/* send all data to this function. */
/***************************************************************************/
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_WRITEFUNCTION,
+ evel_write_callback);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
+
/***************************************************************************/
/* configure local ip address if provided */
}
}
}
+ if( source_ip_bakup != NULL )
+ {
+ snprintf(local_address,sizeof(local_address),source_ip_bakup);
+ if( local_address[0] != '\0' )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_INTERFACE,
+ local_address);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize bakup libCURL with the local address. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+ }
/***************************************************************************/
/* configure SSL options for HTTPS transfers */
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_SSLCERT,
+ cert_file_path);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the client cert. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ }
}
if( key_file_path != NULL )
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_SSLKEY,
+ key_file_path);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the client key. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ }
}
if( ca_info != NULL )
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_CAINFO,
+ ca_info);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the CA cert file. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ }
}
if( ca_file_path != NULL )
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_CAPATH,
+ ca_file_path);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the CA cert path. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ }
}
curl_rc = curl_easy_setopt(curl_handle,
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_SSL_VERIFYPEER,
+ verify_peer);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with SSL Server verification. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_SSL_VERIFYHOST,
+ verify_host);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with Client host verification. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ }
}
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,
+ CURLOPT_USERAGENT,
+ "libcurl-agent/1.0");
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
/***************************************************************************/
/* Specify that we are going to POST data. */
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,CURLOPT_POST, 1L);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
/***************************************************************************/
/* we want to use our own read function. */
"function. Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,CURLOPT_READFUNCTION, read_callback);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
/***************************************************************************/
/* All of our events are JSON encoded. We also suppress the */
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ hdr_chunk2 = curl_slist_append(hdr_chunk2, "Content-type: application/json");
+ hdr_chunk2 = curl_slist_append(hdr_chunk2, "Expect:");
+ curl_rc = curl_easy_setopt(curl_handle2,CURLOPT_HTTPHEADER, hdr_chunk2);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
+
/***************************************************************************/
/* Set the timeout for the operation. */
"Error code=%d (%s)", curl_rc, curl_err_string);
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2,CURLOPT_TIMEOUT, EVEL_API_TIMEOUT);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ }
+
/***************************************************************************/
/* Set that we want Basic authentication with username:password Base-64 */
goto exit_label;
}
+ if( bakup_api_url != NULL )
+ {
+ curl_rc = curl_easy_setopt(curl_handle2, CURLOPT_HTTPAUTH, CURLAUTH_BASIC);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL for Basic Authentication. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ curl_rc = curl_easy_setopt(curl_handle2, CURLOPT_USERNAME, username2);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with username. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ curl_rc = curl_easy_setopt(curl_handle2, CURLOPT_PASSWORD, password2);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with password. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+
+ multi_handle = curl_multi_init();;
+ if (multi_handle == NULL)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to get libCURL Multi handle");
+ goto exit_label;
+ }
+ activmode = activitymode;
+
+ }
+
+
/***************************************************************************/
/* Initialize a message ring-buffer to be used between the foreground and */
/* the thread which sends the messages. This can't fail. */
/* Start the event handler thread. */
/***************************************************************************/
evt_handler_state = EVT_HANDLER_INACTIVE;
- pthread_rc = pthread_create(&evt_handler_thread, NULL, event_handler, NULL);
+ if( curr_global_handles <= 1 )
+ pthread_rc = pthread_create(&evt_handler_thread, NULL, event_handler, NULL);
+ else
+ pthread_rc = pthread_create(&evt_handler_thread, NULL, event_multi_handler, NULL);
+
if (pthread_rc != 0)
{
rc = EVEL_PTHREAD_LIBRARY_FAIL;
/***************************************************************************/
/* Clean-up the cURL library. */
/***************************************************************************/
+ if (multi_handle != NULL)
+ {
+ curl_multi_cleanup(multi_handle);
+ }
if (curl_handle != NULL)
{
curl_easy_cleanup(curl_handle);
curl_handle = NULL;
}
+ if (curl_handle2 != NULL)
+ {
+ curl_easy_cleanup(curl_handle2);
+ curl_handle2 = NULL;
+ }
if (hdr_chunk != NULL)
{
curl_slist_free_all(hdr_chunk);
hdr_chunk = NULL;
}
+ if (hdr_chunk2 != NULL)
+ {
+ curl_slist_free_all(hdr_chunk2);
+ hdr_chunk2 = NULL;
+ }
+
/***************************************************************************/
/* Free off the stored API URL strings. */
return realsize;
}
+
/**************************************************************************//**
- * Event Handler.
- *
- * Watch for messages coming on the internal queue and send them to the
- * listener.
+ * Post an event to the Vendor Event Listener API.
*
- * param[in] arg Argument - unused.
+ * @returns Status code
+ * @retval EVEL_SUCCESS On success
+ * @retval "One of ::EVEL_ERR_CODES" On failure.
*****************************************************************************/
-static void * event_handler(void * arg __attribute__ ((unused)))
+static EVEL_ERR_CODES evel_postmulti_message(char *msg, size_t size, int *still,
+ CURL *handle, CURL *bhandle, int numhandles )
{
- int old_type = 0;
- EVENT_HEADER * msg = NULL;
- EVENT_INTERNAL * internal_msg = NULL;
- int json_size = 0;
- char json_body[EVEL_MAX_JSON_BODY];
int rc = EVEL_SUCCESS;
- CURLcode curl_rc;
+ CURLcode curl_rc = CURLE_OK;
+ MEMORY_CHUNK rx_chunk[2];
+ MEMORY_CHUNK tx_chunk[2];
+ int http_response_code = 0, i;
- EVEL_INFO("Event handler thread started");
+ EVEL_ENTER();
+
+ EVEL_INFO("Sending :%s: %d\n",msg, numhandles);
/***************************************************************************/
- /* Set this thread to be cancellable immediately. */
+ /* Create the memory chunk to be used for the response to the post. The */
+ /* will be realloced. */
/***************************************************************************/
- pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, &old_type);
+for (i=0;i<numhandles;i++)
+{
+ rx_chunk[i].memory = malloc(1);
+ assert(rx_chunk[i].memory != NULL);
+ rx_chunk[i].size = 0;
/***************************************************************************/
- /* Set the handler as active, defending against weird situations like */
- /* immediately shutting down after initializing the library so the */
- /* handler never gets started up properly. */
+ /* Create the memory chunk to be sent as the body of the post. */
/***************************************************************************/
- if (evt_handler_state == EVT_HANDLER_INACTIVE)
- {
- evt_handler_state = EVT_HANDLER_ACTIVE;
- }
- else
+ tx_chunk[i].memory = msg;
+ tx_chunk[i].size = size;
+ EVEL_DEBUG("Sending chunk of size %d", tx_chunk[i].size);
+}
+
+ /***************************************************************************/
+ /* Point to the data to be received. */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(handle, CURLOPT_WRITEDATA, &rx_chunk[0]);
+ if (curl_rc != CURLE_OK)
{
- EVEL_ERROR("Event Handler State was not INACTIVE at start-up - "
- "Handler will exit immediately!");
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL rx to upload. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
}
- while (evt_handler_state == EVT_HANDLER_ACTIVE)
+ /***************************************************************************/
+ /* Pointer to pass to our read function */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(handle, CURLOPT_READDATA, &tx_chunk[0]);
+ if (curl_rc != CURLE_OK)
{
- /*************************************************************************/
- /* Wait for a message to be received. */
- /*************************************************************************/
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to set upload data for libCURL tx to upload. "
+ "Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+ /***************************************************************************/
+ /* Size of the data to transmit. */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(handle,
+ CURLOPT_POSTFIELDSIZE,
+ tx_chunk[0].size);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to set length of upload data for libCURL to "
+ "upload. Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+
+if(numhandles == 2)
+{
+
+ /***************************************************************************/
+ /* Point to the data to be received. */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(bhandle, CURLOPT_WRITEDATA, &rx_chunk[1]);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL2 rx to upload. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+
+ /***************************************************************************/
+ /* Pointer to pass to our read function */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(bhandle, CURLOPT_READDATA, &tx_chunk[1]);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to set upload data for libCURL2 tx to upload. "
+ "Error code=%d (%s)", curl_rc, curl_err_string2);
+ goto exit_label;
+ }
+ /***************************************************************************/
+ /* Size of the data to transmit. */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(bhandle,
+ CURLOPT_POSTFIELDSIZE,
+ tx_chunk[1].size);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to set length of upload data for libCURL2 to "
+ "upload. Error code=%d (%s)", curl_rc, curl_err_string);
+ goto exit_label;
+ }
+
+}
+
+ /***************************************************************************/
+ /* Now run off and do what you've been told! */
+ /***************************************************************************/
+ curl_rc = curl_multi_perform(multi_handle, still);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to transfer an multi event to Vendor Event Listener! "
+ "Error code=%d (%s)", curl_rc, curl_multi_strerror(curl_rc
+));
+ EVEL_ERROR("Dropped event: %s", msg);
+ goto exit_label;
+ }
+
+ /***************************************************************************/
+ /* See what response we got - any 2XX response is good. */
+ /***************************************************************************/
+
+exit_label:
+ for (i=0;i<numhandles;i++)
+ free(rx_chunk[i].memory);
+ EVEL_EXIT();
+
+ return(rc);
+}
+
+
+/**************************************************************************//**
+ * Event Multi Post Handler.
+ *
+ * Watch for messages coming on the internal queue and send them to the
+ * listener.
+ *
+ * param[in] jsonmsg json message to be sent.
+ * param[in] size size of json message
+ * param[in] currhandle Primary handle
+ * param[in] url1 pimary url
+ * param[in] bakkhandle Backup handle
+ * param[in] url2 secondary url
+ *****************************************************************************/
+static int evel_post_multiapi(char *jsonmsg, size_t size,CURL *currhandle, char *url1,
+ CURL *bakkhandle, char *url2)
+{
+ int rc = EVEL_SUCCESS;
+ CURLcode curl_rc = 0;
+ int nhandles = 1;
+ int still_running,i;
+ CURLMsg *msg; /* for picking up messages with the transfer status */
+ int msgs_left; /* how many messages are left */
+
+ /***************************************************************************/
+ /* Set the URL for the API. */
+ /***************************************************************************/
+ curl_rc = curl_easy_setopt(currhandle, CURLOPT_URL, url1);
+ if (curl_rc != CURLE_OK )
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the multi API URL. "
+ "%s Error code=%d (%s)", url1, curl_rc, curl_err_string);
+ }
+
+ if( url2 != NULL && activmode == 1 )
+ {
+ curl_rc = curl_easy_setopt(bakkhandle, CURLOPT_URL, url2);
+ if (curl_rc != CURLE_OK)
+ {
+ rc = EVEL_CURL_LIBRARY_FAIL;
+ log_error_state("Failed to initialize libCURL with the API URL. "
+ "%s Error code=%d (%s)", url2, curl_rc, curl_err_string2);
+ }
+ nhandles = 2;
+ }
+
+ /* we start some action by calling perform right away */
+ curl_multi_add_handle(multi_handle, currhandle);
+ if(nhandles==2){
+ curl_multi_add_handle(multi_handle, bakkhandle);
+ }
+
+ /* we start some action by calling perform right away */
+ curl_multi_perform(multi_handle, &still_running);
+
+ do {
+ struct timeval timeout;
+ int rc; /* select() return code */
+ CURLMcode mc; /* curl_multi_fdset() return code */
+
+ fd_set fdread;
+ fd_set fdwrite;
+ fd_set fdexcep;
+ int maxfd = -1;
+
+ long curl_timeo = -1;
+
+ FD_ZERO(&fdread);
+ FD_ZERO(&fdwrite);
+ FD_ZERO(&fdexcep);
+
+ /* set a suitable timeout to play around with */
+ timeout.tv_sec = 1;
+ timeout.tv_usec = 0;
+
+ curl_multi_timeout(multi_handle, &curl_timeo);
+ if(curl_timeo >= 0) {
+ timeout.tv_sec = curl_timeo / 1000;
+ if(timeout.tv_sec > 1)
+ timeout.tv_sec = 1;
+ else
+ timeout.tv_usec = (curl_timeo % 1000) * 1000;
+ }
+
+ /* get file descriptors from the transfers */
+ mc = curl_multi_fdset(multi_handle, &fdread, &fdwrite, &fdexcep, &maxfd);
+
+ if(mc != CURLM_OK) {
+ EVEL_ERROR("curl_multi_fdset() failed, code %d.\n", mc);
+ break;
+ }
+
+ /* On success the value of maxfd is guaranteed to be >= -1. We call
+ select(maxfd + 1, ...); specially in case of (maxfd == -1) there are
+ no fds ready yet so we call select(0, ...) --or Sleep() on Windows--
+ to sleep 100ms, which is the minimum suggested value in the
+ curl_multi_fdset() doc. */
+
+ if(maxfd == -1) {
+#ifdef _WIN32
+ Sleep(100);
+ rc = 0;
+#else
+ /* Portable sleep for platforms other than Windows. */
+ struct timeval wait = { 0, 300000 }; /* 250ms */
+ rc = select(0, NULL, NULL, NULL, &wait);
+#endif
+ }
+ else {
+ /* Note that on some platforms 'timeout' may be modified by select().
+ If you need access to the original value save a copy beforehand. */
+ rc = select(maxfd + 1, &fdread, &fdwrite, &fdexcep, &timeout);
+ }
+
+ switch(rc) {
+ case -1:
+ /* select error */
+ break;
+ case 0: /* timeout */
+ default: /* action */
+ //curl_multi_perform(multi_handle, &still_running);
+ evel_postmulti_message(jsonmsg, size, &still_running, currhandle, bakkhandle, nhandles);
+ }
+ } while(still_running);
+
+ /* See how the transfers went */
+ while((msg = curl_multi_info_read(multi_handle, &msgs_left))) {
+ if(msg->msg == CURLMSG_DONE) {
+ int idx, found = 0;
+ EVEL_DEBUG("Transfer status - %s\n", curl_multi_strerror(msg->data.result));
+
+ /* Find out which handle this message is about */
+ for(idx = 0; idx<nhandles; idx++) {
+ if (msg->easy_handle == currhandle) break;
+ else if(msg->easy_handle == bakkhandle) break;
+ }
+
+ switch(idx) {
+ case 0:
+ curl_rc = msg->data.result;
+ break;
+ case 1:
+ curl_rc = msg->data.result;
+ break;
+ }
+ }
+ }
+
+ /* we start some action by calling perform right away */
+ curl_multi_remove_handle(multi_handle, currhandle);
+ if(nhandles==2){
+ curl_multi_remove_handle(multi_handle, bakkhandle);
+ }
+
+ EVEL_DEBUG("Transfer completed with status %s\n", curl_multi_strerror(curl_rc));
+ if( curl_rc == 0 || curl_rc == 55 )
+ return EVEL_SUCCESS;
+ else
+ return EVEL_CURL_LIBRARY_FAIL;
+
+}
+/**************************************************************************//**
+ * Event Handler.
+ *
+ * Watch for messages coming on the internal queue and send them to the
+ * listener.
+ *
+ * param[in] arg Argument - unused.
+ *****************************************************************************/
+static void * event_multi_handler(void * arg __attribute__ ((unused)))
+{
+ int old_type = 0;
+ EVENT_HEADER * msg = NULL;
+ EVENT_INTERNAL * internal_msg = NULL;
+ int json_size = 0;
+ char json_body[EVEL_MAX_JSON_BODY];
+ int rc = EVEL_SUCCESS;
+ CURLcode curl_rc;
+ CURL *currhandler = NULL;
+ CURL *bakhandler = NULL;
+ char *send_url = NULL;
+ char *send_url2 = NULL;
+
+ EVEL_INFO("Event multi handler thread started");
+
+ /***************************************************************************/
+ /* Set this thread to be cancellable immediately. */
+ /***************************************************************************/
+ pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, &old_type);
+
+ /***************************************************************************/
+ /* Set the handler as active, defending against weird situations like */
+ /* immediately shutting down after initializing the library so the */
+ /* handler never gets started up properly. */
+ /***************************************************************************/
+ if (evt_handler_state == EVT_HANDLER_INACTIVE)
+ {
+ evt_handler_state = EVT_HANDLER_ACTIVE;
+ }
+ else
+ {
+ EVEL_ERROR("Event Handler State was not INACTIVE at start-up - "
+ "Handler will exit immediately!");
+ }
+
+ currhandler = curl_handle;
+ while (evt_handler_state == EVT_HANDLER_ACTIVE)
+ {
+ /*************************************************************************/
+ /* Wait for a message to be received. */
+ /*************************************************************************/
+ msg = ring_buffer_read(&event_buffer);
+
+ /*************************************************************************/
+ /* Internal events get special treatment while regular events get posted */
+ /* to the far side. */
+ /*************************************************************************/
+ if (msg->event_domain == EVEL_DOMAIN_BATCH )
+ {
+ EVEL_DEBUG("Batch event received");
+
+ /***********************************************************************/
+ /* Encode the event in JSON. */
+ /***********************************************************************/
+ json_size = evel_json_encode_batch_event(json_body, EVEL_MAX_JSON_BODY, msg);
+
+ /***************************************************************************/
+ /* Set the URL for the API. */
+ /***************************************************************************/
+ if( currhandler == curl_handle){
+ send_url = evel_batch_api_url;
+ send_url2 = evel_bbatch_api_url;
+ bakhandler = curl_handle2;
+ } else if(currhandler == curl_handle2) {
+ send_url = evel_bbatch_api_url;
+ send_url2 = evel_batch_api_url;
+ bakhandler = curl_handle;
+ }
+
+ /***********************************************************************/
+ /* Send the JSON across the API. */
+ /***********************************************************************/
+ EVEL_DEBUG("Sending Batch JSON of size %d is: %s", json_size, json_body);
+ rc = evel_post_multiapi(json_body, json_size, currhandler, send_url, bakhandler, send_url2);
+ if (rc != EVEL_SUCCESS)
+ {
+ EVEL_ERROR("Failed to transfer the data to %s. Error code=%d", send_url, rc);
+ EVEL_INFO("Switched Collector ...");
+ if( currhandler == curl_handle){
+ currhandler = curl_handle2;
+ bakhandler = curl_handle;
+ } else if(currhandler == curl_handle2) {
+ currhandler = curl_handle;
+ bakhandler = curl_handle2;
+ }
+ rc = evel_post_multiapi(json_body, json_size, currhandler, send_url2, bakhandler,send_url);
+ if (rc != EVEL_SUCCESS)
+ EVEL_ERROR("Failed to transfer the data to failover %s. Error code=%d", send_url2, rc);
+ }
+ }
+ else if (msg->event_domain != EVEL_DOMAIN_INTERNAL )
+ {
+ EVEL_DEBUG("External event received");
+
+ /***********************************************************************/
+ /* Encode the event in JSON. */
+ /***********************************************************************/
+ json_size = evel_json_encode_event(json_body, EVEL_MAX_JSON_BODY, msg);
+
+ /***************************************************************************/
+ /* Set the URL for the API. */
+ /***************************************************************************/
+ if( currhandler == curl_handle){
+ send_url = evel_event_api_url;
+ send_url2 = evel_bevent_api_url;
+ bakhandler = curl_handle2;
+ } else if(currhandler == curl_handle2) {
+ send_url = evel_bevent_api_url;
+ send_url2 = evel_event_api_url;
+ bakhandler = curl_handle;
+ }
+
+ /***********************************************************************/
+ /* Send the JSON across the API. */
+ /***********************************************************************/
+ EVEL_DEBUG("Sending Batch JSON of size %d is: %s", json_size, json_body);
+ rc = evel_post_multiapi(json_body, json_size, currhandler, send_url, bakhandler, send_url2);
+ if (rc != EVEL_SUCCESS)
+ {
+ EVEL_ERROR("Failed to transfer the data to %s. Error code=%d", send_url, rc);
+ EVEL_INFO("Switched Collector ...");
+ if( currhandler == curl_handle){
+ currhandler = curl_handle2;
+ bakhandler = curl_handle;
+ } else if(currhandler == curl_handle2) {
+ currhandler = curl_handle;
+ bakhandler = curl_handle2;
+ }
+ rc = evel_post_multiapi(json_body, json_size, currhandler, send_url2, bakhandler,send_url);
+ if (rc != EVEL_SUCCESS)
+ EVEL_ERROR("Failed to transfer the data to failover %s. Error code=%d", send_url2, rc);
+ }
+
+ }
+ else
+ {
+ EVEL_DEBUG("Internal event received");
+ internal_msg = (EVENT_INTERNAL *) msg;
+ assert(internal_msg->command == EVT_CMD_TERMINATE);
+ evt_handler_state = EVT_HANDLER_TERMINATING;
+ }
+
+ /*************************************************************************/
+ /* We are responsible for freeing the memory. */
+ /*************************************************************************/
+ evel_free_event(msg);
+ msg = NULL;
+
+ /*************************************************************************/
+ /* There may be a single priority post to be sent. */
+ /*************************************************************************/
+ if (priority_post.memory != NULL)
+ {
+ EVEL_DEBUG("Priority Post");
+
+ /***********************************************************************/
+ /* Set the URL for the throttling API. */
+ /***********************************************************************/
+ curl_rc = curl_easy_setopt(curl_handle, CURLOPT_URL, evel_throt_api_url);
+ if (curl_rc != CURLE_OK)
+ {
+ /*********************************************************************/
+ /* This is only likely to happen with CURLE_OUT_OF_MEMORY, in which */
+ /* case we carry on regardless. */
+ /*********************************************************************/
+ EVEL_ERROR("Failed to set throttling URL. Error code=%d", rc);
+ }
+ else
+ {
+ rc = evel_post_api(priority_post.memory, priority_post.size);
+ if (rc != EVEL_SUCCESS)
+ {
+ EVEL_ERROR("Failed to transfer priority post. Error code=%d", rc);
+ }
+ }
+
+ /***********************************************************************/
+ /* Reinstate the URL for the event API. */
+ /***********************************************************************/
+ curl_rc = curl_easy_setopt(curl_handle, CURLOPT_URL, evel_event_api_url);
+ if (curl_rc != CURLE_OK)
+ {
+ /*********************************************************************/
+ /* This is only likely to happen with CURLE_OUT_OF_MEMORY, in which */
+ /* case we carry on regardless. */
+ /*********************************************************************/
+ EVEL_ERROR("Failed to reinstate events URL. Error code=%d", rc);
+ }
+
+ /***********************************************************************/
+ /* We are responsible for freeing the memory. */
+ /***********************************************************************/
+ free(priority_post.memory);
+ priority_post.memory = NULL;
+ }
+ }
+
+ /***************************************************************************/
+ /* The event handler is now exiting. The ring-buffer could contain events */
+ /* which have not been processed, so deplete those. Because we've been */
+ /* asked to exit we can be confident that the foreground will have stopped */
+ /* sending events in so we know that this process will conclude! */
+ /***************************************************************************/
+ evt_handler_state = EVT_HANDLER_TERMINATING;
+ while (!ring_buffer_is_empty(&event_buffer))
+ {
+ EVEL_DEBUG("Reading event from buffer");
+ msg = ring_buffer_read(&event_buffer);
+ evel_free_event(msg);
+ }
+ evt_handler_state = EVT_HANDLER_TERMINATED;
+ EVEL_INFO("Event handler thread stopped");
+
+ return (NULL);
+}
+
+
+/**************************************************************************//**
+ * Event Handler.
+ *
+ * Watch for messages coming on the internal queue and send them to the
+ * listener.
+ *
+ * param[in] arg Argument - unused.
+ *****************************************************************************/
+static void * event_handler(void * arg __attribute__ ((unused)))
+{
+ int old_type = 0;
+ EVENT_HEADER * msg = NULL;
+ EVENT_INTERNAL * internal_msg = NULL;
+ int json_size = 0;
+ char json_body[EVEL_MAX_JSON_BODY];
+ int rc = EVEL_SUCCESS;
+ CURLcode curl_rc;
+
+ EVEL_INFO("Event handler thread started");
+
+ /***************************************************************************/
+ /* Set this thread to be cancellable immediately. */
+ /***************************************************************************/
+ pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, &old_type);
+
+ /***************************************************************************/
+ /* Set the handler as active, defending against weird situations like */
+ /* immediately shutting down after initializing the library so the */
+ /* handler never gets started up properly. */
+ /***************************************************************************/
+ if (evt_handler_state == EVT_HANDLER_INACTIVE)
+ {
+ evt_handler_state = EVT_HANDLER_ACTIVE;
+ }
+ else
+ {
+ EVEL_ERROR("Event Handler State was not INACTIVE at start-up - "
+ "Handler will exit immediately!");
+ }
+
+ while (evt_handler_state == EVT_HANDLER_ACTIVE)
+ {
+ /*************************************************************************/
+ /* Wait for a message to be received. */
+ /*************************************************************************/
EVEL_DEBUG("Event handler getting any messages");
msg = ring_buffer_read(&event_buffer);
rc = evel_post_api(json_body, json_size);
if (rc != EVEL_SUCCESS)
{
- EVEL_ERROR("Failed to transfer the data. Error code=%d", rc);
+ EVEL_ERROR("Failed to transfer the data %s. Error code=%d", evel_batch_api_url, rc);
}
}
else if (msg->event_domain != EVEL_DOMAIN_INTERNAL )
rc = evel_post_api(json_body, json_size);
if (rc != EVEL_SUCCESS)
{
- EVEL_ERROR("Failed to transfer the data. Error code=%d", rc);
+ EVEL_ERROR("Failed to transfer the data %s. Error code=%d",evel_event_api_url, rc);
}
}
else
}
else
{
- EVEL_DEBUG("Decode JSON response tokens");
- if (!evel_handle_response_tokens(chunk, json_tokens, num_tokens, post))
- {
- EVEL_ERROR("Failed to handle JSON response.");
- }
+ //EVEL_DEBUG("Decode JSON response tokens");
+ //if (!evel_handle_response_tokens(chunk, json_tokens, num_tokens, post))
+ //{
+ // EVEL_ERROR("Failed to handle JSON response.");
+ //}
}
EVEL_EXIT();
Code Review / demo.git / commitdiff