Real Time Notifications

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 . 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 . Specifically, we'll cover: Publishing messages: how to send notifications. Subscribing to channels: how to receive and handle notifications. We'll implement a complete C++ program using Boost.Redis that demonstrates both publishing and subscribing. The example will: Create separate connections for subscribing and publishing, which is required for Pub/Sub. Subscribe to a channel and listen for messages asynchronously. Publish a JSON-formatted message to that channel. Process and display the received message. Messages will use JSON format for structured data: JSON {"user": "alice", "text": "Hello everyone!"}

Section 1: Includes, aliases, and the receive loop

This first section sets up the headers, type aliases, and the function that listens for Pub/Sub messages. C++#include <boost/redis.hpp> #include <boost/redis/src.hpp> // Include in one translation unit only #include <boost/asio.hpp> #include <nlohmann/json.hpp> #include <iostream> #include <memory> #include <chrono> #include <string> namespace net = boost::asio; using boost::redis::connection; using boost::redis::config; using boost::redis::request; using boost::redis::generic_response; using boost::redis::logger; using boost::system::error_code; using json = nlohmann::json; // Listen for Pub/Sub messages on the subscriber connection. void start_receive_loop(std::shared_ptr<connection> sub_conn) { auto resp = std::make_shared<generic_response>(); // Set the response object for the next message. sub_conn->set_receive_response(*resp); // Receive the next push message asynchronously. sub_conn->async_receive( [sub_conn, resp](error_code ec, std::size_t) { if (ec) { if (ec == net::error::operation_aborted) { // Connection was cancelled, just exit the loop. return; } std::cerr << "Receive error: " << ec.message() << "\n"; return; } // RESP3 push message structure for pub/sub: // ["push", "message", "<channel>", "<payload>"] auto const& nodes = resp->value(); if (nodes.size() >= 4) { std::string kind = nodes[1].value; if (kind == "message") { std::string channel = nodes[2].value; std::string payload = nodes[3].value; // Payload is JSON: {"user": "...", "text": "..."} try { json data = json::parse(payload); std::string user = data["user"]; std::string text = data["text"]; std::cout << "Received message from " << user << " on channel '" << channel << "': " << text << "\n"; } catch (const std::exception& e) { std::cerr << "JSON parse error: " << e.what() << "\n"; } // Stop after receiving the first message. sub_conn->cancel(); return; } } // For other push types (e.g., "subscribe" confirmations), keep listening. start_receive_loop(sub_conn); }); } This section prepares the program to receive messages from Redis. It includes the libraries for: Redis communication with Boost.Redis asynchronous execution with Boost.Asio JSON parsing with nlohmann::json It also defines start_receive_loop, which waits for push messages on the subscriber connection. The function creates a generic_response object and gives it to the connection with set_receive_response(). Then it calls async_receive() to wait for the next Pub/Sub message without blocking the program. For Pub/Sub in RESP3, a published message arrives in this structure: text["push", "message", "<channel>", "<payload>"] The code checks that structure, extracts the channel and payload, parses the JSON message, and prints the sender and text. If the push message is something else, such as a subscribe confirmation, the function starts listening again. This receive loop is what allows the subscriber connection to keep handling incoming messages asynchronously.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Section 2: Create connections and subscribe to a channel

This section creates the Redis connections, starts them, and subscribes to a channel.

C++

