#include #include #include #include // Global flag to stop the listener thread std::atomic 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; }