/builddir/build/BUILD/libutil++-1.4.0/src/sella/server/HTTP.cpp

/*
** libutil++
** $Id: HTTP.cpp 1202 2014-10-18 21:14:37Z sella $
** Copyright (c) 2011-2014 Digital Genesis, LLC. All Rights Reserved.
** Released under the LGPL Version 2.1 License.
** http://www.digitalgenesis.com
*/

static const char rcsid[] __attribute__((used)) = "$Id: HTTP.cpp 1202 2014-10-18 21:14:37Z sella $";

#include "HTTP.h"
#include "Code.h"
#include "callbacks.h"
#include "../util/CommonMacro.h"
#include "../util/String.h"
#include "../util/File.h"

#include <cstddef>
#include <string>
#include <algorithm>

using namespace sella::server;
using namespace sella::net;
using namespace sella::util;

const IPAddressPort::vector HTTP::ListenersDefault = { IPAddressPort("0.0.0.0:1080"), IPAddressPort("[::]:1080") };
const std::string HTTP::RootDefault(".");
const std::vector<std::string> HTTP::DirectoryIndexesDefault({ "index.html", "index.htm" });
const std::string HTTP::BannerDefault("Digital Genesis HTTP Server " PACKAGE "/" VERSION);
const std::set<HTTPRequest::Method> HTTP::AllowDefault = { HTTPRequest::Options, HTTPRequest::Get, HTTPRequest::Head, HTTPRequest::Trace };

HTTP::HTTP() throw (ServerException) :
        libutilxx__log(::libutilxx__log),
        running(false),
        listeners(ListenersDefault),
        requestBodyLimit(RequestBodyLimitDefault),
        connectionLimit(ConnectionLimitDefault),
        root(RootDefault),
        indexes(DirectoryIndexesDefault),
        banner(BannerDefault),
        allow(AllowDefault),
        aclCallback(NULL),
        authCallback(NULL),
        reqCallback(http::request_callback),
        errorCallback(http::error_callback)
{ }

HTTP::HTTP(Log::shared log) throw (ServerException) :
        libutilxx__log(log),
        running(false),
        listeners(ListenersDefault),
        requestBodyLimit(RequestBodyLimitDefault),
        connectionLimit(ConnectionLimitDefault),
        root(RootDefault),
        indexes(DirectoryIndexesDefault),
        banner(BannerDefault),
        allow(AllowDefault),
        aclCallback(NULL),
        authCallback(NULL),
        reqCallback(http::request_callback),
        errorCallback(http::error_callback)
{ }

HTTP::~HTTP() throw () {
        if (isRunning()) {
                stop();
        } else {
                shutdown();
        }
}

bool HTTP::receive(void) throw (ServerException) {
        timeval tv = { 0, 0 };

        return receive(&tv);
}

bool HTTP::receive(timeval *tv) throw (ServerException) {
        int c;

        if (servers.empty()) {
                startup();
        }

        if ((c = select(tv)) > 0) {
                while (!pending.empty()) {
                        HTTPConnection::shared connection = pending.front();
                        pending.pop();

                        worker(connection);
                }
        }

        return (c > 0);
}

bool HTTP::run(size_t workers) throw (ServerException) {
        ThreadPool::shared pool = ThreadPool::shared_ptr(workers);

        start(pool);

        ::select(1, NULL, NULL, NULL, NULL);

        return stop();
}

void HTTP::start(size_t workers) throw (ServerException) {
        ThreadPool::shared pool = ThreadPool::shared_ptr(workers);

        start(pool);
}

void HTTP::start(size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers) throw (ServerException) {
        ThreadPool::shared pool = ThreadPool::shared_ptr(maximumWorkers, minimumWorkers, spareWorkers);

        start(pool);
}

void HTTP::start(ThreadPool::shared pool) throw (ServerException) {
        std::lock_guard<std::mutex> lg(mutex);

        if (running) {
                THROW(ServerException, "HTTP server is already running");
        } else if (pool == NULL) {
                THROW(ServerException, "ThreadPool is null");
        }

        startup();
        this->pool = pool;
        running = true;

        try {
                manager = std::thread(&HTTP::server, this);
        } catch (std::exception &e) {
                TRACE(PACKAGE, "HTTP server failed to start");

                this->pool.reset();
                running = false;

                RETHROW_CODE(ServerException, 0, e);
        }

        TRACE(PACKAGE, "HTTP server started");
}

bool HTTP::stop(void) throw () {
        TRACE(PACKAGE, "stop");

        mutex.lock();

        if (running) {
                running = false;

                trigger.set();

                mutex.unlock();

                if (manager.joinable()) {
                        manager.join();
                }

                pool.reset();

                return true;
        }

        mutex.unlock();

        return false;
}

void HTTP::setListeners(const sella::net::IPAddressPort::vector &listeners) throw () {
        if (!listeners.empty()) {
                this->listeners = listeners;
        } else {
                this->listeners = ListenersDefault;
        }
}

