ex/ios/examples/TopicDispatchExample.m

/*
 *  Copyright 2007-2024 Solace Corporation. All rights reserved.
 */

#import "TopicDispatchExample.h"

@implementation TopicDispatchExample

@synthesize solMsgRxCount;

- (id)initWithExampleInterface:(ExampleInterface *)exampleInterface {
    self = [super initWithExampleInterface:exampleInterface];

    // Set example name and description
    self.name = @"TopicDispatch";
    self.description = @"Demonstrates the use of topic dispatch to direct "
        @"received messages into specialized received data paths";

    return self;
}

- (solClient_rxMsgCallback_returnCode_t)
    messageReceiveCallbackWithSession:
        (solClient_opaqueSession_pt)opaqueSession_p
                              message:(solClient_opaqueMsg_pt)msg_p
                             userData:(void *)user_p {
    int index = ((__bridge NSNumber *)user_p).intValue;
    int oldValue = ((NSNumber *)[solMsgRxCount objectAtIndex:index]).intValue;
    solMsgRxCount[index] = [NSNumber numberWithInt:oldValue + 1];

    return SOLCLIENT_CALLBACK_OK;
}

- (void)run {
    [super run];

    solClient_returnCode_t rc = SOLCLIENT_OK;

    // Context variables
    solClient_opaqueContext_pt context_p;
    solClient_context_createFuncInfo_t contextFuncInfo =
        SOLCLIENT_CONTEXT_CREATEFUNC_INITIALIZER;

    // Session variables
    solClient_opaqueSession_pt session_p;
    solClient_session_createFuncInfo_t sessionFuncInfo =
        SOLCLIENT_SESSION_CREATEFUNC_INITIALIZER;
    const char *sessionProps[40];
    int propIndex;

    // Dispatch function variables
    solClient_session_rxMsgDispatchFuncInfo_t dispatchInfo =
        SOLCLIENT_SESSION_DISPATCHFUNC_INITIALIZER(
            SOLCLIENT_DISPATCH_TYPE_CALLBACK);
    solMsgRxCount = [[NSMutableArray alloc] init];
    for (int i = 0; i < 4; i++) {
        solMsgRxCount[i] = [NSNumber numberWithInt:0];
    }

    // Callback bridges
    callbackBridge_t bridges[4];
    for (int i = 0; i < 4; i++) {
        bridges[i].example_p = (__bridge void *)self;
        bridges[i].user_p = (__bridge void *)[[NSNumber alloc] initWithInt:i];
    }

    // Initialize the API; this must be done prior to first usage
    if ((rc = solClient_initialize(SOLCLIENT_LOG_DEFAULT_FILTER, NULL)) !=
        SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc
                            errorString:"solClient_initialize()"];
        goto notInitialized;
    }

    // Set up logging level and log example and API information
    [self setLoggingLevel];
    solClient_log(SOLCLIENT_LOG_NOTICE, "TopicDispatchExample.m (Copyright 2007-2024 Solace Corporation. All rights reserved.)\n");
    [self logCCSMPVersion];

    sessionFuncInfo.rxMsgInfo.callback_p = messageReceiveCallback;
    sessionFuncInfo.rxMsgInfo.user_p = &bridges[0];
    sessionFuncInfo.eventInfo.callback_p = eventCallback;
    sessionFuncInfo.eventInfo.user_p = &bridges[0];

    // Create a context to use for the session. In this case, specify that the
    // API automatically create the context thread instead of having the
    // application create its own context thread.
    solClient_log(SOLCLIENT_LOG_INFO, "Creating solClient context");
    if ((rc = solClient_context_create(
             SOLCLIENT_CONTEXT_PROPS_DEFAULT_WITH_CREATE_THREAD, &context_p,
             &contextFuncInfo, sizeof(contextFuncInfo))) != SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc
                            errorString:"solClient_context_create()"];
        goto cleanup;
    }

    // Create a session for sending and receiving messages
    propIndex = 0;
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_USERNAME;
    sessionProps[propIndex++] =
        [[self.parameters parameterWithId:PARAMETER_USERNAME].value
            cStringUsingEncoding:NSASCIIStringEncoding];
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_PASSWORD;
    sessionProps[propIndex++] =
        [[self.parameters parameterWithId:PARAMETER_PASSWORD].value
            cStringUsingEncoding:NSASCIIStringEncoding];

    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_HOST;
    sessionProps[propIndex++] =
        [[self.parameters parameterWithId:PARAMETER_HOST].value
            cStringUsingEncoding:NSASCIIStringEncoding];
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_CONNECT_RETRIES;
    sessionProps[propIndex++] = "3";
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_RECONNECT_RETRIES;
    sessionProps[propIndex++] = "3";
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_COMPRESSION_LEVEL;
    sessionProps[propIndex++] =
        ([self.parameters parameterWithId:PARAMETER_COMPRESSION]
             .value.boolValue)
            ? "9"
            : "0";
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_REAPPLY_SUBSCRIPTIONS;
    sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_GENERATE_SEND_TIMESTAMPS;
    sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_GENERATE_SENDER_ID;
    sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_GENERATE_SEQUENCE_NUMBER;
    sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;

    // Enable topic dispatch on the session
    sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_TOPIC_DISPATCH;
    sessionProps[propIndex++] = SOLCLIENT_PROP_ENABLE_VAL;

    if ([[self.parameters parameterWithId:PARAMETER_VPN].value
            cStringUsingEncoding:NSASCIIStringEncoding][0]) {
        sessionProps[propIndex++] = SOLCLIENT_SESSION_PROP_VPN_NAME;
        sessionProps[propIndex++] =
            [[self.parameters parameterWithId:PARAMETER_VPN].value
                cStringUsingEncoding:NSASCIIStringEncoding];
    }
    // The certificate validation property is ignored on non-SSL sessions.
    // For simple demo applications, disable it on SSL sessions (host
    // string begins with tcps:) so a local trusted root and certificate
    // store is not required. See the API users 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_SESSION_PROP_SSL_VALIDATE_CERTIFICATE;
    sessionProps[propIndex++] = SOLCLIENT_PROP_DISABLE_VAL;

    sessionProps[propIndex] = NULL;

    solClient_log(SOLCLIENT_LOG_INFO, "Creating solClient session");
    if ((rc = solClient_session_create(
             (char **)sessionProps, context_p, &session_p, &sessionFuncInfo,
             sizeof(sessionFuncInfo))) != SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc
                            errorString:"solClient_session_create()"];
        goto cleanup;
    }

    // Connect the session
    solClient_log(SOLCLIENT_LOG_INFO, "Connecting solClient session");
    if ((rc = solClient_session_connect(session_p)) != SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc
                            errorString:"solClient_session_connect()"];
        goto cleanup;
    }

    // Add subscription
    if ((rc = solClient_session_topicSubscribeExt(
             session_p, SOLCLIENT_SUBSCRIBE_FLAGS_WAITFORCONFIRM, "a/>")) !=
        SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc
                            errorString:"solClient_session_topicSubscribe()"];
        goto sessionConnected;
    }

    // Add first dispatch function. This dispatch is local only (does not add
    // the subscription on the appliance).
    dispatchInfo.callback_p = messageReceiveCallback;
    dispatchInfo.user_p = &bridges[1];
    if ((rc = solClient_session_topicSubscribeWithDispatch(
             session_p, (SOLCLIENT_SUBSCRIBE_FLAGS_LOCAL_DISPATCH_ONLY), "a/b",
             &dispatchInfo, 0)) != SOLCLIENT_OK) {
        [self
            handleErrorWithReturnCode:rc
                          errorString:
                              "solClient_session_topicSubscribeWithDispatch()"];
        goto sessionConnected;
    }

    // Add second dispatch function. This dispatch includes adding the
    // subscription to the appliance.
    dispatchInfo.callback_p = messageReceiveCallback;
    dispatchInfo.user_p = &bridges[2];
    if ((rc = solClient_session_topicSubscribeWithDispatch(
             session_p, SOLCLIENT_SUBSCRIBE_FLAGS_WAITFORCONFIRM, "c/>",
             &dispatchInfo, 0)) != SOLCLIENT_OK) {
        [self
            handleErrorWithReturnCode:rc
                          errorString:
                              "solClient_session_topicSubscribeWithDispatch()"];
        goto sessionConnected;
    }

    // Add third dispatch function. This dispatch is local only (no subscription
    // is added to appliance).
    dispatchInfo.callback_p = messageReceiveCallback;
    dispatchInfo.user_p = &bridges[3];
    if ((rc = solClient_session_topicSubscribeWithDispatch(
             session_p, (SOLCLIENT_SUBSCRIBE_FLAGS_LOCAL_DISPATCH_ONLY), "c/d",
             &dispatchInfo, 0)) != SOLCLIENT_OK) {
        [self
            handleErrorWithReturnCode:rc
                          errorString:
                              "solClient_session_topicSubscribeWithDispatch()"];
        goto sessionConnected;
    }

    solClient_log(SOLCLIENT_LOG_NOTICE, "Publishing messages...\n");
    [self publishMessageWithSession:session_p
                              topic:"a/c"
                       deliveryMode:SOLCLIENT_DELIVERY_MODE_DIRECT];
    [self publishMessageWithSession:session_p
                              topic:"a/b"
                       deliveryMode:SOLCLIENT_DELIVERY_MODE_DIRECT];
    [self publishMessageWithSession:session_p
                              topic:"c/d"
                       deliveryMode:SOLCLIENT_DELIVERY_MODE_DIRECT];
    [self publishMessageWithSession:session_p
                              topic:"c/e"
                       deliveryMode:SOLCLIENT_DELIVERY_MODE_DIRECT];

    // Pause to let messages return
    [NSThread sleepForTimeInterval:1];

    // Verify expected results
    if (((NSNumber *)solMsgRxCount[0]).intValue == 1) {
        solClient_log(
            SOLCLIENT_LOG_NOTICE,
            "Received exactly one message on session callback as expected.\n");
    } else {
        solClient_log(
            SOLCLIENT_LOG_ERROR,
            "Received %d message(s) on session callback - 1 expected.\n",
            solMsgRxCount[0]);
    }
    if (((NSNumber *)solMsgRxCount[1]).intValue == 1) {
        solClient_log(
            SOLCLIENT_LOG_NOTICE,
            "Received exactly one message on dispatch callback for topic 'a/b' "
            "as expected.\n");
    } else {
        solClient_log(SOLCLIENT_LOG_ERROR,
                      "Received %d message(s) on dispatch "
                      "callback for topic 'a/b' - 1 expected.\n",
                      solMsgRxCount[1]);
    }
    if (((NSNumber *)solMsgRxCount[2]).intValue == 2) {
        solClient_log(
            SOLCLIENT_LOG_NOTICE,
            "Received exactly two messages on dispatch callback for topic "
            "'c/>' as expected.\n");
    } else {
        solClient_log(SOLCLIENT_LOG_ERROR,
                      "Received %d message(s) on dispatch "
                      "callback for topic 'c/>' - 2 expected.\n",
                      solMsgRxCount[2]);
    }
    if (((NSNumber *)solMsgRxCount[3]).intValue == 1) {
        solClient_log(
            SOLCLIENT_LOG_NOTICE,
            "Received exactly one message on dispatch callback for topic 'c/d' "
            "as expected.\n");
    } else {
        solClient_log(SOLCLIENT_LOG_ERROR,
                      "Received %d message(s) on dispatch "
                      "callback for topic 'c/d' - 1 expected.\n",
                      solMsgRxCount[3]);
    }

sessionConnected:
    // Disconnect the session
    if ((rc = solClient_session_disconnect(session_p)) != SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc
                            errorString:"solClient_session_disconnect()"];
    }

cleanup:
    // Cleanup solclient
    if ((rc = solClient_cleanup()) != SOLCLIENT_OK) {
        [self handleErrorWithReturnCode:rc errorString:"solClient_cleanup()"];
    }

notInitialized:
    [self cleanup];
}

@end