PubSub+ Messaging API For C  7.29.0.6
 All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
ex/RRDirectReplier.c
/*
* This sample shows how to implement a Requester for direct Request-Reply messaging, where
*
* RRDirectRequester: A message Endpoint that sends a request message and waits to receive a
* reply message as a response.
* RRDirectReplier: A message Endpoint that waits to receive a request message and responses
* to it by sending a reply message.
*
* |-------------------| ---RequestTopic --> |------------------|
* | RRDirectRequester | | RRDirectReplier |
* |-------------------| <--ReplyToTopic---- |------------------|
*
* Copyright 2013-2024 Solace Corporation. All rights reserved.
*
*/
#include "os.h"
#include "common.h"
#include "RRcommon.h"
/*****************************************************************************
* Received message handling code
*****************************************************************************/
requestMsgReceiveCallback ( solClient_opaqueSession_pt opaqueSession_p, solClient_opaqueMsg_pt msg_p, void *user_p )
{
solClient_bool_t resultOk = FALSE;
solClient_int8_t operation = -1;
solClient_int32_t operand1 = -1;
solClient_int32_t operand2 = -1;
double result;
/*
* Get the operator, operand1 and operand2 from the stream in the binary
* attachment.
*/
if ( ( rc = solClient_msg_getBinaryAttachmentStream ( msg_p, &stream_p ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_msg_getBinaryAttachmentStream()" );
goto createReply;
}
/* Get the operation, operand1 and operand2 from the stream. */
if ( ( rc = solClient_container_getInt8 ( stream_p, &operation, NULL ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_container_getInt8() for operation" );
goto createReply;
}
if ( ( rc = solClient_container_getInt32 ( stream_p, &operand1, NULL ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_container_getInt32() for operand1" );
goto createReply;
}
if ( ( rc = solClient_container_getInt32 ( stream_p, &operand2, NULL ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_container_getInt32() for operand2" );
goto createReply;
}
/* Do the requested calculation. */
switch ( operation ) {
case plusOperation:
result = (double)operand1 + (double)operand2;
resultOk = TRUE;
break;
case minusOperation:
result = (double)operand1 - (double)operand2;
resultOk = TRUE;
break;
case timesOperation:
result = (double)operand1 * (double)operand2;
resultOk = TRUE;
break;
case divideOperation:
if (operand2 != 0) {
result = (double)operand1 / (double)operand2;
resultOk = TRUE;
}
break;
default:
break;
}
createReply:
if ( resultOk ) {
printf( " Received request for %d %s %d, sending reply with result %f. \n",
operand1, RR_operationToString ( operation ), operand2, result );
} else {
printf( " Received request for %d %s %d, sending reply with a failure status.\n",
operand1, RR_operationToString ( operation ), operand2 );
}
/*
* Allocate a message to construct the reply, and put in the status and result in a
* stream.
*/
if ( ( rc = solClient_msg_alloc ( &replyMsg_p ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_msg_alloc()" );
}
if ( ( rc = solClient_msg_createBinaryAttachmentStream ( replyMsg_p, &replyStream_p, 32 ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_msg_createBinaryAttachmentStream()" );
goto freeMsg;
}
if ( ( rc = solClient_container_addBoolean ( replyStream_p, resultOk, NULL ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_container_addBoolean()" );
goto freeMsg;
}
if ( resultOk ) {
if ( ( rc = solClient_container_addDouble ( replyStream_p, result, NULL ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_container_addDouble()" );
goto freeMsg;
}
}
if ( ( rc = solClient_session_sendReply ( opaqueSession_p, msg_p, replyMsg_p ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_session_sendReply()" );
goto freeMsg;
}
freeMsg:
if ( ( rc = solClient_msg_free ( &replyMsg_p ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_msg_free()" );
}
}
/*
* fn main()
* param appliance_ip The message backbone IP address.
* param appliance_username The client username.
* param request topic
*
* The entry point to the application.
*/
int
main ( int argc, char *argv[] )
{
struct commonOptions commandOpts;
/***********Context-related variable definitions*********/
/***********Session-related variable definitions*********/
const char *sessionProps[50];
int propIndex = 0;
printf ( "\nRRDirectReplier.c (Copyright 2013-2024 Solace Corporation. All rights reserved.)\n" );
/* Intialize Control C handling */
initSigHandler ( );
/*************************************************************************
* Parse command options
*************************************************************************/
common_initCommandOptions(&commandOpts,
( USER_PARAM_MASK |
DEST_PARAM_MASK ), /* required parameters */
( HOST_PARAM_MASK |
PASS_PARAM_MASK |
LOG_LEVEL_MASK |
USE_GSS_MASK |
ZIP_LEVEL_MASK)); /* optional parameters */
if ( common_parseCommandOptions ( argc, argv, &commandOpts, NULL ) == 0 ) {
exit (1);
}
/*************************************************************************
* Initialize the API and setup logging level
*************************************************************************/
/* solClient needs to be initialized before any other API calls are made. */
common_handleError ( rc, "solClient_initialize()" );
goto cleanup;
}
common_printCCSMPversion ( );
/*
* Standard logging levels can be set independently for the API and the
* application. In this case, the ALL category is used to set the log level for
* both at the same time.
*/
/*************************************************************************
* CREATE A CONTEXT
*************************************************************************/
solClient_log ( SOLCLIENT_LOG_INFO, "Creating solClient context" );
/*
* When creating the Context, specify that the Context thread should be
* created automatically instead of having the application create its own
* Context thread.
*/
&context_p, &contextFuncInfo, sizeof ( contextFuncInfo ) ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_context_create()" );
goto cleanup;
}
/*************************************************************************
* Create and connect a Session
*************************************************************************/
solClient_log ( SOLCLIENT_LOG_INFO, "Creating solClient sessions." );
sessionFuncInfo.rxMsgInfo.callback_p = requestMsgReceiveCallback;
sessionFuncInfo.rxMsgInfo.user_p = NULL;
sessionFuncInfo.eventInfo.callback_p = common_eventCallback;
sessionFuncInfo.eventInfo.user_p = NULL;;
propIndex = 0;
if ( commandOpts.targetHost[0] != (char) 0 ) {
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_HOST;
sessionProps[propIndex++] = commandOpts.targetHost;
}
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_COMPRESSION_LEVEL;
sessionProps[propIndex++] = ( commandOpts.enableCompression ) ? "9" : "0";
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_CONNECT_RETRIES;
sessionProps[propIndex++] = "3";
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_RECONNECT_RETRIES;
sessionProps[propIndex++] = "3";
/*
* Note: Reapplying subscriptions allows Sessions to reconnect after failure and
* have all their subscriptions automatically restored. For Sessions with many
* subscriptions, this can increase the amount of time required for a successful
* reconnect.
*/
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_REAPPLY_SUBSCRIPTIONS;
sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
/*
* Note: Including meta data fields such as sender timestamp, sender ID, and sequence
* number can reduce the maximum attainable throughput as significant extra encoding/
* decodingis required. This is true whether the fields are autogenerated or manually
* added.
*/
sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_GENERATE_SENDER_ID;
sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
if ( commandOpts.vpn[0] ) {
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_VPN_NAME;
sessionProps[propIndex++] = commandOpts.vpn;
}
/*
* The certificate validation property is ignored on non-SSL sessions.
* For simple demo applications, disable it on SSL sesssions (host
* string begins with tcps:) so a local trusted root and certificate
* store is not required. See the API usres guide for documentation
* on how to setup a trusted root so the servers certificate returned
* on the secure connection can be verified if this is desired.
*/
sessionProps[propIndex++] = SOLCLIENT_PROP_DISABLE_VAL;
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_USERNAME;
sessionProps[propIndex++] = commandOpts.username;
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_PASSWORD;
sessionProps[propIndex++] = commandOpts.password;
if ( commandOpts.useGSS ) {
sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_AUTHENTICATION_SCHEME;
}
sessionProps[propIndex] = NULL;
/*
* Create a session.
*/
if ( ( rc = solClient_session_create ( sessionProps,
context_p,
&session_p, &sessionFuncInfo, sizeof ( sessionFuncInfo ) ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_session_create()" );
goto cleanup;
}
/*
* Connect the session.
*/
if ( ( rc = solClient_session_connect ( session_p ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_session_connect()" );
goto cleanup;
}
/*************************************************************************
* Subscribe to the request topic
*************************************************************************/
if ( ( rc = solClient_session_topicSubscribeExt ( session_p,
commandOpts.destinationName) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_session_topicSubscribe()" );
goto sessionConnected;
}
/*************************************************************************
* Serve requests, CTRL-C to stop
*************************************************************************/
printf ( "Serving requests on topic '%s', Ctrl-C to stop.....\n", commandOpts.destinationName );
while ( !gotCtlC ) {
sleepInSec ( 1 );
}
printf ( "Got Ctrl-C, cleaning up\n" );
/*************************************************************************
* CLEANUP
*************************************************************************/
sessionConnected:
/* Disconnect the Session. */
if ( ( rc = solClient_session_disconnect ( session_p ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_session_disconnect()" );
}
cleanup:
/* Cleanup solClient. */
if ( ( rc = solClient_cleanup ( ) ) != SOLCLIENT_OK ) {
common_handleError ( rc, "solClient_cleanup()" );
}
goto notInitialized;
notInitialized:
return 0;
}