From 822e8c835cd51e743fded8d44f3c90778c38555c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 14 Oct 2026 10:23:31 +0200 Subject: [PATCH] Refactor PubSubServer and RabbitMqServer to improve error handling and session management - Updated PubSubServer to use a unique pointer for PubSubSession, allowing for better error handling during session creation with retries. - Enhanced RabbitMqServer to establish socket connections with improved error handling and logging, ensuring robust consumer setup and message processing. - Refactored message processing logic to handle exceptions more gracefully and ensure proper acknowledgment of messages. --- pubsub/PubSubPlugin/PubSubServer.cpp | 33 ++- rabbitmq/RabbitmqPlugin/RabbitMqServer.cpp | 291 +++++++++++---------- 2 files changed, 176 insertions(+), 148 deletions(-) diff --git a/pubsub/PubSubPlugin/PubSubServer.cpp b/pubsub/PubSubPlugin/PubSubServer.cpp index 7c1ac89..15ba0d7 100644 --- a/pubsub/PubSubPlugin/PubSubServer.cpp +++ b/pubsub/PubSubPlugin/PubSubServer.cpp @@ -187,14 +187,29 @@ namespace Graftcode::Plugins::PubSub "Pub/Sub server: processMessage callback is not configured"); } - PubSubSession session(config); - if (config.autoCreate) { - session.EnsureTopic(config.requestTopic); - session.EnsureSubscription( - config.requestSubscription, - config.requestTopic); + std::unique_ptr session; + while (session == nullptr && + !stopRequested_.load(std::memory_order_acquire)) { + try { + auto candidate = std::make_unique(config); + if (config.autoCreate) { + candidate->EnsureTopic(config.requestTopic); + candidate->EnsureSubscription( + config.requestSubscription, + config.requestTopic); + } + session = std::move(candidate); + } + catch (const std::exception& error) { + if (!stopRequested_.load(std::memory_order_acquire)) { + LogWarn(std::string("setup failed: ") + error.what() + + ", retrying in 2 seconds"); + SleepUnlessStopped( + stopRequested_, + std::chrono::seconds(2)); + } + } } - LogInfo( "receiving subscription '" + config.requestSubscription + "' on topic '" + config.requestTopic + @@ -203,7 +218,7 @@ namespace Graftcode::Plugins::PubSub while (!stopRequested_.load(std::memory_order_acquire)) { try { - const auto batch = session.Pull(config.requestSubscription, 1); + const auto batch = session->Pull(config.requestSubscription, 1); if (batch.empty()) { SleepUnlessStopped( stopRequested_, @@ -214,7 +229,7 @@ namespace Graftcode::Plugins::PubSub if (stopRequested_.load(std::memory_order_acquire)) { break; } - HandleMessage(session, config, processMessage, message); + HandleMessage(*session, config, processMessage, message); } } catch (const std::exception& error) { diff --git a/rabbitmq/RabbitmqPlugin/RabbitMqServer.cpp b/rabbitmq/RabbitmqPlugin/RabbitMqServer.cpp index 43e1fd0..4965889 100644 --- a/rabbitmq/RabbitmqPlugin/RabbitMqServer.cpp +++ b/rabbitmq/RabbitmqPlugin/RabbitMqServer.cpp @@ -447,170 +447,183 @@ void RabbitMqServer::start() { " | Broker: " + config.host + ":" + std::to_string(config.port)); while (!shouldStop()) { - SocketHandle socket = connectSocketToBroker(config); - if (socket == kInvalidSocket) { - if (!shouldStop()) std::this_thread::sleep_for(std::chrono::seconds(2)); - continue; - } - - setActiveSocket(socket); - logInfo("RabbitMQ consumer TCP connection established"); - - ConsumerRuntimeState runtimeState{}; - SocketConnectionHandler handler(socket, runtimeState); - AMQP::Connection connection( - &handler, - AMQP::Login(config.user, config.password), - config.vhost.empty() ? "/" : config.vhost); - AMQP::Channel channel(&connection); - - channel.onError([&](const char* errorMessage) { - runtimeState.connectionError.store(true, std::memory_order_release); - throw std::runtime_error( - std::string("RabbitMQ consumer channel error: ") + (errorMessage ? errorMessage : "unknown")); - }); - - channel.setQos(kRabbitMqPrefetchCount); - auto& consumer = channel.consume(config.queue); - consumer.onSuccess([&](const std::string& consumerTag) { - logInfo("RabbitMQ consumer started with tag: " + consumerTag); - }); - consumer.onError([&](const char* errorMessage) { - runtimeState.connectionError.store(true, std::memory_order_release); - throw std::runtime_error( - std::string("RabbitMQ consumer setup failed: ") + (errorMessage ? errorMessage : "unknown")); - }); - consumer.onReceived([&](const AMQP::Message& message, uint64_t deliveryTag, bool /*redelivered*/) { - const auto messageSize = message.bodySize(); - if (messageSize == 0) { - logWarn("RabbitMQ consumer: received empty message body"); - channel.ack(deliveryTag); - return; + SocketHandle socket = kInvalidSocket; + try { + socket = connectSocketToBroker(config); + if (socket == kInvalidSocket) { + if (!shouldStop()) std::this_thread::sleep_for(std::chrono::seconds(2)); + continue; } - std::vector request(messageSize); - std::memcpy(request.data(), message.body(), messageSize); + setActiveSocket(socket); + logInfo("RabbitMQ consumer TCP connection established"); - try { - const auto processMessage = currentProcessMessage(); - if (processMessage == nullptr) { - throw std::runtime_error("RabbitMQ consumer: processMessage callback is not configured"); + ConsumerRuntimeState runtimeState{}; + SocketConnectionHandler handler(socket, runtimeState); + AMQP::Connection connection( + &handler, + AMQP::Login(config.user, config.password), + config.vhost.empty() ? "/" : config.vhost); + AMQP::Channel channel(&connection); + + channel.onError([&](const char* errorMessage) { + runtimeState.connectionError.store(true, std::memory_order_release); + throw std::runtime_error( + std::string("RabbitMQ consumer channel error: ") + (errorMessage ? errorMessage : "unknown")); + }); + + channel.setQos(kRabbitMqPrefetchCount); + auto& consumer = channel.consume(config.queue); + consumer.onSuccess([&](const std::string& consumerTag) { + logInfo("RabbitMQ consumer started with tag: " + consumerTag); + }); + consumer.onError([&](const char* errorMessage) { + runtimeState.connectionError.store(true, std::memory_order_release); + throw std::runtime_error( + std::string("RabbitMQ consumer setup failed: ") + (errorMessage ? errorMessage : "unknown")); + }); + consumer.onReceived([&](const AMQP::Message& message, uint64_t deliveryTag, bool /*redelivered*/) { + const auto messageSize = message.bodySize(); + if (messageSize == 0) { + logWarn("RabbitMQ consumer: received empty message body"); + channel.ack(deliveryTag); + return; } - std::vector response; - auto writeResponse = [](void* context, const byte* data, std::size_t size) { - auto* out = static_cast*>(context); - if (out == nullptr) { - return; + + std::vector request(messageSize); + std::memcpy(request.data(), message.body(), messageSize); + + try { + const auto processMessage = currentProcessMessage(); + if (processMessage == nullptr) { + throw std::runtime_error("RabbitMQ consumer: processMessage callback is not configured"); + } + std::vector response; + auto writeResponse = [](void* context, const byte* data, std::size_t size) { + auto* out = static_cast*>(context); + if (out == nullptr) { + return; + } + if (data == nullptr || size == 0) { + out->clear(); + return; + } + out->assign(data, data + size); + }; + if (!processMessage( + request.data(), + request.size(), + writeResponse, + &response)) { + throw std::runtime_error("RabbitMQ consumer: processMessage callback returned failure"); } - if (data == nullptr || size == 0) { - out->clear(); + if (!publishResponseIfRequested(channel, message, response)) { + // Requeue when response path failed so request can be retried. + channel.reject(deliveryTag, AMQP::requeue); return; } - out->assign(data, data + size); - }; - if (!processMessage( - request.data(), - request.size(), - writeResponse, - &response)) { - throw std::runtime_error("RabbitMQ consumer: processMessage callback returned failure"); - } - if (!publishResponseIfRequested(channel, message, response)) { - // Requeue when response path failed so request can be retried. - channel.reject(deliveryTag, AMQP::requeue); - return; - } - logDebug( - "RabbitMQ consumer: processed message (" + std::to_string(request.size()) + - "B), response size: " + std::to_string(response.size()) + "B"); - if (message.hasReplyTo() && !message.replyTo().empty()) { logDebug( - "RabbitMQ consumer: response published to reply_to queue: " + message.replyTo()); + "RabbitMQ consumer: processed message (" + std::to_string(request.size()) + + "B), response size: " + std::to_string(response.size()) + "B"); + if (message.hasReplyTo() && !message.replyTo().empty()) { + logDebug( + "RabbitMQ consumer: response published to reply_to queue: " + message.replyTo()); + } + else { + logWarn( + "RabbitMQ consumer: request has no reply_to, response was not published"); + } + channel.ack(deliveryTag); } - else { - logWarn( - "RabbitMQ consumer: request has no reply_to, response was not published"); + catch (const std::exception& ex) { + throw std::runtime_error(std::string("RabbitMQ consumer: processing failed: ") + ex.what()); + channel.reject(deliveryTag); } - channel.ack(deliveryTag); - } - catch (const std::exception& ex) { - throw std::runtime_error(std::string("RabbitMQ consumer: processing failed: ") + ex.what()); - channel.reject(deliveryTag); - } - catch (...) { - throw std::runtime_error("RabbitMQ consumer: processing failed with unknown error"); - channel.reject(deliveryTag); - } - }); - - std::vector receiveBuffer(kReadBufferSize); - std::vector pendingBytes; - pendingBytes.reserve(kReadBufferSize); - - while (!shouldStop() && - !runtimeState.connectionClosed.load(std::memory_order_acquire) && - !runtimeState.connectionError.load(std::memory_order_acquire)) { - const WaitStatus waitStatus = waitForSocketReadable(socket, 1000); - if (waitStatus == WaitStatus::Timeout) { - continue; - } - if (waitStatus == WaitStatus::Error) { - if (!shouldStop()) { - throw std::runtime_error(lastSocketErrorMessage("RabbitMQ consumer: select failed")); + catch (...) { + throw std::runtime_error("RabbitMQ consumer: processing failed with unknown error"); + channel.reject(deliveryTag); } - runtimeState.connectionError.store(true, std::memory_order_release); - break; - } + }); -#ifdef _WIN32 - const int bytesRead = ::recv(socket, receiveBuffer.data(), static_cast(receiveBuffer.size()), 0); -#else - const ssize_t bytesRead = ::recv(socket, receiveBuffer.data(), receiveBuffer.size(), 0); -#endif + std::vector receiveBuffer(kReadBufferSize); + std::vector pendingBytes; + pendingBytes.reserve(kReadBufferSize); - if (bytesRead == 0) { - runtimeState.connectionClosed.store(true, std::memory_order_release); - break; - } - if (bytesRead < 0) { - if (isWouldBlockError()) { + while (!shouldStop() && + !runtimeState.connectionClosed.load(std::memory_order_acquire) && + !runtimeState.connectionError.load(std::memory_order_acquire)) { + const WaitStatus waitStatus = waitForSocketReadable(socket, 1000); + if (waitStatus == WaitStatus::Timeout) { continue; } - if (!shouldStop()) { - throw std::runtime_error(lastSocketErrorMessage("RabbitMQ consumer: recv failed")); + if (waitStatus == WaitStatus::Error) { + if (!shouldStop()) { + throw std::runtime_error(lastSocketErrorMessage("RabbitMQ consumer: select failed")); + } + runtimeState.connectionError.store(true, std::memory_order_release); + break; } - runtimeState.connectionError.store(true, std::memory_order_release); - break; - } - const size_t chunkSize = static_cast(bytesRead); - pendingBytes.insert( - pendingBytes.end(), - receiveBuffer.begin(), - receiveBuffer.begin() + static_cast(chunkSize)); + #ifdef _WIN32 + const int bytesRead = ::recv(socket, receiveBuffer.data(), static_cast(receiveBuffer.size()), 0); + #else + const ssize_t bytesRead = ::recv(socket, receiveBuffer.data(), receiveBuffer.size(), 0); + #endif - while (!pendingBytes.empty()) { - const uint64_t parsed = connection.parse(pendingBytes.data(), pendingBytes.size()); - if (parsed == 0 || parsed > pendingBytes.size()) { + if (bytesRead == 0) { + runtimeState.connectionClosed.store(true, std::memory_order_release); + break; + } + if (bytesRead < 0) { + if (isWouldBlockError()) { + continue; + } + if (!shouldStop()) { + throw std::runtime_error(lastSocketErrorMessage("RabbitMQ consumer: recv failed")); + } + runtimeState.connectionError.store(true, std::memory_order_release); break; } - pendingBytes.erase( - pendingBytes.begin(), - pendingBytes.begin() + static_cast(parsed)); + + const size_t chunkSize = static_cast(bytesRead); + pendingBytes.insert( + pendingBytes.end(), + receiveBuffer.begin(), + receiveBuffer.begin() + static_cast(chunkSize)); + + while (!pendingBytes.empty()) { + const uint64_t parsed = connection.parse(pendingBytes.data(), pendingBytes.size()); + if (parsed == 0 || parsed > pendingBytes.size()) { + break; + } + pendingBytes.erase( + pendingBytes.begin(), + pendingBytes.begin() + static_cast(parsed)); + } } - } - if (runtimeState.connectionError.load(std::memory_order_acquire)) { - connection.fail("RabbitMQ consumer connection failure"); - } + if (runtimeState.connectionError.load(std::memory_order_acquire)) { + connection.fail("RabbitMQ consumer connection failure"); + } - clearActiveSocket(socket); - closeSocket(socket); + clearActiveSocket(socket); + closeSocket(socket); - if (!shouldStop()) { - logWarn("RabbitMQ consumer disconnected, retrying in 2 seconds"); - std::this_thread::sleep_for(std::chrono::seconds(2)); + if (!shouldStop()) { + logWarn("RabbitMQ consumer disconnected, retrying in 2 seconds"); + std::this_thread::sleep_for(std::chrono::seconds(2)); + } + } + catch (const std::exception& ex) { + if (socket != kInvalidSocket) { + clearActiveSocket(socket); + closeSocket(socket); + } + if (!shouldStop()) { + logWarn(std::string("RabbitMQ consumer error: ") + ex.what() + ", retrying in 2 seconds"); + std::this_thread::sleep_for(std::chrono::seconds(2)); + } } }