Pub/Sub Messaging with Redis and C++

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;
}
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