void HTTP::setListeners(const std::string &listeners) throw () {
        this->listeners.clear();

        const auto &list = String::split(listeners, ",; ", false, false);

        for (auto l = list.begin(); l != list.end(); ++l) {
                try {
                        this->listeners.push_back(IPAddressPort(*l));
                } catch (Exception &e) {
                        TRACE(PACKAGE, "Bad listener: %s", e.what());
                }
        }

        if (this->listeners.empty()) {
                this->listeners = ListenersDefault;
        }
}

void HTTP::setRequestBodyLimit(size_t requestBodyLimit) throw () {
        if (requestBodyLimit > 0) {
                this->requestBodyLimit = requestBodyLimit;
        } else {
                this->requestBodyLimit = RequestBodyLimitDefault;
        }
}

void HTTP::setConnectionLimit(size_t connectionLimit) throw () {
        if (connectionLimit > 0) {
                this->connectionLimit = connectionLimit;
        } else {
                this->connectionLimit = ConnectionLimitDefault;
        }
}

void HTTP::setRoot(const std::string &root) throw (ServerException) {
        try {
                this->root = File::getRealPath(root);
        } catch (FileException &e) {
                RETHROW(ServerException, e);
        }
}

void HTTP::setDirectoryIndex(const std::vector<std::string> &indexes) throw () {
        this->indexes = indexes;
}

void HTTP::addDirectoryIndex(const std::string &index) throw () {
        if (std::find(indexes.begin(), indexes.end(), index) != indexes.end()) {
                this->indexes.push_back(index);
        }
}

void HTTP::setBanner(const std::string &banner) throw () {
        this->banner = banner;
}

void HTTP::setCustomHeaders(const std::vector<std::string> &headers) throw () {
        this->headers = headers;
}

void HTTP::setAllow(HTTPRequest::Method method) throw () {
        allow.insert(method);
}

void HTTP::setACLCallback(const acl_callback &callback) throw () {
        if (callback) {
                this->aclCallback = callback;
        } else {
                this->aclCallback = NULL;
        }
}

void HTTP::setAuthenticationCallback(const authentication_callback &callback) throw () {
        if (callback) {
                this->authCallback = callback;
        } else {
                this->authCallback = NULL;
        }
}

void HTTP::setRequestCallback(const request_callback &callback) throw () {
        if (callback) {
                this->reqCallback = callback;
        } else {
                this->reqCallback = http::request_callback;
        }
}

void HTTP::setErrorCallback(const error_callback &callback) throw () {
        if (callback) {
                this->errorCallback = callback;
        } else {
                this->errorCallback = http::error_callback;
        }
}

IPAddressPort::vector HTTP::getListeners(void) const throw () {
        return listeners;
}

std::string HTTP::getListenersString(void) const throw () {
        std::string buf;

        if (!listeners.empty()) {
                for (auto l = listeners.begin(); l != listeners.end(); ++l) {
                        buf += l->str() + ",";
                }

                return buf.erase(buf.size() - 1);
        }

        return buf;
}

size_t HTTP::getRequestBodyLimit(void) const throw () {
        return requestBodyLimit;
}

size_t HTTP::getConnectionLimit(void) const throw () {
        return connectionLimit;
}

const std::string& HTTP::getRoot(void) const throw () {
        return root;
}

const std::vector<std::string>& HTTP::getDirectoryIndexes(void) const throw () {
        return indexes;
}

const std::string& HTTP::getBanner(void) const throw () {
        return banner;
}

const std::vector<std::string>& HTTP::getCustomHeaders(void) const throw () {
        return headers;
}

std::set<HTTPRequest::Method> HTTP::getAllow(void) const throw () {
        return allow;
}

void HTTP::clearAllow(HTTPRequest::Method method) throw () {
        allow.erase(method);
}

bool HTTP::isRunning(void) const throw () {
        return running;
}

void HTTP::clear(void) {
        if (isRunning()) {
                stop();
        }

        pool.reset();
        mux.clear();
        servers.clear();
        listeners = ListenersDefault;
        running = false;
        std::queue<HTTPConnection::shared>().swap(pending);
        requestBodyLimit = RequestBodyLimitDefault;
        connectionLimit = ConnectionLimitDefault;
        root = RootDefault;
        indexes = DirectoryIndexesDefault;
        banner = BannerDefault;
        headers.clear();
        allow = AllowDefault;
        aclCallback = NULL;
        authCallback = NULL;
        reqCallback = http::request_callback;
        errorCallback = http::error_callback;
}

void HTTP::print(void) {
        INFO("[HTTP]");
        INFO("  listeners: %s", getListenersString().c_str());
        INFO("  requestBodyLimit: %zd", requestBodyLimit);
        INFO("  connectionLimit: %zd (pending: %zd)", connectionLimit, pending.size());
        INFO("  root: %s", root.c_str());
        INFO("  indexes: %s", String::join(indexes, ",").c_str());
        INFO("  banner: %s", banner.c_str());
        INFO("  headers: %zd", headers.size());
        INFO("  aclCallback: %s", (aclCallback) ? "set" : "null");
        INFO("  authCallback: %s", (authCallback) ? "set" : "null");
        INFO("  reqCallback: %s", (reqCallback) ? "set" : "null");
        INFO("  errorCallback: %s", (errorCallback) ? "set" : "null");
}

