Pub/Sub Messaging with C++

Welcome back! In this lesson, we will explore another powerful feature of Redis: Publish/Subscribe (Pub/Sub) messaging. This lesson builds on our understanding of Redis and introduces a dynamic way to enable real-time communication within your applications using C++ and the hiredis library.

What You'll Learn

In this lesson, you will learn how to set up and use Redis Pub/Sub messaging in C++ to send and receive messages between different parts of your application. This is useful for creating real-time features like notifications, chat systems, or live updates.

Here's how you can set up a simple Pub/Sub system in Redis using C++:

#include <iostream>
#include <hiredis/hiredis.h>
#include <thread>
#include <atomic>

// Global flag to stop the listener thread
std::atomic<bool> keepRunning(true);

// Message handler function
void messageHandler(const redisReply* reply) {
    if (reply && reply->type == REDIS_REPLY_ARRAY && reply->elements == 3) {
        std::cout << "Received message: " << reply->element[2]->str << std::endl;
    } else {
        std::cerr << "Unexpected message format or error in reply." << std::endl;
    }
}

// Pub/Sub listener function
void runPubSub(redisContext* context) {
    while (keepRunning) {
        redisReply* reply = nullptr;
        if (redisGetReply(context, (void**)&reply) == REDIS_OK) {
            if (reply) {
                messageHandler(reply);
                freeReplyObject(reply);
            }
        } else {
            std::cerr << "Error receiving message: " << context->errstr << std::endl;
            break;
        }
    }
}

int main() {
    // Connect to the Redis server for subscribing
    redisContext* subContext = redisConnect("127.0.0.1", 6379);
    if (subContext == nullptr || subContext->err) {
        if (subContext) {
            std::cerr << "Connection error (sub): " << subContext->errstr << std::endl;
        } else {
            std::cerr << "Connection error: can't allocate Redis context (sub)" << std::endl;
        }
        return 1;
    }

    // Connect to the Redis server for publishing
    redisContext* pubContext = redisConnect("127.0.0.1", 6379);
    if (pubContext == nullptr || pubContext->err) {
        if (pubContext) {
            std::cerr << "Connection error (pub): " << pubContext->errstr << std::endl;
        } else {
            std::cerr << "Connection error: can't allocate Redis context (pub)" << std::endl;
        }
        redisFree(subContext);
        return 1;
    }

    // Subscribe to the "notifications" channel
    std::cout << "Subscribing to channel 'notifications'..." << std::endl;
    redisReply* reply = (redisReply*)redisCommand(subContext, "SUBSCRIBE notifications");
    if (!reply || reply->type != REDIS_REPLY_ARRAY) {
        std::cerr << "Failed to subscribe to channel or unexpected reply type." << std::endl;
        if (reply) freeReplyObject(reply);
        redisFree(subContext);
        redisFree(pubContext);
        return 1;
    }
    freeReplyObject(reply);

    // Start the Pub/Sub listener thread
    std::thread listenerThread(runPubSub, subContext);

    // Sleep to allow listener to set up
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Publish a message to the "notifications" channel
    std::cout << "Publishing a test message..." << std::endl;
    redisReply* publishReply = (redisReply*)redisCommand(pubContext, "PUBLISH notifications %s", "Hello, Redis!");
    if (publishReply && publishReply->type == REDIS_REPLY_INTEGER) {
        std::cout << "Message published, number of subscribers that received the message: " << publishReply->integer << std::endl;
    } else {
        std::cerr << "Failed to publish message or unexpected reply type." << std::endl;
    }
    if (publishReply) freeReplyObject(publishReply);

    // Unsubscribe and stop the listener
    std::cout << "Unsubscribing and stopping listener..." << std::endl;
    keepRunning = false;
    redisCommand(subContext, "UNSUBSCRIBE notifications");
    listenerThread.join();

    // Free the Redis contexts
    redisFree(subContext);
    redisFree(pubContext);
    std::cout << "Program finished." << std::endl;

    return 0;
}
Explanation of the Code

  • We define a messageHandler function that processes messages received from the Redis server. This function checks the type and structure of the message, then prints the received message to the standard output.

  • In the runPubSub function, we handle the subscription to the notifications channel. This function loops continuously, using redisGetReply to listen for incoming messages, which are then processed by the messageHandler. This loop continues running until the keepRunning flag is set to false.

  • We start the Pub/Sub listener in a separate thread using std::thread, executing the runPubSub function. This allows the main function to proceed without waiting for the incoming messages, thus implementing non-blocking behavior.

  • After allowing some time for the listener to initialize using std::this_thread::sleep_for, we publish a test message to the notifications channel with the PUBLISH command. The command returns the number of subscribers that received the message, which is then displayed.

  • Resources are properly managed by explicitly unsubscribing from the notifications channel and setting the keepRunning flag to false, effectively stopping the listener thread. We ensure that the listenerThread is joined, meaning it finishes executing before the program ends. Lastly, we free the Redis contexts using redisFree to prevent any memory leaks.

Why It Matters

The Pub/Sub messaging model is crucial for enabling real-time communication in modern applications. Whether it's sending notifications to users, making chat applications, or updating dashboards in real-time, Pub/Sub can help you achieve these goals efficiently in C++.

The benefits of mastering Pub/Sub messaging using Redis include:

  • Real-Time Communication: Instantly update parts of your application as events occur, providing a seamless user experience.
  • Decoupled Architecture: Senders and receivers are independent, promoting modularity and easier maintenance of your application.
  • Scalability: Scale your application by adding more subscribers or publishers without altering the core logic.

Learning how to leverage Pub/Sub messaging with Redis using C++ will enable you to build responsive, scalable, and maintainable applications. Ready to get hands-on with the example? Let’s move forward and start implementing!

Previous
Next
Sign up
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal