/**
 * Copyright 2016 Yahoo Inc.
 *
 * Licensed 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.
 */

#include "ClientImpl.h"

#include "LogUtils.h"
#include "ConsumerImpl.h"
#include "ProducerImpl.h"
#include "DestinationName.h"
#include "PartitionedProducerImpl.h"
#include "PartitionedConsumerImpl.h"
#include <boost/bind.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <sstream>
#include <openssl/sha.h>
#include "boost/date_time/posix_time/posix_time.hpp"

DECLARE_LOG_OBJECT()

namespace pulsar {

    static const char hexDigits[] = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };

    const std::string generateRandomName() {
        unsigned char hash[SHA_DIGEST_LENGTH];  // == 20;
        boost::posix_time::ptime t(boost::posix_time::microsec_clock::universal_time());
        long nanoSeconds = t.time_of_day().total_nanoseconds();
        std::stringstream ss;
        ss << nanoSeconds;
        SHA1(reinterpret_cast<const unsigned char*>(ss.str().c_str()), ss.str().length(), hash);

        const int nameLength = 6;
        std::stringstream hexHash;
        for (int i = 0; i < nameLength / 2; i++) {
            hexHash << hexDigits[(hash[i] & 0xF0) >> 4];
            hexHash << hexDigits[hash[i] & 0x0F];
        }

        return hexHash.str();
    }
    typedef boost::unique_lock<boost::mutex> Lock;

    ClientImpl::ClientImpl(const std::string& serviceUrl,
                           const ClientConfiguration& clientConfiguration, bool poolConnections)
        : mutex_(),
          state_(Open),
          serviceUrl_(serviceUrl),
          clientConfiguration_(clientConfiguration),
          ioExecutorProvider_(boost::make_shared<ExecutorServiceProvider>(clientConfiguration.getIOThreads())),
          listenerExecutorProvider_(boost::make_shared<ExecutorServiceProvider>(clientConfiguration.getMessageListenerThreads())),
          partitionListenerExecutorProvider_(boost::make_shared<ExecutorServiceProvider>(clientConfiguration.getMessageListenerThreads())),
          pool_(clientConfiguration, ioExecutorProvider_, clientConfiguration.getAuthenticationPtr(), poolConnections),
          lookup_(pool_, serviceUrl),
          producerIdGenerator_(0),
          consumerIdGenerator_(0),
          requestIdGenerator_(0) {
        LogUtils::init(clientConfiguration.getLogConfFilePath());
    }

    ClientImpl::~ClientImpl() {
        shutdown();
    }

    const ClientConfiguration& ClientImpl::conf() const {
        return clientConfiguration_;
    }

    ExecutorServiceProviderPtr ClientImpl::getIOExecutorProvider() {
        return ioExecutorProvider_;
    }

    ExecutorServiceProviderPtr ClientImpl::getListenerExecutorProvider() {
        return listenerExecutorProvider_;
    }

    ExecutorServiceProviderPtr ClientImpl::getPartitionListenerExecutorProvider() {
        return partitionListenerExecutorProvider_;
    }
    void ClientImpl::createProducerAsync(const std::string& topic,
                                         ProducerConfiguration conf,
                                         CreateProducerCallback callback) {
        DestinationNamePtr dn;
        {
            Lock lock(mutex_);
            if (state_ != Open) {
                lock.unlock();
                callback(ResultAlreadyClosed, Producer());
                return;
            } else if (!(dn = DestinationName::get(topic))) {
                lock.unlock();
                callback(ResultInvalidTopicName, Producer());
                return;
            }
        }
        lookup_.getPartitionMetadataAsync(dn).addListener(boost::bind(&ClientImpl::handleCreateProducer,
                                    shared_from_this(), _1, _2, dn, conf, callback));
    }

    void ClientImpl::handleCreateProducer(const Result result,
            const LookupDataResultPtr partitionMetadata,
            DestinationNamePtr dn,
            ProducerConfiguration conf,
            CreateProducerCallback callback) {
        if (!result) {
            ProducerImplBasePtr producer;
            if (partitionMetadata->getPartitions() > 1) {
                producer = boost::make_shared<PartitionedProducerImpl>(shared_from_this(),
                                                                       dn, partitionMetadata->getPartitions(), conf);
            } else {
                producer = boost::make_shared<ProducerImpl>(shared_from_this(), dn->toString(), conf);
            }
            producer->getProducerCreatedFuture().addListener(boost::bind(&ClientImpl::handleProducerCreated,
                                                                         shared_from_this(), _1, _2, callback,
                                                                         producer));
            Lock lock(mutex_);
            producers_.push_back(producer);
            lock.unlock();
            producer->start();
        } else {
            LOG_ERROR("Error Checking/Getting Partition Metadata while creating producer on " << dn->toString() << " -- " << result );
            callback (result, Producer());
        }

    }

    void ClientImpl::handleProducerCreated(Result result, ProducerImplBaseWeakPtr producerBaseWeakPtr,
                                           CreateProducerCallback callback, ProducerImplBasePtr producer) {
        callback(result, Producer(producer));
    }

    void ClientImpl::subscribeAsync(const std::string& topic, const std::string& consumerName,
                                    const ConsumerConfiguration& conf, SubscribeCallback callback) {
        DestinationNamePtr dn;
        {
            Lock lock(mutex_);
            if (state_ != Open) {
                lock.unlock();
                callback(ResultAlreadyClosed, Consumer());
                return;
            } else if (!(dn = DestinationName::get(topic))) {
                lock.unlock();
                callback(ResultInvalidTopicName, Consumer());
                return;
            }
        }

        lookup_.getPartitionMetadataAsync(dn).addListener(boost::bind(&ClientImpl::handleSubscribe,
                                    shared_from_this(), _1, _2, dn, consumerName, conf, callback));
    }

    void ClientImpl::handleSubscribe(const Result result,
            const LookupDataResultPtr partitionMetadata,
            DestinationNamePtr dn,
            const std::string& consumerName,
            ConsumerConfiguration conf,
            SubscribeCallback callback) {
        if (!result) {
            // generate random name if not supplied by the customer.
            if(conf.getConsumerName().empty()) {
                conf.setConsumerName(generateRandomName());
            }
            ConsumerImplBasePtr consumer;
            if (partitionMetadata->getPartitions() > 1) {
                if (conf.getReceiverQueueSize() == 0) {
                    LOG_ERROR("Can't use partitioned topic if the queue size is 0.");
                    callback (ResultInvalidConfiguration, Consumer());
                    return;
                }
                consumer = boost::make_shared<PartitionedConsumerImpl>(shared_from_this(),
                                                                       consumerName,
                                                                       dn,
                                                                       partitionMetadata->getPartitions(),
                                                                       conf);
            } else  {
                consumer = boost::make_shared<ConsumerImpl>(shared_from_this(), dn->toString(),
                                                            consumerName, conf);
            }
            consumer->getConsumerCreatedFuture().addListener(boost::bind(&ClientImpl::handleConsumerCreated,
                                                                         shared_from_this(), _1, _2, callback,
                                                                         consumer));
            Lock lock(mutex_);
            consumers_.push_back(consumer);
            lock.unlock();
            consumer->start();
        } else {
            LOG_ERROR("Error Checking/Getting Partition Metadata while Subscribing- " <<  result );
            callback (result, Consumer());
        }

    }

    void ClientImpl::handleConsumerCreated(Result result,
                                           ConsumerImplBaseWeakPtr consumerImplBaseWeakPtr,
                                           SubscribeCallback callback,
                                           ConsumerImplBasePtr consumer) {
        callback(result, Consumer(consumer));
    }

    Future<Result, ClientConnectionWeakPtr> ClientImpl::getConnection(const std::string& topic) {
        Promise<Result, ClientConnectionWeakPtr> promise;
        lookup_.lookupAsync(topic).addListener(boost::bind(&ClientImpl::handleLookup, this, _1, _2, promise));
        return promise.getFuture();
    }

    void ClientImpl::handleLookup(Result result, LookupDataResultPtr data,
                                      Promise<Result, ClientConnectionWeakPtr> promise) {
            if (data) {
                LOG_DEBUG("Getting connection to broker: " << data->getBrokerUrl());
                Future<Result, ClientConnectionWeakPtr> future = pool_.getConnectionAsync(
                                                                                          data->getBrokerUrl());
                future.addListener(boost::bind(&ClientImpl::handleNewConnection, this, _1, _2, promise));
            } else {
                promise.setFailed(result);
            }
        }

    void ClientImpl::handleNewConnection(Result result, const ClientConnectionWeakPtr& conn,
                                         Promise<Result, ClientConnectionWeakPtr> promise) {
        if (result == ResultOk) {
            promise.setValue(conn);
        } else {
            promise.setFailed(ResultConnectError);
        }
    }

    void ClientImpl::closeAsync(CloseCallback callback) {
        Lock lock(mutex_);
        ProducersList producers(producers_);
        ConsumersList consumers(consumers_);

        if (state_ != Open && callback) {
            lock.unlock();
            callback(ResultAlreadyClosed);
            return;
        }
        // Set the state to Closing so that no producers could get added
        state_ = Closing;
        lock.unlock();

        LOG_DEBUG("Closing Pulsar client");
        SharedInt numberOfOpenHandlers = boost::make_shared<int>(producers.size() + consumers.size());

        for (ProducersList::iterator it = producers.begin(); it != producers.end(); ++it) {
            ProducerImplBasePtr producer = it->lock();
            if (producer && !producer->isClosed()) {
                producer->closeAsync(
                                     boost::bind(&ClientImpl::handleClose, shared_from_this(), _1, numberOfOpenHandlers,
                                                 callback));
            } else {
                // Since the connection is already closed
                (*numberOfOpenHandlers)--;
            }
        }

        for (ConsumersList::iterator it = consumers.begin(); it != consumers.end(); ++it) {
            ConsumerImplBasePtr consumer = it->lock();
            if (consumer && !consumer->isClosed()) {
                consumer->closeAsync(
                                     boost::bind(&ClientImpl::handleClose, shared_from_this(), _1, numberOfOpenHandlers,
                                                 callback));
            } else {
                // Since the connection is already closed
                (*numberOfOpenHandlers)--;
            }
        }

        if (*numberOfOpenHandlers == 0 && callback) {
            callback(ResultOk);
        }
    }

    void ClientImpl::handleClose(Result result, SharedInt numberOfOpenHandlers, ResultCallback callback) {
        static bool errorClosing = false;
        static Result failResult = ResultOk;
        if (result != ResultOk) {
            errorClosing = true;
            failResult = result;
        }
        if(*numberOfOpenHandlers > 0) {
            --(*numberOfOpenHandlers);
        }
        if (*numberOfOpenHandlers == 0) {

            Lock lock(mutex_);
            state_ = Closed;
            lock.unlock();
            if (errorClosing) {
                LOG_DEBUG("Problem in closing client, could not close one or more consumers or producers");
                if (callback) {
                    callback(failResult);
                }
            }

            LOG_DEBUG("Shutting down producers and consumers for client");
            shutdown();
            if (callback) {
                callback(ResultOk);
            }
        }
    }

    void ClientImpl::shutdown() {
        Lock lock(mutex_);
        ProducersList producers;
        ConsumersList consumers;

        producers.swap(producers_);
        consumers.swap(consumers_);
        lock.unlock();

        for (ProducersList::iterator it = producers.begin(); it != producers.end(); ++it) {
            ProducerImplBasePtr producer = it->lock();
            if (producer) {
                producer->shutdown();
            }
        }

        for (ConsumersList::iterator it = consumers.begin(); it != consumers.end(); ++it) {
            ConsumerImplBasePtr consumer = it->lock();
            if (consumer) {
                consumer->shutdown();
            }
        }

        ioExecutorProvider_->close();
        listenerExecutorProvider_->close();
        partitionListenerExecutorProvider_->close();
    }

    uint64_t ClientImpl::newProducerId() {
        Lock lock(mutex_);
        return producerIdGenerator_++;
    }

    uint64_t ClientImpl::newConsumerId() {
        Lock lock(mutex_);
        return consumerIdGenerator_++;
    }

    uint64_t ClientImpl::newRequestId() {
        Lock lock(mutex_);
        return requestIdGenerator_++;
    }

} /* namespace pulsar */