int main()
{
    net::io_context ioc;

    // One connection dedicated to SUBSCRIBE (subscriber mode).
    auto sub_conn = std::make_shared<connection>(ioc);

    // Separate connection for regular commands like PUBLISH.
    auto pub_conn = std::make_shared<connection>(ioc);

    config cfg;
    cfg.addr.host = "127.0.0.1";
    cfg.addr.port = "6379";

    // Start both connections.
    sub_conn->async_run(cfg, logger{logger::level::disabled},
                        net::consign(net::detached, sub_conn));

    pub_conn->async_run(cfg, logger{logger::level::disabled},
                        net::consign(net::detached, pub_conn));

    std::string channel_name = "chat_room";

    // Start listening for Pub/Sub push messages on the subscriber connection.
    start_receive_loop(sub_conn);

    // Subscribe to the "chat_room" channel.
    auto sub_req  = std::make_shared<request>();
    auto sub_resp = std::make_shared<boost::redis::response<boost::redis::ignore_t>>();

    sub_req->push("SUBSCRIBE", channel_name);

    sub_conn->async_exec(
        *sub_req, *sub_resp,
        [sub_conn, sub_req, sub_resp, channel_name](error_code ec, std::size_t) {
            if (ec) {
                std::cerr << "SUBSCRIBE error: " << ec.message() << "\n";
                sub_conn->cancel();
                return;
            }
            std::cout << "Subscribed to '" << channel_name
                      << "'. Waiting for messages...\n";
        });

int main()
{
    net::io_context ioc;

    // One connection dedicated to SUBSCRIBE (subscriber mode).
    auto sub_conn = std::make_shared<connection>(ioc);

    // Separate connection for regular commands like PUBLISH.
    auto pub_conn = std::make_shared<connection>(ioc);

    config cfg;
    cfg.addr.host = "127.0.0.1";
    cfg.addr.port = "6379";

    // Start both connections.
    sub_conn->async_run(cfg, logger{logger::level::disabled},
                        net::consign(net::detached, sub_conn));

    pub_conn->async_run(cfg, logger{logger::level::disabled},
                        net::consign(net::detached, pub_conn));

    std::string channel_name = "chat_room";

    // Start listening for Pub/Sub push messages on the subscriber connection.
    start_receive_loop(sub_conn);

    // Subscribe to the "chat_room" channel.
    auto sub_req  = std::make_shared<request>();
    auto sub_resp = std::make_shared<boost::redis::response<boost::redis::ignore_t>>();

    sub_req->push("SUBSCRIBE", channel_name);

    sub_conn->async_exec(
        *sub_req, *sub_resp,
        [sub_conn, sub_req, sub_resp, channel_name](error_code ec, std::size_t) {
            if (ec) {
                std::cerr << "SUBSCRIBE error: " << ec.message() << "\n";
                sub_conn->cancel();
                return;
            }
            std::cout << "Subscribed to '" << channel_name
                      << "'. Waiting for messages...\n";
        });

This part sets up the runtime environment and the two Redis connections.

The program creates:

sub_conn for SUBSCRIBE
pub_conn for PUBLISH

This separation is required for Redis Pub/Sub, because a connection in subscriber mode is dedicated to Pub/Sub commands.

Both connections are started with async_run(), which lets them process Redis communication through the io_context.

Next, the code defines the channel name as "chat_room" and starts the receive loop on the subscriber connection.

Then it builds a SUBSCRIBE request and sends it using async_exec(). The response type uses ignore_t because the main Pub/Sub events will be handled through the receive loop rather than decoded as a normal command result.

At this point, the program is connected and ready to receive messages published to the subscribed channel.

Section 3: Publish a JSON message and run the event loop

This final section publishes a message after a short delay and then runs the event loop.

C++

    // After a short delay, publish a JSON message to the "chat_room" channel.
    net::steady_timer timer{ioc};
    timer.expires_after(std::chrono::seconds(1));

    timer.async_wait(
        [pub_conn, channel_name](error_code ec) {
            if (ec) {
                return;
            }

            auto pub_req  = std::make_shared<request>();
            auto pub_resp = std::make_shared<boost::redis::response<int>>();

            // Create a JSON message
            json message = {{"user", "alice"}, {"text", "Hello everyone!"}};
            pub_req->push("PUBLISH", channel_name, message.dump());

            pub_conn->async_exec(
                *pub_req, *pub_resp,
                [pub_conn, pub_req, pub_resp](error_code ec2, std::size_t) {
                    if (ec2) {
                        std::cerr << "PUBLISH error: " << ec2.message() << "\n";
                        pub_conn->cancel();
                        return;
                    }

                    try {
                        int delivered = std::get<0>(*pub_resp).value();
                        std::cout << "Message published, delivered to "
                                  << delivered << " subscriber(s).\n";
                    } catch (std::exception const& e) {
                        std::cerr << "Error decoding PUBLISH response: "
                                  << e.what() << "\n";
                    }

                    // Done with publisher connection.
                    pub_conn->cancel();
                });
        });

    ioc.run();
    return 0;
}

    // After a short delay, publish a JSON message to the "chat_room" channel.
    net::steady_timer timer{ioc};
    timer.expires_after(std::chrono::seconds(1));

    timer.async_wait(
        [pub_conn, channel_name](error_code ec) {
            if (ec) {
                return;
            }

            auto pub_req  = std::make_shared<request>();
            auto pub_resp = std::make_shared<boost::redis::response<int>>();

            // Create a JSON message
            json message = {{"user", "alice"}, {"text", "Hello everyone!"}};
            pub_req->push("PUBLISH", channel_name, message.dump());

            pub_conn->async_exec(
                *pub_req, *pub_resp,
                [pub_conn, pub_req, pub_resp](error_code ec2, std::size_t) {
                    if (ec2) {
                        std::cerr << "PUBLISH error: " << ec2.message() << "\n";
                        pub_conn->cancel();
                        return;
                    }

                    try {
                        int delivered = std::get<0>(*pub_resp).value();
                        std::cout << "Message published, delivered to "
                                  << delivered << " subscriber(s).\n";
                    } catch (std::exception const& e) {
                        std::cerr << "Error decoding PUBLISH response: "
                                  << e.what() << "\n";
                    }

                    // Done with publisher connection.
                    pub_conn->cancel();
                });
        });

    ioc.run();
    return 0;
}

This section sends the actual notification.

A steady_timer delays the publish operation slightly so the subscription can be established first. When the timer fires, the program creates a PUBLISH request and builds a JSON message:

JSON

{"user": "alice", "text": "Hello everyone!"}

{"user": "alice", "text": "Hello everyone!"}

The JSON object is converted to a string with message.dump() and sent to Redis on the chat_room channel.

The response to PUBLISH is an integer representing how many subscribers received the message. The code prints that number, then cancels the publisher connection.

Finally, ioc.run() starts the event loop. This allows all asynchronous operations to execute, including connection handling, subscription, publishing, and receiving the Pub/Sub message.