const HTTP::shared HTTP::shared_ptr() throw (ServerException) {
        return std::make_shared<HTTP>();
}

const HTTP::shared HTTP::shared_ptr(Log::shared log) throw (ServerException) {
        return std::make_shared<HTTP>(log);
}

void HTTP::startup(void) throw (ServerException) {
        TCPSocket::shared socket;

        mux.insertRead(trigger.getSocket());

        for (auto l = listeners.begin(); l != listeners.end(); ++l) {
                try {
                        socket = TCPSocket::shared_ptr(l->getFamily());
                        socket->setSoReuseAddr();
                        socket->setBlock(true);

                        if (socket->getFamily() == AF_INET6) {
                                socket->setIpv6V6Only(true);
                        }

                        socket->bind(*l);

                        mux.insertRead(socket);
                        servers.insert(socket->getFD());

                        socket->listen();
                } catch (SocketException &e) {
                        RETHROW(ServerException, e);
                }
        }
}

void HTTP::shutdown(void) throw () {
        mux.clear();
        servers.clear();
}

int HTTP::select(timeval *tv) throw (ServerException) {
        int c;

        if ((c = mux.rselect(tv)) > 0) {
                const SocketMux::shared_flist &sockets = mux.getReadList();
                for (auto s = sockets.cbegin(); s != sockets.cend(); ++s) {
                        if (*s == trigger.getSocket()) {
                                trigger.clear();

                                return 0;
                        }

                        try {
                                HTTPConnection::shared connection = accept(*s);
                                        
                                if (pending.size() < connectionLimit) {
                                        pending.push(std::move(connection));
                                } else {
                                        try {
                                                errorCallback(connection, Code::ServiceUnavailable, std::string()); /* Service Unavailable */
                                        } catch (ServerException &e) {
                                                WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                                        }

                                        connection->clear();
                                        c--; /* Reduce new socket count. */
                                }
                        } catch (SocketException &e) {
                                WARNING(e);
                        }
                }
        }

        return c;
}

HTTPConnection::shared HTTP::accept(const Socket::shared &socket) throw (ServerException) {
   try {
                TCPSocket::shared server(std::dynamic_pointer_cast<TCPSocket>(socket));
                TCPSocket::shared client = server->accept();

                TRACE(PACKAGE "-detail", "Incoming HTTP connection from %s", client->getRemoteIPAddressPort().c_str());

                return HTTPConnection::shared_ptr(*this, client);
   } catch (SocketException &e) {
      RETHROW(ServerException, e);
   }
}

void HTTP::process(HTTPConnection::shared &connection) throw () {
        try {
                if (aclCallback && !aclCallback(connection)) {
                        try {   
                                errorCallback(connection, Code::Forbidden, std::string()); /* Forbidden */
                        } catch (ServerException &e) {
                                WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                        }
                } else if (authCallback && !authCallback(connection)) {
                        try {
                                errorCallback(connection, Code::Unauthorized, std::string()); /* Unauthorized */
                        } catch (ServerException &e) {
                                WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                        }
                } else {
                        reqCallback(connection);
                }
        } catch (ServerException &e) {
                try {
                        if (e.getCode() > 0) {
                                errorCallback(connection, e.getCode(), e.getMessage());
                        } else {
                                errorCallback(connection, Code::InternalServerError, std::string()); /* Internal Server Error */
                        }
                } catch (ServerException &e) {
                        WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                }
        } catch (std::exception &e) {
                try {
                        errorCallback(connection, Code::InternalServerError, std::string()); /* Internal Server Error */
                } catch (ServerException &e) {
                        WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                }
        }
}

void HTTP::server(void) throw () {
        int c;
        timeval tv;
        Task::shared task;
        Task::callback function;

        while (running) {
                tv = { DelaySecondsDefault, 0 };

                while ((c = select(&tv)) > 0) {
                        while (!pending.empty()) {
                                HTTPConnection::shared connection = pending.front();
                                pending.pop();

                                task = Task::shared_ptr();
                                function = std::bind(&HTTP::worker, this, connection);
                                task->setCallback(function);

                                pool->schedule(task);
                        }
                }
        }

        TRACE(PACKAGE, "HTTP server stopping...");

        std::lock_guard<std::mutex> lg(mutex);

        shutdown();
        std::queue<HTTPConnection::shared>().swap(pending);

        pool->clear();
        pool->wait();

        TRACE(PACKAGE, "HTTP server stopped");
}

void HTTP::worker(HTTPConnection::shared &connection) throw () {
        process(connection);

        connection->clear();
}

/*
** vim: noet ts=3 sw=3
*/

---