Just a second...

Start subscribing with C

Create a C client within minutes that connects to the Diffusion™ server. This example creates a client that subscribes to a JSON topic called "processes" and prints its value to the console when the topic is updated.

The C client libraries rely on a number of dependencies. Depending on the platform you are using, these dependencies might be included in the client library. If they are not included in the client library, ensure that the dependencies are available on your development system.

For more information about dependencies on each supported platform, see C.

The C client library statically links to Apache Portable Runtime (APR) version 1.5 with APR-util. Ensure that you set APR_DECLARE_STATIC and APU_DECLARE_STATIC before you use any APR includes. You can set these values in the following ways:
  • By including diffusion.h before any APR includes. The diffusion.h file sets these values.
  • As command-line flags
For more information, see http://apr.apache.org

To complete this example, you need a Diffusion server.

You also must either:
  • create a named user that has a role with the select_topic and read_topic permissions
  • assign a role with those permissions to anonymous client sessions.
An example of a suitable role is "CLIENT". For more information about roles and permissions, see Role-based authorization.
This example steps through the lines of code required to subscribe to a topic. The full code example is provided after the steps.
  1. Get the Diffusion C client library for your platform and extract the ZIP file.
    The C client library is available in the clients/c directory of the Diffusion installation.
  2. Create a C file called cjson-subscribing-example.c.
    1. Include the following libraries:
      #include <stdio.h>
      #include <apr.h>
      
      #include "diffusion.h"
      #include "args.h"
    2. Define some values to be used later.
              #define SYNC_DEFAULT_TIMEOUT 5000 * 1000
      
              apr_pool_t *pool = NULL;
              apr_thread_mutex_t *mutex = NULL;
              apr_thread_cond_t *cond = NULL;
      
              ARG_OPTS_T arg_opts[] = {
              ARG_OPTS_HELP,
              {'u', "url", "Diffusion server URL", ARG_OPTIONAL, ARG_HAS_VALUE, "ws://localhost:8080"},
              {'p', "principal", "Principal (username) for the connection", ARG_OPTIONAL, ARG_HAS_VALUE, "user"},
              {'c', "credentials", "Credentials (password) for the connection", ARG_OPTIONAL, ARG_HAS_VALUE, "password"},
              {'t', "topic", "Topic name to subscribe", ARG_OPTIONAL, ARG_HAS_VALUE, "time"},
              END_OF_ARG_OPTS
              };

      The ARG_OPTS_T array is used to store arguments that can be passed when running the code, along with defaults if a value is not provided. You may wish to replace the defaults with your own values. hostname is the name of the system hosting your Diffusion server, port is the port the Diffusion server accepts client connections on, user is the name of a principal with the permissions required to subscribe to a topic, and password is the principal's password.

  3. Create a main method.
    Inside the main method add the following lines:
    int
    main(int argc, char **argv)
    {
            /*
             * Standard command-line parsing.
             */
            HASH_T *options = parse_cmdline(argc, argv, arg_opts);
            if(options == NULL || hash_get(options, "help") != NULL) {
                    show_usage(argc, argv, arg_opts);
                    return EXIT_FAILURE;
            }
    
            const char *url = hash_get(options, "url");
            const char *principal = hash_get(options, "principal");
            CREDENTIALS_T *credentials = NULL;
            const char *password = hash_get(options, "credentials");
            if(password != NULL) {
                    credentials = credentials_create_password(password);
            }
            const char *topic_name = hash_get(options, "topic");
    }
    This does standard parsing of the command-line parameters.
  4. Set up a mutex and condition variable, then create a session with the server.
            apr_initialize();
            apr_pool_create(&pool, NULL);
            apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_UNNESTED, pool);
            apr_thread_cond_create(&cond, pool);
    
            DIFFUSION_ERROR_T error = { 0 };
            SESSION_T *session = session_create(url, principal, credentials, NULL, NULL, &error);
            if(session == NULL) {
                    fprintf(stderr, "Failed to create session: %s\n", error.message);
                    return EXIT_FAILURE;
            }
  5. Handle any incoming subscription notifications, subscribe to a topic provided in the command line parameters (or the default one), process topic updates for 60 seconds then shut down.
            notify_subscription_register(session,(NOTIFY_SUBSCRIPTION_PARAMS_T) { .on_notify_subscription = on_notify_subscription });
    
            subscribe(session, (SUBSCRIPTION_PARAMS_T) { .topic_selector = topic_name, .on_topic_message = on_topic_message, .on_subscribe = on_subscribe });
    
            // Receive updates for 2 minutes
            sleep(120);
    
            session_close(session, NULL);
            session_free(session);
    
            credentials_free(credentials);
            hash_free(options, NULL, free);
    
            apr_thread_cond_destroy(cond);
            apr_thread_mutex_destroy(mutex);
            apr_pool_destroy(pool);
            apr_terminate();
    
            return EXIT_SUCCESS;
    Note that this code relies on callbacks which we will add in the following steps.
  6. Above the main method, create the on_notify_subscription callback to handle when this client receives a notification that it has been subscribed to a topic. Note that this could be called for topics that we have not explicitly subscribed to. Other control clients (or publishers) may request to subscribe this client to a topic.
    static int
    on_notify_subscription(SESSION_T *session, const SVC_NOTIFY_SUBSCRIPTION_REQUEST_T *request, void *context)
    {
            printf("on_notify_subscription: %d: \"%s\"\n",
                   request->topic_info.topic_id,
                   request->topic_info.topic_path);
            return HANDLER_SUCCESS;
    }
  7. Create the on_subscribe callback to handle when the server responds to say that a topic * subscription request has been received and processed.
    static int
    on_subscribe(SESSION_T *session, void *context_data)
    {
            printf("on_subscribe\n");
            return HANDLER_SUCCESS;
    }
  8. Create the on_topic_message callback. When a subscribed message is received, this callback is invoked to print the JSON value.
    static int
    on_topic_message(SESSION_T *session, const TOPIC_MESSAGE_T *msg)
    {
            printf("Received message for topic %s (%ld bytes)\n", msg->name, msg->payload->len);
    
            if(msg->details != NULL) {
                if(msg->details->topic_type == TOPIC_TYPE_JSON) {
                    // Convert payload to a JSON string
                    BUF_T *json = cbor_to_json(msg->payload->data, msg->payload->len);
                    printf("As JSON: %.*s\n", (int)json->len, json->data);
                    buf_free(json);
    
                }
                else {
                    printf("Hexdump of binary data:\n");
                    hexdump_buf(msg->payload);
                }
            }
            else {
                printf("Payload: %.*s\n",
                       (int)msg->payload->len,
                       msg->payload->data);
            }
    
            return HANDLER_SUCCESS;
    }
  9. Build your C client.
    1. Create a Makefile in the same directory as your C file.
      An example Makefile is provided after the steps.
    2. Run the make command to build the example.
      The cjson-subscribing-example binary is created in the target/bin directory.
  10. Run your C client from the command line.

