/**
 * Copyright 2014 (c) Baidu, 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.
 */
#ifndef BAIDU_BOS_CPPSDK_HTTP_HTTP_CLIENT_H
#define BAIDU_BOS_CPPSDK_HTTP_HTTP_CLIENT_H

#include "bcesdk/http/http_request.h"
#include "bcesdk/http/http_response.h"
#include "bcesdk/model/bce_request.h"
#include <map>
#include <queue>
#include <atomic>
#include <curl/curl.h>
#include <mutex>
#include <condition_variable>
#include <cassert>

BEGIN_CPPSDK_NAMESPACE

struct HttpRequestContext {
    HttpRequestContext() : request(NULL), response(NULL), rc(0), retry(0), timestamp_ms(0),
        data(NULL) {}
    HttpRequest *request;
    HttpResponse *response;
    int rc;
    int retry;
    int64_t timestamp_ms;

    // attach the caller's custom data pointer
    void *data;
};


class CurlPool {
public:
    CurlPool() : _shutdown(false) {}
    CURL* get() {
        std::unique_lock<std::mutex> lck(_lck);
        while (!_shutdown.load() && _curls.size() == 0) {
            _cv.wait(lck, [&]() { return _shutdown.load() || _curls.size() > 0; });
        }
        assert(!_shutdown.load());
        CURL* curl = _curls.front();
        _curls.pop();
        return curl;
    }

    void release(CURL* curl) {
        std::unique_lock<std::mutex> lck(_lck);
        _curls.push(curl);
        lck.unlock();
        _cv.notify_one();
    }

    bool has_available() {
        std::lock_guard<std::mutex> lck(_lck);
        return _curls.size() > 0 && !_shutdown.load();
    }

    std::vector<CURL*> shutdown(size_t cnt) {
        std::vector<CURL*> curls;
        std::unique_lock<std::mutex> lck(_lck);
        _shutdown = true;
        while (_curls.size() < cnt) {
            _cv.wait(lck, [&]() { return _curls.size() == cnt; });
        }
        while (!_curls.empty()) {
            curls.push_back(_curls.front());
            _curls.pop();
        } 
        return curls;
    }

    int64_t size() {
        std::unique_lock<std::mutex> lck(_lck);
        return _curls.size();
    }

private:
    std::queue<CURL*> _curls;
    std::mutex _lck;
    std::condition_variable _cv;
    std::atomic<bool> _shutdown;
};

class EasyCurlPool {
public:
    EasyCurlPool(int64_t max_size) :
            _max_pool_size(max_size),
            _pool_size(0) {}

    ~EasyCurlPool() {
        for (CURL *curl : _pool.shutdown(_pool_size)) {
            curl_easy_cleanup(curl);
        }
    }

    CURL* get() {
        if (!_pool.has_available()) {
            grow_up();
        }
        CURL *curl = _pool.get();
        return curl;
    }

    void release(CURL *curl) {
        if (curl) {
            curl_easy_reset(curl);
            _pool.release(curl);
        }
    }

    int64_t size() {
        return _pool.size();
    }

    bool fillin() {
        std::lock_guard<std::mutex> guard(_lck);
        int64_t success_added = 0;
        for (int i = 0; i < _max_pool_size; ++i) {
            CURL *curl = curl_easy_init();
            if (curl) {
                _pool.release(curl);
                ++success_added;
            } else {
                break;
            }
        }
        _pool_size += success_added;
        return _pool_size == success_added;
    }

private:
    bool grow_up() {
        std::lock_guard<std::mutex> guard(_lck);
        if (_pool_size < _max_pool_size) {
            unsigned add_base = _pool_size > 0 ? _pool_size : 1;
            unsigned all_to_add = (add_base * 2) < (_max_pool_size - _pool_size) ?
                                 (add_base * 2) : (_max_pool_size - _pool_size);

            int64_t success_added = 0;
            for (unsigned i = 0; i < all_to_add; ++i) {
                CURL *curl = curl_easy_init();
                if (curl) {
                    _pool.release(curl);
                    ++success_added;
                } else {
                    break;
                }
            }
            _pool_size += success_added;
            return success_added > 0;
        }
        return false;
    }
    
private:
    CurlPool _pool;
    int64_t _max_pool_size;
    int64_t _pool_size;
    std::mutex _lck;
};

class HttpClient {
public:
    HttpClient() : _easy_pool(nullptr), _options(nullptr) {}
    void *prepare_curl(HttpRequest &request, HttpResponse *response);
    void *prepare_curl(void *curl_handle, HttpRequest &request, HttpResponse *response);
    int execute(HttpRequest &request, HttpResponse *response);
    void set_curl_pool(EasyCurlPool* easy_pool);
    void set_options(ClientOptions* options);
private:
    static size_t write_stream(void *ptr, size_t size, size_t nmemb, void *stream);
    static size_t read_stream(char *ptr, size_t size, size_t nmemb, void *stream);
    static size_t recv_header_line(void* ptr, size_t size, size_t nmemb, void* user_data);
    static int progress_callback(void *clientp,
                                curl_off_t dltotal,
                                curl_off_t dlnow,
                                curl_off_t ultotal,
                                curl_off_t ulnow); // It is a progress callback function merely used for canceling a request.
private:
    EasyCurlPool* _easy_pool;
    ClientOptions* _options;
};

typedef void (*DetachContextCallback)(HttpRequestContext *);
typedef std::map<void *, HttpRequestContext *> HandleContextMap;

class HttpReactor {
public:
    HttpReactor();
    ~HttpReactor();

    int execute(HttpRequestContext *ctx);

    HttpRequestContext *perform();
    void wait();

    void set_detach_context_callback(DetachContextCallback cb) {
        _detach_context_callback = cb;
    }
    void set_options(ClientOptions* options);
private:
    HttpClient _client;
    void *_mcurl;
    HandleContextMap _handle_ctx;
    std::deque<void *> _idle_handles;
    DetachContextCallback _detach_context_callback;
    ClientOptions* _options;
};


END_CPPSDK_NAMESPACE
#endif