consumer_base.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/urabbitmq/consumer_base.hpp
/// @brief Base class for your consumers.
 
#include <memory>
 
#include <userver/utils/periodic_task.hpp>
 
#include <userver/urabbitmq/consumer_settings.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace urabbitmq {
 
class Client;
class ConsumerBaseImpl;
 
/// @ingroup userver_base_classes
///
/// @brief Base class for your consumers.
/// You should derive from it and override `Process` method, which gets called
/// when a new message arrives from the broker.
///
/// If your configuration is known upfront and doesn't change ar runtime
/// consider using `ConsumerComponentBase` instead.
///
/// Library takes care of handling start failures and runtime failures
/// (connection breakage/broker node downtime etc.) and will try it's best to
/// restart the consumer.
///
/// @note Since messages are delivered asynchronously in the background you
/// must call `Stop` before derived class is destroyed, otherwise a race is
/// possible, when `Process` is called concurrently with
/// derived class destructor, which is UB.
///
/// @note Library guarantees `at least once` delivery, hence some deduplication
/// might be needed ou your side.
class ConsumerBase {
public:
    ConsumerBase(std::shared_ptr<Client> client, const ConsumerSettings& settings);
    virtual ~ConsumerBase();
 
    /// @brief Start consuming messages from the broker.
    /// Calling this method on running consumer has no effect.
    ///
    /// Should not throw, in case of initial setup failure library will restart
    /// the consumer in the background.
    void Start();
 
    /// @brief Stop consuming messages from the broker.
    /// Calling this method on stopped consumer has no effect.
    ///
    /// @note You must call this method before your derived class is destroyed,
    /// otherwise it's UB.
    void Stop();
 
protected:
    /// @brief You may override this method in derived class and implement
    /// message handling logic. By default it does nothing.
    ///
    /// If this method returns successfully message would be acked (best effort)
    /// to the broker, if this method throws the message would be requeued.
    ///
    /// Please keep in mind that it is possible for the message to be delivered
    /// again even if `Process` returns successfully: sadly we can't guarantee
    /// that `ack` ever reached the broker (network issues or unexpected shutdown,
    /// for example).
    /// It is however guaranteed for message to be requeued if `Process` fails.
    virtual void Process(std::string) { /* do nothing */
    }
 
    /// @brief You may override this method in derived class and implement
    /// message handling logic. By default it just calls `Process(std::string)`
    /// with message body.
    ///
    virtual void Process(ConsumedMessage msg) { Process(std::move(msg.message)); }
 
private:
    std::shared_ptr<Client> client_;
    const ConsumerSettings settings_;
 
    std::unique_ptr<ConsumerBaseImpl> impl_;
    utils::PeriodicTask monitor_{};
};
 
}  // namespace urabbitmq
 
USERVER_NAMESPACE_END