/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
#ifndef LIB_CLIENTIMPL_H_
#define LIB_CLIENTIMPL_H_

#include <pulsar/Client.h>
#include "ExecutorService.h"
#include "BinaryProtoLookupService.h"
#include "MemoryLimitController.h"
#include "ConnectionPool.h"
#include "LookupDataResult.h"
#include <mutex>
#include <lib/TopicName.h>
#include "ProducerImplBase.h"
#include "ConsumerImplBase.h"
#include <atomic>
#include <vector>

namespace pulsar {

class ClientImpl;
class PulsarFriend;
typedef std::shared_ptr<ClientImpl> ClientImplPtr;
typedef std::weak_ptr<ClientImpl> ClientImplWeakPtr;

class ReaderImpl;
typedef std::shared_ptr<ReaderImpl> ReaderImplPtr;
typedef std::weak_ptr<ReaderImpl> ReaderImplWeakPtr;

std::string generateRandomName();

class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
   public:
    ClientImpl(const std::string& serviceUrl, const ClientConfiguration& clientConfiguration,
               bool poolConnections);
    ~ClientImpl();

    void createProducerAsync(const std::string& topic, ProducerConfiguration conf,
                             CreateProducerCallback callback);

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

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

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

    void createReaderAsync(const std::string& topic, const MessageId& startMessageId,
                           const ReaderConfiguration& conf, ReaderCallback callback);

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

    Future<Result, ClientConnectionWeakPtr> getConnection(const std::string& topic);
    void handleLookup(Result result, LookupDataResultPtr data,
                      Promise<Result, ClientConnectionWeakPtr> promise);
    void handleNewConnection(Result result, const ClientConnectionWeakPtr& conn,
                             Promise<Result, ClientConnectionWeakPtr> promise);

    void closeAsync(CloseCallback callback);
    void shutdown();

    MemoryLimitController& getMemoryLimitController();

    uint64_t newProducerId();
    uint64_t newConsumerId();
    uint64_t newRequestId();

    uint64_t getNumberOfProducers();
    uint64_t getNumberOfConsumers();

    const ClientConfiguration& getClientConfig() const;

    const ClientConfiguration& conf() const;
    ExecutorServiceProviderPtr getIOExecutorProvider();
    ExecutorServiceProviderPtr getListenerExecutorProvider();
    ExecutorServiceProviderPtr getPartitionListenerExecutorProvider();
    LookupServicePtr getLookup();
    friend class PulsarFriend;

   private:
    void handleCreateProducer(const Result result, const LookupDataResultPtr partitionMetadata,
                              TopicNamePtr topicName, ProducerConfiguration conf,
                              CreateProducerCallback callback);

    void handleSubscribe(const Result result, const LookupDataResultPtr partitionMetadata,
                         TopicNamePtr topicName, const std::string& consumerName, ConsumerConfiguration conf,
                         SubscribeCallback callback);

    void handleReaderMetadataLookup(const Result result, const LookupDataResultPtr partitionMetadata,
                                    TopicNamePtr topicName, MessageId startMessageId,
                                    ReaderConfiguration conf, ReaderCallback callback);

    void handleGetPartitions(const Result result, const LookupDataResultPtr partitionMetadata,
                             TopicNamePtr topicName, GetPartitionsCallback callback);

    void handleProducerCreated(Result result, ProducerImplBaseWeakPtr producerWeakPtr,
                               CreateProducerCallback callback, ProducerImplBasePtr producer);
    void handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr consumerWeakPtr,
                               SubscribeCallback callback, ConsumerImplBasePtr consumer);

    typedef std::shared_ptr<int> SharedInt;

    void handleClose(Result result, SharedInt remaining, ResultCallback callback);

    void createPatternMultiTopicsConsumer(const Result result, const NamespaceTopicsPtr topics,
                                          const std::string& regexPattern, const std::string& consumerName,
                                          const ConsumerConfiguration& conf, SubscribeCallback callback);

    enum State
    {
        Open,
        Closing,
        Closed
    };

    std::mutex mutex_;

    State state_;
    std::string serviceUrl_;
    ClientConfiguration clientConfiguration_;
    MemoryLimitController memoryLimitController_;

    ExecutorServiceProviderPtr ioExecutorProvider_;
    ExecutorServiceProviderPtr listenerExecutorProvider_;
    ExecutorServiceProviderPtr partitionListenerExecutorProvider_;

    LookupServicePtr lookupServicePtr_;
    ConnectionPool pool_;

    uint64_t producerIdGenerator_;
    uint64_t consumerIdGenerator_;
    uint64_t requestIdGenerator_;

    typedef std::vector<ProducerImplBaseWeakPtr> ProducersList;
    ProducersList producers_;

    typedef std::vector<ConsumerImplBaseWeakPtr> ConsumersList;
    ConsumersList consumers_;

    std::atomic<Result> closingError;

    friend class Client;
};
} /* namespace pulsar */

#endif /* LIB_CLIENTIMPL_H_ */