Consumer.h

 
#ifndef CONSUMER_HPP_
#define CONSUMER_HPP_
 
#include <pulsar/BrokerConsumerStats.h>
#include <pulsar/ConsumerConfiguration.h>
#include <pulsar/TypedMessage.h>
#include <pulsar/defines.h>
 
#include <iostream>
 
namespace pulsar {
class PulsarWrapper;
class ConsumerImplBase;
class PulsarFriend;
typedef std::shared_ptr<ConsumerImplBase> ConsumerImplBasePtr;
class PULSAR_PUBLIC Consumer {
   public:
    Consumer();
    virtual ~Consumer() = default;
 
    const std::string& getTopic() const;
 
    const std::string& getSubscriptionName() const;
 
    const std::string& getConsumerName() const;
 
    Result unsubscribe();
 
    void unsubscribeAsync(ResultCallback callback);
 
    Result receive(Message& msg);
 
    template <typename T>
    Result receive(TypedMessage<T>& msg, typename TypedMessage<T>::Decoder decoder) {
        Message rawMsg;
        auto result = receive(rawMsg);
        msg = TypedMessage<T>{rawMsg, decoder};
        return result;
    }
 
    Result receive(Message& msg, int timeoutMs);
 
    template <typename T>
    Result receive(TypedMessage<T>& msg, int timeoutMs, typename TypedMessage<T>::Decoder decoder) {
        Message rawMsg;
        auto result = receive(rawMsg, timeoutMs);
        msg = TypedMessage<T>{rawMsg, decoder};
        return result;
    }
 
    void receiveAsync(ReceiveCallback callback);
 
    template <typename T>
    void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
                      typename TypedMessage<T>::Decoder decoder) {
        receiveAsync([callback, decoder](Result result, const Message& msg) {
            callback(result, TypedMessage<T>{msg, decoder});
        });
    }
 
    Result batchReceive(Messages& msgs);
 
    void batchReceiveAsync(BatchReceiveCallback callback);
 
    Result acknowledge(const Message& message);
 
    Result acknowledge(const MessageId& messageId);
 
    Result acknowledge(const MessageIdList& messageIdList);
 
    void acknowledgeAsync(const Message& message, ResultCallback callback);
 
    void acknowledgeAsync(const MessageId& messageId, ResultCallback callback);
 
    void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback);
 
    Result acknowledgeCumulative(const Message& message);
 
    Result acknowledgeCumulative(const MessageId& messageId);
 
    void acknowledgeCumulativeAsync(const Message& message, ResultCallback callback);
 
    void acknowledgeCumulativeAsync(const MessageId& messageId, ResultCallback callback);
 
    void negativeAcknowledge(const Message& message);
 
    void negativeAcknowledge(const MessageId& messageId);
 
    Result close();
 
    void closeAsync(ResultCallback callback);
 
    Result pauseMessageListener();
 
    Result resumeMessageListener();
 
    void redeliverUnacknowledgedMessages();
 
    Result getBrokerConsumerStats(BrokerConsumerStats& brokerConsumerStats);
 
    void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback);
 
    Result seek(const MessageId& messageId);
 
    Result seek(uint64_t timestamp);
 
    virtual void seekAsync(const MessageId& messageId, ResultCallback callback);
 
    virtual void seekAsync(uint64_t timestamp, ResultCallback callback);
 
    bool isConnected() const;
 
    void getLastMessageIdAsync(GetLastMessageIdCallback callback);
 
    Result getLastMessageId(MessageId& messageId);
 
   private:
    ConsumerImplBasePtr impl_;
    explicit Consumer(ConsumerImplBasePtr);
 
    friend class PulsarFriend;
    friend class PulsarWrapper;
    friend class MultiTopicsConsumerImpl;
    friend class ConsumerImpl;
    friend class ClientImpl;
    friend class ConsumerTest;
};
}  // namespace pulsar
 
#endif /* CONSUMER_HPP_ */