httpconnection.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_HTTP
#include <functional>
#include <promeki/function.h>
#include <promeki/namespace.h>
#include <promeki/objectbase.h>
#include <promeki/error.h>
#include <promeki/buffer.h>
#include <promeki/string.h>
#include <promeki/uniqueptr.h>
#include <promeki/list.h>
#include <promeki/httprequest.h>
#include <promeki/httpresponse.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class TcpSocket;
class EventLoop;
 
class HttpConnection : public ObjectBase {
                PROMEKI_OBJECT(HttpConnection, ObjectBase)
        public:
                using List = ::promeki::List<HttpConnection *>;
 
                static constexpr unsigned int DefaultIdleTimeoutMs = 60'000;
 
                static constexpr int64_t DefaultMaxBodyBytes = 64 * 1024 * 1024;
 
                using RequestHandler = Function<void(HttpRequest &request, HttpResponse &response)>;
 
                HttpConnection(TcpSocket *socket, ObjectBase *parent = nullptr);
 
                void setNeedsServerHandshake();
 
                ~HttpConnection() override;
 
                void setRequestHandler(RequestHandler handler);
 
                Error start();
 
                void close();
 
                bool isOpen() const { return _state != State::Closed; }
 
                String peerAddress() const;
 
                // ----------------------------------------------------
                // Limits / tuning
                // ----------------------------------------------------
 
                void setIdleTimeoutMs(unsigned int ms) { _idleTimeoutMs = ms; }
 
                void setMaxBodyBytes(int64_t bytes) { _maxBodyBytes = bytes; }
 
                Error postResponse(HttpResponse response);
 
                PROMEKI_SIGNAL(requestReceived, HttpRequest);
 
                PROMEKI_SIGNAL(responseSent, HttpRequest, HttpResponse);
 
                PROMEKI_SIGNAL(closed);
 
                PROMEKI_SIGNAL(errorOccurred, Error);
 
        private:
                // High-level state machine.  We never re-pipeline more
                // than one in-flight request; once parsed we wait for
                // the dispatcher to populate a response, drain it,
                // then either resume reading (keep-alive) or close.
                enum class State {
                        Handshaking,      
                        Reading,          
                        AwaitingResponse, 
                        Writing,          
                        Closing,          
                        Closed
                };
 
                struct Impl; 
                UniquePtr<Impl> _impl;
 
                TcpSocket *_socket = nullptr; 
                EventLoop *_loop = nullptr;
                int        _ioHandle = -1;
                int        _timerId = -1;
                State      _state = State::Reading;
 
                RequestHandler _handler;
                Buffer         _readBuf;    
                Buffer         _writeQueue; 
                size_t         _writeOffset = 0;
 
                HttpRequest      _pendingRequest;
                HttpRequest      _lastRequest; 
                IODevice::Shared _streamSource;
                int64_t          _streamRemaining = -1;
                bool             _streamChunked = false;
                bool             _keepAlive = true;
 
                // Async-read parking: when the body IODevice returns
                // read()==0 with atEnd()==false, the pump unsubscribes
                // from IoWrite and waits on the device's readyRead
                // signal.  _streamReadyReadSlotId is the slot id from
                // Signal::connect — non-negative while parked.
                size_t _streamReadyReadSlotId = 0;
                bool   _streamReadyReadConnected = false;
                bool   _streamParked = false;
 
                // Protocol upgrade.  When the response carries an
                // upgrade hook (HttpResponse::upgradeHook()) and is a
                // 101 Switching Protocols, _pendingUpgradeHook is set
                // here and fires once the response finishes writing.
                HttpResponse::UpgradeHook _pendingUpgradeHook;
 
                unsigned int _idleTimeoutMs = DefaultIdleTimeoutMs;
                int64_t      _maxBodyBytes = DefaultMaxBodyBytes;
                int64_t      _bodyBytesSoFar = 0;
                bool         _needsServerHandshake = false;
 
                // Header-collection scratch.  llhttp emits header
                // fields and values as separate callbacks (and may
                // fragment a single field across multiple callbacks)
                // so we accumulate into these strings until both
                // sides are complete, then push into _pendingRequest.
                String _hdrField;
                String _hdrValue;
                bool   _hdrFieldComplete = false;
                bool   _hdrValueComplete = false;
                String _urlBuf;
 
                void onIoReady(int fd, uint32_t events);
                void onIdleTimeout();
                void readSome();
                void pumpWrite();
                void deliverRequest();
                void enqueueResponse(HttpResponse response);
                void scheduleStreamPump();
                void completeProtocolUpgrade();
 
                // Async-read parking: hook the stream's readyRead so
                // pumpWrite resumes when the producer pushes more
                // bytes; detach when the stream finishes or the
                // connection closes.  Both are no-ops if the stream
                // isn't a sequential async device.
                void attachStreamReadyRead();
                void detachStreamReadyRead();
                void onStreamReadyRead();
 
                void flushPendingHeaderPair();
                void resetForNextRequest();
 
                // llhttp callbacks (defined in cpp; bridge via parser->data).
                static int cbMessageBegin(void *parser);
                static int cbUrl(void *parser, const char *at, size_t len);
                static int cbHeaderField(void *parser, const char *at, size_t len);
                static int cbHeaderValue(void *parser, const char *at, size_t len);
                static int cbHeadersComplete(void *parser);
                static int cbBody(void *parser, const char *at, size_t len);
                static int cbMessageComplete(void *parser);
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_HTTP