Redis PubSub Messaging

Introduction

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

What You'll Learn

In this lesson, you will learn how to set up and use Redis Pub/Sub messaging to send and receive messages using C++ and Boost.Redis . We'll create a single program that demonstrates both subscribing to and publishing messages on a channel. Important notes: Dedicated connection for subscribers : The subscriber uses a separate Redis connection that operates in subscription mode. Asynchronous event-driven model : We use Boost.Asio's io_context to handle asynchronous operations. RESP3 push messages : Redis sends Pub/Sub messages as push events that we receive through async_receive . Two connections : One for subscribing (enters special subscriber mode) and one for regular commands like PUBLISH.

Setting Up the Connections

Let's start by setting up our includes and creating two Redis connections: C++#include <boost/redis.hpp> #include <boost/redis/src.hpp> // Include in one translation unit only #include <boost/asio.hpp> #include <iostream> #include <memory> #include <chrono> 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; 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)); What's happening here: We create an io_context, which is the event loop that drives all asynchronous operations. We create two separate connections: sub_conn for subscribing and pub_conn for publishing. Why two connections? Once a connection enters subscriber mode (after SUBSCRIBE), it can only receive push messages and cannot execute regular commands. That's why we need a separate connection for PUBLISH. Both connections are started with async_run(), which establishes the connection asynchronously.

Join the 1M+ learners on CodeSignal

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

The Message Receive Loop

Before subscribing, we need to set up a loop that continuously listens for incoming messages:

C++

// 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](boost::system::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;
            }

            // Parse the RESP3 push message.
            // Format: ["message", "<channel>", "<payload>"]
            auto const& nodes = resp->value();

            if (nodes.size() >= 2) {
                std::string kind = nodes[1].value;

                if (kind == "message" && nodes.size() >= 4) {
                    std::string channel = nodes[2].value;
                    std::string payload = nodes[3].value;

                    std::cout << "Received message on channel '" << channel
                              << "': " << payload << "\n";

                    // Stop after receiving the first message.
                    sub_conn->cancel();
                    return;
                }
            }

            // For other push types (like "subscribe" confirmations), keep listening.
            start_receive_loop(sub_conn);
        });
}

Breaking it down:

generic_response can parse any Redis response, including push messages.
set_receive_response() tells the connection where to store the next incoming message.
async_receive() waits asynchronously for Redis to push a message to us.
When a message arrives, we parse the RESP3 format:
- nodes[1].value: The message type ("message", "subscribe", "unsubscribe", etc.)
- nodes[2].value: The channel name
- nodes[3].value: The payload (for "message") or subscriber count (for "subscribe")
If it's a regular message, we print it and stop. Otherwise, we recursively call start_receive_loop() to keep listening.

Important note about production usage: In this example, we stop the subscriber after receiving the first message (sub_conn->cancel()) to keep the demonstration simple. In production systems, you would typically:

Keep the receive loop running continuously to process all incoming messages
Implement graceful shutdown handling (e.g., using signals or shutdown flags)
Add reconnection logic with exponential backoff for handling connection failures
Include proper error handling and logging for long-running operations

This simplified approach helps you understand the core Pub/Sub mechanics without the complexity of production-grade error handling and lifecycle management.

Publishing Messages

Now let's publish a message after a short delay to ensure the subscription is active:

C++

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

    timer.async_wait(
        [pub_conn](boost::system::error_code ec) {
            if (ec) {
                return;
            }

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

            pub_req->push("PUBLISH", "notifications", "Hello, Redis!");

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

                    // PUBLISH returns the number of subscribers that received the message.
                    try {
                        int delivered = std::get<0>(*pub_resp).value();
                        std::cout << "Message published, number of subscribers that received the message: "
                                  << delivered << "\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;
}

What's happening:

We use steady_timer to wait 1 second before publishing. This ensures the subscription is fully active.
We create a PUBLISH command with the channel name "notifications" and payload "Hello, Redis!".
The response type is int because PUBLISH returns the number of subscribers who received the message.
We use std::get<0>(*pub_resp).value() to extract the integer count from the response.
After publishing, we cancel the publisher connection since we're done.
ioc.run() blocks and processes all asynchronous events until all work is complete.

Introduction

What You'll Learn

Setting Up the Connections

Join the 1M+ learners on CodeSignal

Two-Connection Pub/Sub Architecture

The Message Receive Loop

Subscribing to a Channel

Publishing Messages

Running the Program

Key Pub/Sub Concepts

Why Pub/Sub Messaging Matters