service.cpp

#include <ldp_service.hpp>

#include <event2/buffer.h>
#include <event2/event.h>
#include <event2/http.h>
#include <event2/listener.h>

#include <yyjson.h>

#include <cstdlib>
#include <sstream>

namespace httpfifo
{
    class RequestHandler
    {
    protected:
        void pushMessage(evhttp_request *req, LdpMessageQueueSink::Ptr &sink)
        {
            // Step 1: read the request
            evbuffer *inputBuffer(evhttp_request_get_input_buffer(req));
            evbuffer_iovec inputVector;
            std::unique_ptr<yyjson_doc, decltype(&yyjson_doc_free)> jsonDoc(nullptr, &yyjson_doc_free);

            LDP_DEBUG("httpfifo::RequestHandler::pushMessage: reading message from request");
            if (evbuffer_peek(inputBuffer, -1, nullptr, &inputVector, 1) == 1)
            {
                jsonDoc.reset(yyjson_read(reinterpret_cast<const char *>(inputVector.iov_base), inputVector.iov_len, 0));
            }

            // step 2: parse the request
            const bool readOk = (jsonDoc != nullptr);
            yyjson_val *jsonDocRoot(nullptr);
            yyjson_val *messageVal(nullptr);
            const char *messageBuffer(nullptr);
            size_t messageLength(0);

            if (readOk)
            {
                LDP_DEBUG("httpfifo::RequestHandler::pushMessage: parsing message from request");
                jsonDocRoot = yyjson_doc_get_root(jsonDoc.get());
                if (jsonDocRoot && yyjson_is_obj(jsonDocRoot))
                {
                    messageVal = yyjson_obj_get(jsonDocRoot, "message");
                    if (messageVal && yyjson_is_str(messageVal))
                    {
                        messageBuffer = yyjson_get_str(messageVal);
                        messageLength = yyjson_get_len(messageVal);
                    }
                }
            }

            // step 3: publish the message to the sink
            const bool parseOk = (messageBuffer != nullptr && messageLength > 0);

            std::function<bool()> publish = [&]
            {
                LDP_DEBUG("httpfifo::RequestHandler::pushMessage: publishing message to sink");
                return sink->send(reinterpret_cast<const uint8_t *>(messageBuffer), messageLength);
            };

            const bool publishOk = (parseOk && publish());

            if (readOk && parseOk && publishOk)
            {
                LDP_DEBUG("httpfifo::RequestHandler::pushMessage: message published successfully");
                evhttp_send_reply(req, HTTP_OK, "OK", nullptr);
            }
            else if (readOk && parseOk && !publishOk)
            {
                LDP_ERROR("httpfifo::RequestHandler::pushMessage: failed to publish message to sink");
                evhttp_send_error(req, HTTP_INTERNAL, "Internal Error");
            }
            else
            {
                LDP_ERROR("httpfifo::RequestHandler::pushMessage: failed to read or parse message from request");
                evhttp_send_error(req, HTTP_BADREQUEST, "Bad Request");
            }
        }

        void popMessage(LdpMessageQueueSource::Ptr &source, evhttp_request *req)
        {
            std::unique_ptr<yyjson_mut_doc, decltype(&yyjson_mut_doc_free)> jsonDoc(yyjson_mut_doc_new(nullptr), &yyjson_mut_doc_free);
            size_t jsonLen(0);
            std::unique_ptr<char, decltype(&free)> jsonStr(nullptr, &free);
            bool timeout(true);

            // step 1: poll the source for a message
            if (source->poll([&](const uint8_t *data, size_t size, LdpMessageDelivery &delivery)
                             {
                LDP_DEBUG("httpfifo::RequestHandler::popMessage: message received from source, preparing response");
                timeout = false;
                yyjson_mut_val *root(yyjson_mut_obj(jsonDoc.get()));
                if (root && yyjson_mut_obj_add_strcpy(jsonDoc.get(), root, "message", std::string(reinterpret_cast<const char *>(data), size).c_str()))
                {
                    yyjson_mut_doc_set_root(jsonDoc.get(), root);
                    jsonStr.reset(yyjson_mut_write_opts(jsonDoc.get(), 0, nullptr, &jsonLen, nullptr));
                    delivery.acknowledge();
                }
                else
                {
                    LDP_ERROR("httpfifo::RequestHandler::popMessage: failed to create JSON response");
                    delivery.reject(true);
                } 
            }, std::chrono::milliseconds(100)))
            {
                if (!timeout)
                {
                    // step 2: send the response
                    LDP_DEBUG("httpfifo::RequestHandler::popMessage: sending message response");
                    std::unique_ptr<evbuffer, decltype(&evbuffer_free)> buffer(evbuffer_new(), &evbuffer_free);
                    evbuffer_add(buffer.get(), jsonStr.get(), jsonLen);
                    evhttp_send_reply(req, HTTP_OK, "OK", buffer.get());
                }
                else
                {
                    LDP_ERROR("httpfifo::RequestHandler::popMessage: message poll timed out");
                    evhttp_send_error(req, HTTP_NOTFOUND, "Not Found");
                }
            }
            else
            {
                LDP_ERROR("httpfifo::RequestHandler::popMessage: failed to poll message from source");
                evhttp_send_error(req, HTTP_INTERNAL, "Internal Error");
            }
        }

    public:
        virtual void handle(evhttp_request *req) = 0;
    };

    class GetRequestHandler : public RequestHandler
    {
    private:
        LdpMessageQueue::SourcePool &_sourcePool;

