Client.h

#ifndef PULSAR_CLIENT_HPP_
#define PULSAR_CLIENT_HPP_
 
#include <pulsar/ClientConfiguration.h>
#include <pulsar/ConsoleLoggerFactory.h>
#include <pulsar/Consumer.h>
#include <pulsar/FileLoggerFactory.h>
#include <pulsar/Message.h>
#include <pulsar/MessageBuilder.h>
#include <pulsar/Producer.h>
#include <pulsar/Reader.h>
#include <pulsar/Result.h>
#include <pulsar/Schema.h>
#include <pulsar/ServiceInfo.h>
#include <pulsar/ServiceInfoProvider.h>
#include <pulsar/TableView.h>
#include <pulsar/defines.h>
 
#include <memory>
#include <string>
#include <utility>
#include <variant>
#include <vector>
 
namespace pulsar {
typedef std::function<void(Result, Producer)> CreateProducerCallback;
typedef std::function<void(Result, Consumer)> SubscribeCallback;
typedef std::function<void(Result, Reader)> ReaderCallback;
typedef std::function<void(Result, TableView)> TableViewCallback;
typedef std::function<void(Result, const std::vector<std::string>&)> GetPartitionsCallback;
typedef std::function<void(Result)> CloseCallback;
 
using CreateProducerV2Callback = std::function<void(std::variant<Error, Producer>)>;
using CreateConsumerV2Callback = std::function<void(std::variant<Error, Consumer>)>;
using SubscribeV2Callback = CreateConsumerV2Callback;
using ReaderV2Callback = std::function<void(std::variant<Error, Reader>)>;
using TableViewV2Callback = std::function<void(std::variant<Error, TableView>)>;

struct TopicRegex {
    explicit TopicRegex(std::string pattern) : pattern(std::move(pattern)) {}
 
    std::string pattern;
};
 
using SubscribeTopics = std::variant<std::string, std::vector<std::string>, TopicRegex>;
 
class ClientImpl;
class PulsarFriend;
class PulsarWrapper;
 
class PULSAR_PUBLIC Client {
   public:
    Client(const std::string& serviceUrl);

    Client(const std::string& serviceUrl, const ClientConfiguration& clientConfiguration);

    static Client create(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,
                         const ClientConfiguration& clientConfiguration);

    Result createProducer(const std::string& topic, Producer& producer);

    [[deprecated("use createProducerV2")]] Result createProducer(const std::string& topic,
                                                                 const ProducerConfiguration& conf,
                                                                 Producer& producer);

    [[deprecated("use createProducerAsyncV2")]] void createProducerAsync(
        const std::string& topic, const CreateProducerCallback& callback);

    [[deprecated("use createProducerAsyncV2")]] void createProducerAsync(
        const std::string& topic, const ProducerConfiguration& conf, const CreateProducerCallback& callback);
 
    void createProducerAsyncV2(const std::string& topic, const ProducerConfiguration& conf,
                               CreateProducerV2Callback callback);
 
    std::variant<Error, Producer> createProducerV2(const std::string& topic,
                                                   const ProducerConfiguration& conf);

    Result subscribe(const std::string& topic, const std::string& subscriptionName, Consumer& consumer);

    Result subscribe(const std::string& topic, const std::string& subscriptionName,
                     const ConsumerConfiguration& conf, Consumer& consumer);

    void subscribeAsync(const std::string& topic, const std::string& subscriptionName,
                        const SubscribeCallback& callback);

    void subscribeAsync(const std::string& topic, const std::string& subscriptionName,
                        const ConsumerConfiguration& conf, const SubscribeCallback& callback);
 
    void subscribeAsyncV2(const SubscribeTopics& topics, const std::string& subscriptionName,
                          const ConsumerConfiguration& conf, SubscribeV2Callback callback);
 
    std::variant<Error, Consumer> subscribeV2(const SubscribeTopics& topics,
                                              const std::string& subscriptionName,
                                              const ConsumerConfiguration& conf);

    Result subscribe(const std::vector<std::string>& topics, const std::string& subscriptionName,
                     Consumer& consumer);

    Result subscribe(const std::vector<std::string>& topics, const std::string& subscriptionName,
                     const ConsumerConfiguration& conf, Consumer& consumer);

    void subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
                        const SubscribeCallback& callback);

    void subscribeAsync(const std::vector<std::string>& topics, const std::string& subscriptionName,
                        const ConsumerConfiguration& conf, const SubscribeCallback& callback);

    Result subscribeWithRegex(const std::string& regexPattern, const std::string& subscriptionName,
                              Consumer& consumer);

    Result subscribeWithRegex(const std::string& regexPattern, const std::string& subscriptionName,
                              const ConsumerConfiguration& conf, Consumer& consumer);

    void subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
                                 const SubscribeCallback& callback);

    void subscribeWithRegexAsync(const std::string& regexPattern, const std::string& subscriptionName,
                                 const ConsumerConfiguration& conf, const SubscribeCallback& callback);

    Result createReader(const std::string& topic, const MessageId& startMessageId,
                        const ReaderConfiguration& conf, Reader& reader);

    void createReaderAsync(const std::string& topic, const MessageId& startMessageId,
                           const ReaderConfiguration& conf, const ReaderCallback& callback);
 
    void createReaderAsyncV2(const std::string& topic, const MessageId& startMessageId,
                             const ReaderConfiguration& conf, ReaderV2Callback callback);
 
    std::variant<Error, Reader> createReaderV2(const std::string& topic, const MessageId& startMessageId,
                                               const ReaderConfiguration& conf);

    Result createTableView(const std::string& topic, const TableViewConfiguration& conf,
                           TableView& tableView);

    void createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
                              const TableViewCallback& callBack);
 
    void createTableViewAsyncV2(const std::string& topic, const TableViewConfiguration& conf,
                                TableViewV2Callback callback);
 
    std::variant<Error, TableView> createTableViewV2(const std::string& topic,
                                                     const TableViewConfiguration& conf);

    Result getPartitionsForTopic(const std::string& topic, std::vector<std::string>& partitions);

    void getPartitionsForTopicAsync(const std::string& topic, const GetPartitionsCallback& callback);

    Result close();

    void closeAsync(const CloseCallback& callback);

    void shutdown();

    uint64_t getNumberOfProducers();

    uint64_t getNumberOfConsumers();

    void getSchemaInfoAsync(const std::string& topic, int64_t version,
                            std::function<void(Result, const SchemaInfo&)> callback);

    ServiceInfo getServiceInfo() const;
 
   private:
    Client(const std::shared_ptr<ClientImpl>&);
 
    friend class PulsarFriend;
    friend class PulsarWrapper;
    std::shared_ptr<ClientImpl> impl_;
};
}  // namespace pulsar
 
#endif /* PULSAR_CLIENT_HPP_ */