The client prints to the console every time the value of the subscribed topic is updated. You can update the value of the topic by creating a publishing client to update the topic. By default, the client subscribes to the processes topic used in the Start publishing with C example.

The completed cjson-subscribing-example.c file contains the following code:
                            
                        
The Makefile contains the following code:
        # The following two variables must be set.
        #
        # Directory containing the C client include files.
        # DIFFUSION_C_CLIENT_INCDIR =
        #
        # Directory containing libdiffusion.a
        # DIFFUSION_C_CLIENT_LIBDIR =
        ifndef DIFFUSION_C_CLIENT_INCDIR
        $(error DIFFUSION_C_CLIENT_INCDIR is not set)
        endif
        ifndef DIFFUSION_C_CLIENT_LIBDIR
        $(error DIFFUSION_C_CLIENT_LIBDIR is not set)
        endif
        CC      = gcc
        # Extra definitions from parent directory, if they exist.
        -include ../makefile.defs
        CFLAGS      += -g -Wall -Werror -std=c99 -D_POSIX_C_SOURCE=200112L -D_XOPEN_SOURCE=700 -I$(DIFFUSION_C_CLIENT_INCDIR)
        LDFLAGS     += $(DIFFUSION_C_CLIENT_LIBDIR)/libdiffusion.a -lpthread -lpcre -lz $(LIBS)
        # If you have openssl installed then you can uncomment these.
        ifdef HAVE_OPEN_SSL
        LDFLAGS     += -lssl -lcrypto
        endif
        ARFLAGS     +=
        SOURCES = json/cjson-subscribing-example.c
        TARGETDIR   = target
        OBJDIR      = $(TARGETDIR)/objs
        BINDIR      = $(TARGETDIR)/bin
        OBJECTS     = $(SOURCES:.c=.o)
        TARGETS     = cjson-subscribing-example
        all:        prepare $(TARGETS)
        .PHONY:     all
        prepare:
        mkdir -p $(OBJDIR) $(BINDIR)
        $(OBJDIR)/%.o:  %.c
        $(CC) $(CFLAGS) -c -o [email protected] $<
        cjson-subscribing-example:  json/cjson-subscribing-example.c
        $(CC) $^ $(CFLAGS) $(LDFLAGS) -lm -o $(BINDIR)/[email protected]
        clean:
        rm -rf $(TARGETS) $(OBJECTS) $(TARGETDIR) core a.out