C/C++ API Interface adapters

An overview of the behaviour of the C/C++ API interface adapters, including their functions and parameters.

As we mentioned earlier, a typical out-of-process adapter performs the following tasks:

1. Acquiring an address (URI) that will uniquely identify a specific stream and telling the communication layer (provided by ) how to find that stream.

2. Opening a connection to a tream.

3. Reading (or writing) the desired data. Typically the read (or write) operation is in a loop.

4. Closing the connection (if the adapter is not going to run indefinitely).

If the adapter is reading from the stream, then the adapter may need to wait until the next message arrives. There are three possible ways that the adapter can wait for the next message:

• The adapter can "poll" until a message arrives.

• The adapter can make a blocking call that will wait until a message arrives.

• The adapter can register a callback function that will be called when a message arrives.

To poll, the adapter should call the C8SubscriberGetNextMessage() function and specify a maximum amount of time to wait. If a message does not arrive within the specified time, C8SubscriberGetNextMessage() will return control to the caller, which may do some work (or yield the processor) and then continue polling later by calling C8SubscriberGetNextMessage() again. Passing 0 as the wait time returns control immediately, without waiting.

To register a callback function, use the C8SubscriberSetCallback() function.

Below we describe all the functions in the API, including the functions that allow users to attach to streams, read or write streams, and detach from streams. (Acquiring a stream's address is done outside the API, and will be explained later.) The header file c8adapter.h describes the interface.

Remember to call C8ClientSDKInitialize() before calling any functions in an out-of-process adapter.

• C8Publisher *C8PublisherCreate(const C8Char *i_uri) – Purpose: Create a publisher to the given URI for publishing messages. The publisher object returned by this function is used later when publishing messages.

  Parameters:

  • i_uri: The URI of the stream you want to publish to.

  Returns: A pointer to a C8Publisher object (or NULL if error). Publishers returned by this function must be freed with C8PublisherDestroy().

• C8Publisher *C8PublisherCreateA(const C8Char *i_uri, const C8UserCredentials *i_credentials) – Purpose: Like C8PublisherCreate(), this creates a publisher to the given stream URI, but this function requires user authentication. The returned publisher object is used later when sending (publishing) messages.

  Parameters:

  • i_uri: The URI of the stream you want to publish to.

  • i_credentials: A pointer to a structure containing your credentials (username and password).

  Returns: A pointer to a C8Publisher object (or NULL if error). Publishers returned by this function must be freed with C8PublisherDestroy().

• C8Publisher *C8PublisherCreateGD(const C8Char *i_uri, const C8UserCredentials *i_credentials, const C8Char *i_session_id, C8Interval i_timeout – Purpose: Creates a Guaranteed Delivery publisher to the given stream URI (see Implementing Guaranteed Processing for information about Guaranteed Delivery) and establishes (or re-establishes) a session with CEP Server. The returned publisher object is used later when sending (publishing) messages.

  Parameters:

  • i_uri: The URI of the stream you want to publish to.

  • i_credentials: (Optional). A pointer to a structure containing your credentials (user name and password).

  • i_session_id: The unique session ID for Guaranteed Delivery. If you are reconnecting after a failure, be sure to use the same session ID.

  • i_timeout: How long to wait to establish a session before timing out.

  Returns: A pointer to a C8Publisher object (or NULL if error). Publishers returned by this function must be freed with C8PublisherDestroy().

• void C8PublisherDestroy(C8Publisher *i_pub) – Purpose: Frees a C8Publisher object and any associated schema objects.

  Parameters:

  • i_pub: A pointer to the publisher object to be destroyed (de-allocated).

  Returns:

• C8Bool C8PublisherIsConnected(C8Publisher * i_pub) – Purpose: Determines whether or not a publisher is connected and ready to send messages.

  Parameters:

  • i_pub: The publisher.

  Returns: C8_TRUE if connected, C8_FALSE otherwise.

• const C8Schema * C8PublisherGetSchema(const C8Publisher *i_pub) – Purpose: Get the schema of the stream that is being published to.

  Parameters:

  • i_pub: The publisher for which you want to get the schema.

  Returns: A pointer to a schema. The returned schema remains valid while the publisher remains valid. you should not C8Free() the schema; the schema will be de-allocated when the publisher is de-allocated.

• C8Status C8PublisherSendMessage(C8Publisher *i_pub, C8Message *i_msg) – Purpose: Publishes a message to the stream whose URI was specified when the publisher *i_pub was created by calling C8PublisherCreate().

  Once a message is published, modifying the message is not allowed. However, you may deallocate a message with C8MessageDestroy() at any time. If the publisher is a Guaranteed Publisher, the message is not actually sent to the stream until you call C8PublisherCommit.

  Parameters:

  • i_pub: The publisher for which you want to get the schema.

  • i_msg: The message that you want to publish.

  Returns: C8_OK if successful, C8_FAIL otherwise.

• C8Status C8PublisherSendMessages(C8Publisher *i_pub, C8Message **i_msg, C8UInt i_cnt) – Purpose: Publishes an array of messages to a stream. If the publisher is a Guaranteed Publisher, the messages are not actually sent to the stream until you call C8PublisherCommit.

  Parameters:

  • i_pub: The publisher that specifies which stream to publish to.

  • i_msg: A pointer to an array of messages to be sent.

  • i_cnt: The number of messages in the array.

  Returns: C8_OK if successful, C8_FAIL otherwise.

• C8Status *C8PublisherCommit(C8Publisher *i_pub, const C8Char *i_batch_id) – Purpose: Send all messages that have been queued with C8PublisherSendMessage or C8PublisherSendMessages as a batch. Used with a Guaranteed Delivery publisher.

  Parameters:

  • i_pub: The Guaranteed Delivery publisher that specifies which stream to publish to.

  • i_batch_id: The ID of this batch. See Implementing Guaranteed Processing for information about Guaranteed Delivery.

  Returns: C8_OK if successful, C8_FAIL otherwise.

• C8Status * C8PublisherSendMessageBatch(C8Publisher * i_pub, C8Message** i_msgs, C8UInt i_cnt, const C8Char * i_batch_id ) – Purpose: Send a batch of messages. Used with a Guaranteed Delivery publisher.

  Parameters:

  • i_pub: The Guaranteed Delivery publisher that specifies which stream to publish to.

  • i_msgs: An array of messages

  • i_cnt: The number of messages in i_msgs

  • i_batch_id: The unique ID of this batch. See Implementing Guaranteed Processing for information about Guaranteed Delivery.

  Returns: C8_OK if successful, C8_FAIL otherwise.

• const C8Char *C8PublisherGetLastBatchId(C8Publisher *i_pub) – Purpose: Retrieves the ID of the last batch of messages received from a Guaranteed Delivery publisher.

  Parameters:

  • i_pub: The Guaranteed Delivery publisher.

  Returns: A constant pointer to the ID of the last batch processed by the server, or NULL if no batch has been processed since the last restart with clean slate.

• C8Subscriber *C8SubscriberCreate(const C8Char *i_uri) – Purpose: Creates a subscriber to a given stream.

  Parameters:

  • i_uri: The URI of the stream to subscribe to.

  Returns: Pointer to a subscriber object, or NULL if an error occurs. Subscribers should be freed with C8SubscriberDestroy().

• C8Subscriber *C8SubscriberCreateA(const C8Char *i_uri, const C8UserCredentials *i_credentials) – Purpose: Creates a subscriber to a given stream that requires user authentication.

  Parameters:

  • i_uri: The URI of the stream to subscribe to.

  • i_credentials: A pointer to a structure containing your credentials (username and password).

  Returns: A pointer to a subscriber object or NULL if an error occurs. Subscribers should be freed with C8SubscriberDestroy().

• C8Subscriber *C8SubscriberCreateGD(const C8Char *i_uri, const C8UserCredentials *i_credentials, const C8Char * i_session_id, const C8Char * i_last_batch_id, C8Interval i_timeout) – Purpose: Creates a Guaranteed Delivery subscriber to a given stream and establishes (or re-establishes) a session with the server.

  Parameters:

  • i_uri: The URI of the stream to subscribe to.

  • i_credentials: A pointer to a structure containing your credentials (username and password). (Optional.)

  • i_session_id : The unique session ID for Guaranteed Delivery. If you are reconnecting after a failure, be sure to use the same session ID.

  • i_last_batch_id: The ID of the last batch processed by this adapter. The server should begin sending messages with the batch following the one specified here.

  • i_timeout: How long to wait to establish a session before timing out.

  See Implementing Guaranteed Processing for information about Guaranteed Delivery.

  Returns: A pointer to a subscriber object or NULL if an error occurs. Subscribers should be freed with C8SubscriberDestroy().

• C8Message * C8SubscriberGetNextMessage(C8Subscriber *i_sub, C8Interval i_to) – Purpose: Gets the next pending message from the subscriber. If this is a Guaranteed Delivery subscriber, call C8SubscriberGetNextBatchID() to determine if this message begins a new batch of messages.

  Parameters:

  • i_sub: The subscriber object from which to get the next message.

  • i_to: Indicates the maximum amount of time (in microseconds) to wait for the next message.

  Returns: The next pending message, or NULL if no messages are pending. In case of a timeout, the call to C8ErrorGetCode() returns C8_ERR_TIMEOUT; a different error code is returned if there was an actual error. The message must be deleted with C8MessageDestroy().

  Warning!  

  Note that whether or not the timeout was reached, this function may return NULL to indicate that there were no messages, so you must always check the return value from this function - do not assume that you will always get a message.

• C8Status C8SubscriberGetNextBatch(C8Subscriber * i_sub, C8Message** o_msgs, C8UInt* io_cnt, const C8Char** o_batch_id, C8Interval i_to ) – Purpose: Get the next batch of messages from the guaranteed delivery subscriber.

  Parameters:

  • i_sub: The Guaranteed Delivery subscriber object from which to get the next batch of messages.

  • o_msgs: A pointer to an array to hold the returned messages.

  • io_cnt: The number of messages that will fit into 0_msgs (input) or the number of messages in o_msgs (output). If the entire batch of messages does not fit in the allocated array, subsequent calls to this method will retrieve the rest of the batch.

  • o_batch_id: The ID of this batch of messages.

  • i_to : Indicates the maximum amount of time (in microseconds) to wait for the batch.

  Returns: C8_OK if successful, C8_FAIL otherwise. The messages must be deleted with C8MessageDestroy().

• const C8Char *C8SubscriberGetLastBatchId(C8Subscriber *i_sub) – Purpose: Retrieves the ID of the last batch of messages sent by the server to a Guaranteed Delivery subscriber.

  Parameters:

  • i_sub: The Guaranteed Delivery subscriber.

  Returns: A constant pointer the ID of the last batch sent.

• void C8SubscriberSetCallback(C8Subscriber *i_sub, C8SubscriberCallbackFn i_cb, void *i_cb_prm) – Purpose: Sets a callback function that will be called when a new message is available on the stream that has been subscribed to.

  The callback function will be called by the adapter library code (which is linked into your out-of-process adapter) and therefore will be running in a thread that is running the library code. Thus, the callback function should be as short as possible and require minimum locking. The callback function should not itself process the newly-arrived message. In many cases, the callback function may simply set a flag and then return, leaving other parts of your code to check the flag, retrieve the message, and then process the message when their thread executes.

  Warning!  

  In some cases the callback routine may be called even if there is no message. When you call C8SubscriberGetNextMessage() (which you should NOT call inside the callback routine itself), be sure to check that C8SubscriberGetNextMessage() returned a valid C8Message pointer, not NULL.

  Parameters:

  • i_sub: The subscription whose stream you want to be notified about.

  • i_cb : The new callback function for this subscriber.

  • i_db_prm: The parameter with which to call the callback.

  Returns: Nothing.

• void C8SubscriberDestroy(C8Subscriber *i_sub) – Purpose: Frees the subscriber.

  Parameters:

  • i_sub: The subscriber that you want to free or de-allocate.

  Returns: Nothing.

• C8Bool C8SubscriberIsConnected(C8Subscriber * i_sub) – Purpose: Determines whether or not a subscriber is connected and ready to receive messages.

  Parameters:

  • i_sub: The subscriber.

  Returns: C8_TRUE if connected, C8_FALSE otherwise.

• const C8Schema *C8SubscriberGetSchema(const C8Subscriber *i_sub) – Purpose: Get the schema of the stream subscribed to.

  Parameters:

  • i_sub: The subscriber specifies which stream you want to know the schema of.

  Returns: The schema of the stream you've subscribed to or NULL if an error occurs. The returned schema remains valid while the subscriber remains valid.

• C8Char * C8ResolveUri(const C8Char * i_uri); – Purpose: Resolves a CCL URI into an HTTP/HTTPS URI.

  Parameters:

  • i_uri: The URI to resolve.

  Returns: The HTTP URL or NULL if an error occurs. The pointer returned must be freed with C8Free().

• C8Char * C8ResolveUriA(const C8Char * i_uri, const C8UserCredentials *i_credentials); – Purpose: Resolves a CCL URI into an HTTP/HTTPS URI, requiring that you pass appropriate user authentication credentials before doing so.

  Parameters:

  • i_uri: The URI to resolve.

  • i_credentials: A pointer to a structure containing your credentials (user name and password).

  Returns: The HTTP URL. This returns NULL if an error occurs. The pointer returned must be freed with C8Free().
• C8Char ** C8ResolveClusterUri(const C8Char * i_uri); – Purpose: Resolves a CCL URI into one or more HTTP/HTTPS URIs.

  Parameters:

  • i_uri: The URI to resolve.

  Returns: A pointer to a NULL-terminated array of string pointers, or NULL if an error occurs. The returned array pointer and all of it's contained pointers must be freed with C8Free().

• C8Char ** C8ResolveClusterUri(const C8Char * i_uri, const C8UserCredentials *i_credentials); – Purpose: Resolves a CCL URI into one or more HTTP/HTTPS URIs, requiring that you pass appropriate user authentication credentials before doing so.

  Parameters:

  • i_uri: The URI to resolve.

  • i_credentials: A pointer to a structure containing your credentials (user name and password).

  Returns: A pointer to a NULL-terminated array of string pointers, or NULL if an error occurs. The returned array pointer and all of it's contained pointers must be freed with C8Free().

• C8Char * C8UriAppendQueryString(const C8Char* i_base_uri, const C8Char* i_query_string); – Purpose: Appends a query string to the passed-in URI. Useful if you want to specify filters when subscribing to a stream.

  Parameters:

  • i_base_uri: The URI to extend.

  • i_query_string: The URL-encoded string to append to i_base_uri.

  Returns: The URI in the form base_uri?query_string. This returns NULL if an error occurs. The pointer returned must be freed with C8Free().

• C8Char * C8UriAppendQueryStringParameter(const C8Char* i_base_uri, const C8Char *i_key, const C8Char* i_value); – Purpose: Appends query parameters to the passed-in URI. Useful if you want to specify filters when subscribing to a stream.

  Parameters:

  • i_base_uri: The URI to extend.

  • i_key: The unencoded key.

  • i_value: The unencoded value.

  Returns: The URL-encoded URI in the form base_uri?key=value. This returns NULL if an error occurs. The pointer returned must be freed with C8Free().