    public:
        GetRequestHandler(LdpMessageQueue::SourcePool::Ptr &sourcePool)
            : _sourcePool(*sourcePool) {}
        void handle(evhttp_request *req) override
        {
            LDP_DEBUG("httpfifo::GetRequestHandler::handle: acquiring source from pool");
            LdpMessageQueueSource::Ptr source = _sourcePool.acquire(std::chrono::milliseconds(100));
            if (source)
            {
                bool timeout(true);
                LDP_DEBUG("httpfifo::GetRequestHandler::handle: tranferring message from source to response");
                popMessage(source, req);
            }
            else
            {
                LDP_ERROR("httpfifo::GetRequestHandler::handle: failed to acquire source from pool");
            }
        }
    };

    class PostRequestHandler : public RequestHandler
    {
    private:
        LdpMessageQueue::SinkPool &_sinkPool;

    public:
        PostRequestHandler(LdpMessageQueue::SinkPool::Ptr &sinkPool)
            : _sinkPool(*sinkPool) {}
        void handle(evhttp_request *req) override
        {
            LDP_DEBUG("httpfifo::PostRequestHandler::handle: acquiring sink from pool");
            LdpMessageQueueSink::Ptr sink = _sinkPool.acquire(std::chrono::milliseconds(100));
            if (sink)
            {
                bool timeout(true);
                LDP_DEBUG("httpfifo::PostRequestHandler::handle: tranferring message from request to sink");
                pushMessage(req, sink);
            }
            else
            {
                LDP_ERROR("httpfifo::PostRequestHandler::handle: failed to acquire sink from pool");
            }
        }
    };

    class Service : public ldp::Service
    {
    private:
        // resources
        LdpMessageQueue::Ptr _messageQueue;
        LdpMessageQueue::SourcePool::Ptr _sourcePool;
        LdpMessageQueue::SinkPool::Ptr _sinkPool;
        
        bool connectResources(LdpConnector &connector)
        {
            bool result(false);
            LDP_DEBUG("httpfifo::Service::connectResources: acquiring message queue resource");
            _messageQueue = connector.acquireMessageQueue("mailbox");
            if (_messageQueue)
            {
                LDP_DEBUG("httpfifo::Service::connectResources: creating source and sink pools");
                _sourcePool = _messageQueue->sourcePool(10);
                _sinkPool = _messageQueue->sinkPool(10);
                result = (_sourcePool && _sinkPool);
                if (!result)
                {
                    LDP_ERROR("httpfifo::Service::connectResources: failed to create source and sink pools");
                }
            }
            else
            {
                LDP_ERROR("httpfifo::Service::connectResources: failed to acquire message queue resource");
            }
            return result;
        }

        // server
        std::unique_ptr<httpfifo::RequestHandler> _getHandler;
        std::unique_ptr<httpfifo::RequestHandler> _postHandler;
        std::unique_ptr<event_base, decltype(&event_base_free)> _eventBase{nullptr, &event_base_free};
        std::unique_ptr<evhttp, decltype(&evhttp_free)> _httpServer{nullptr, &evhttp_free};

        bool connectServer(const LdpConfig::Object &settings)
        {
            bool result(false);
            LdpConfig::ObjectReader reader(settings);
            const std::string host = reader.getString("host").value_or("0.0.0.0");
            const int port = reader.getInt("port").value_or(8080);

            _eventBase.reset(event_base_new());
            _httpServer.reset(evhttp_new(_eventBase.get()));

            LDP_DEBUG("httpfifo::Service::connectServer: binding server to %s:%d", host.c_str(), port);
            const int bindResult = evhttp_bind_socket(_httpServer.get(), host.c_str(), port);
            if (bindResult == 0)
            {
                LDP_DEBUG("httpfifo::Service::connectServer: setting request callback");
                evhttp_set_gencb(_httpServer.get(), requestCallback, this);
                result = true;
            }
            else
            {
                LDP_ERROR("httpfifo::Service::connectServer: failed to bind server(%d)", bindResult);
            }

            return result;
        }

        static void requestCallback(evhttp_request *req, void *arg)
        {
            static_cast<httpfifo::Service *>(arg)->handleRequest(req);
        }

        void handleRequest(evhttp_request *req)
        {
            switch (evhttp_request_get_command(req))
            {
            case EVHTTP_REQ_GET:
                _getHandler->handle(req);
                break;
            case EVHTTP_REQ_POST:
                _postHandler->handle(req);
                break;
            default:
                evhttp_send_error(req, HTTP_BADMETHOD, "Unsupported request method");
                break;
            }
        }

    public:
        bool connect(LdpConnector &connector, const LdpConfig::Object &settings) override
        {
            bool result(false);

            LDP_DEBUG("httpfifo::Service::connect: connecting resources");
            if (connectResources(connector))
            {
                LDP_DEBUG("httpfifo::Service::connect: connecting server");
                _getHandler = std::make_unique<httpfifo::GetRequestHandler>(_sourcePool);
                _postHandler = std::make_unique<httpfifo::PostRequestHandler>(_sinkPool);
                result = connectServer(settings);
                if (!result)
                {
                    LDP_ERROR("httpfifo::Service::connect: failed to connect server");
                }
            }
            else
            {
                LDP_ERROR("httpfifo::Service::connect: failed to connect resources");
            }

            return result;
        }

        int launch() override
        {
            LDP_DEBUG("httpfifo::Service::launch: starting event loop");
            return event_base_dispatch(_eventBase.get());
        }

        void shutdown() override
        {
            LDP_DEBUG("httpfifo::Service::shutdown: stopping event loop");
            event_base_loopbreak(_eventBase.get());
        }
    };

}

LDP_EXPORT_SERVICE(httpfifo::Service)