Implementing Pub/Sub for Notifications

Welcome! In this unit, we will delve into implementing Pub/Sub for notifications within our Redis-based backend system project using C++. You've already learned how to manage user data, handle transactions, and use streams for event logging. Now, we'll add another powerful feature to our project: real-time notifications using Redis Pub/Sub (publish/subscribe). This will enable our system to send and receive messages instantaneously.

What You'll Build

In this unit, we'll focus on creating a simple real-time notification system using Redis Pub/Sub in C++. Specifically, we'll cover:

  1. Publishing Messages: How to send notifications.
  2. Subscribing to Channels: How to receive and handle notifications.

Here is a quick refresh of how Pub/Sub works in Redis using C++ and hiredis:

#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;
}
//Note: The exact order of the output messages may still vary due to the multi-threaded nature of the program.

/* Output: 
    Subscribing to channel 'notifications'...
    Publishing a test message...
    Message published, number of subscribers that received the message: 1
    Unsubscribing and stopping listener...
    Received message: Hello, Redis!
    Program finished.
*/

In this C++ snippet, the runPubSub function continuously listens for messages on the "notifications" channel and uses messageHandler to process and print incoming messages. The main function establishes two connections to the Redis server—one for subscribing and one for publishing messages—and utilizes threads to handle these tasks concurrently, ensuring real-time message delivery between publisher and subscriber.

Exciting, isn’t it? Now it's time to put this into practice. Let's implement the complete code to build our real-time notification system.

Happy coding!

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