stream/stream.h

/*--------------------------------------------------------------------------

  This is part of the Socket stream system of the Karoo project.

  (C) 2009 Brian Modra <brian@zwartberg.com>

  This library is free software; you can redistribute them and/or modify
  it under the terms of the GNU Lesser General Public License as published by
  the Free Software Foundation; either version 2.1 of the License,
  or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  See the GNU Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public License
  along with these libraries; if not, write to the Free Software Foundation,
  Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

--------------------------------------------------------------------------*/
#ifndef stream_h
#define stream_h

#ifndef _REENTRANT
#define _REENTRANT
#endif

#include "config.h"
#include <openssl/evp.h>
#include <openssl/ssl.h>
#include <netinet/in.h>
#include "util.h"
#include "queue.h"
#include <vector>
using std::vector;
#include <map>
using std::map;
using std::pair;

namespace karoo {

    enum stream_status {
        STREAM_OK = 0,
        STREAM_UNINITIALISED,
        STREAM_PIPE_FAILED,
        STREAM_PIPE_GET_FLAGS_FAILED,
        STREAM_PIPE_SET_FLAGS_FAILED,
        STREAM_PIPE_OK,
        STREAM_CREATE_FAILED,
        STREAM_SET_OPTIONS_FAILED,
        STREAM_BIND_FAILED,
        STREAM_LISTEN_FAILED,
        STREAM_GET_FLAGS_FAILED,
        STREAM_SET_FLAGS_FAILED,
        STREAM_SOCKET_FAILED,
        STREAM_RESOLVE_NAME_FAILED,
        STREAM_CONNECT_FAILED,
        STREAM_ACCEPT_FAILED,
        STREAM_SSL_FAILED,
    };

    extern const char* stream_status_to_str(enum stream_status status);

    class stream_socket;
    class stream_service;

    class stream_factory {
    protected:
        stream_service* parent;
    public:
        stream_factory(stream_service* parent);
        virtual ~stream_factory();

        virtual stream_socket* createSocket();
    };

    class stream_service : public pebble {
    private:
        mutable karoo_mutex sem;
        enum stream_status status;
        int fh;
        int port;
        exeque* run_queue;
        stream_factory* factory;
        vector<stream_socket*> sockets;
        sigset_t origSigSet;
        sighandler_t prevSigHandler;
        time_t close_linger_time;
        time_t close_inactivity_time;

        int pfd[2];

        mutable karoo_mutex ssl_sem;
        bool has_ssl;
        SSL_CTX* ctx;
        SSL_CTX* client_ctx;

        void clearSSL();


    protected:
        log logger;
    public:
        stream_service(exeque* run_queue, time_t close_linger_time = 15, time_t close_inactivity_time = 45);
        virtual ~stream_service();

        void setFactory(stream_factory* factory);

        bool initialiseSSL(const text& cert_fname, const text& privatekey_fname,  int type);

        bool logSSLErrorStatus(stream_socket* sock, const text& context);

        SSL_CTX* getSSLContext();
        SSL_CTX* getClientSSLContext();

        void pushSocket(stream_socket* sock);
        void eraseSocket(stream_socket* sock);

        enum stream_status startListening(int port);
        void stopListening();

        void setLogger(const log& logger);
        log getLogger();

        void run();

        int getPort() const;

        text describe() const;
    };


    class stream_socket : public pebble {
    protected:
        enum stream_status status;
        text host;
        int port;
        struct sockaddr_in remoteAddr;
        int fh;
        time_t close_pending;
        stream_service* parent;
        stream_buffer write_buffer;
        stream_buffer read_buffer;
        size_t max_read_buffer_size;
        time_t last_network_io;
        SSL* ssl;
        mutable bool needs_read;
        mutable bool needs_write;
        mutable karoo_mutex sem;

        void clearSSL();
    protected:
        friend void stream_service::run();
        size_t getWriteBufferSize() const;
        size_t getReadBufferSize() const;
        //ssize_t takeFromWriteBuffer(char* data, size_t sz);
        ssize_t dumpWriteBuffer(int fh);
        //void addToReadBuffer(char* data, size_t sz);
        ssize_t dumpReadBuffer(int fh);

    public:
        stream_socket(stream_service* parent, size_t max_read_buffer_size = 1024);
        stream_socket(stream_service* parent, const text& host, int port, size_t max_read_buffer_size = 1024);
        virtual ~stream_socket();

        log getLogger();

        void close();
        virtual void scheduleClose();
        time_t closeScheduled() const;

        int initByAccept(int ah);

        int getFileHandle() const;
        enum stream_status getStatus() const;

        ssize_t write(const char* data, size_t sz);
        ssize_t write(const text& data);
        ssize_t read(char* data, size_t sz);
        ssize_t available() const;

        int getPort() const;
        text getHost() const;
        int getRemotePort() const;
        text getRemoteHost() const;
        bool isIncoming() const;
        bool needsRead(bool clear = false) const;
        bool needsWrite(bool clear = false) const;

        void run() {}

        virtual text describe() const;

        time_t getLastNetworkIO() const;
    };

    enum HTTP_parser_state {
        HTTP_PARSER_STATE_REQUEST,
        HTTP_PARSER_STATE_HEADER_KEY,
        HTTP_PARSER_STATE_HEADER_VALUE,
        HTTP_PARSER_STATE_BODY,
    };

    enum HTTP_response_type {
        HTTP_OK                 = 200,
        HTTP_CREATED            = 201,
        HTTP_ACCEPTED           = 202,
        HTTP_NOT_AUTHORITATIVE  = 203,
        HTTP_NO_CONTENT         = 204,
        HTTP_RESET              = 205,
        HTTP_PARTIAL            = 206,
        
        HTTP_MULT_CHOICE        = 300,
        HTTP_MOVED_PERM         = 301,
        HTTP_MOVED_TEMP         = 302,
        HTTP_SEE_OTHER          = 303,
        HTTP_NOT_MODIFIED       = 304,
        HTTP_USE_PROXY          = 305,

        HTTP_BAD_REQUEST        = 400,
        HTTP_UNAUTHORIZED       = 401,
        HTTP_PAYMENT_REQUIRED   = 402,
        HTTP_FORBIDDEN          = 403,
        HTTP_NOT_FOUND          = 404,
        HTTP_BAD_METHOD         = 405,
        HTTP_NOT_ACCEPTABLE     = 406,
        HTTP_PROXY_AUTH         = 407,
        HTTP_CLIENT_TIMEOUT     = 408,
        HTTP_CONFLICT           = 409,
        HTTP_GONE               = 410,
        HTTP_LENGTH_REQUIRED    = 411,
        HTTP_PRECON_FAILED      = 412,
        HTTP_ENTITY_TOO_LARGE   = 413,
        HTTP_REQ_TOO_LONG       = 414,
        HTTP_UNSUPPORTED_TYPE   = 415,
        
        HTTP_INTERNAL_ERROR     = 500,
        HTTP_NOT_IMPLEMENTED    = 501,
        HTTP_BAD_GATEWAY        = 502,
        HTTP_UNAVAILABLE        = 503,
        HTTP_GATEWAY_TIMEOUT    = 504,
        HTTP_VERSION            = 505,
    };

    enum HTTP_method {
        HTTP_METHOD_NOT_SUPPORTED = 0,
        HTTP_METHOD_GET,
        HTTP_METHOD_POST,
        HTTP_METHOD_HEAD,
        HTTP_METHOD_PUT,
        HTTP_METHOD_DELETE,
        HTTP_METHOD_TRACE,
        HTTP_METHOD_OPTIONS,
        HTTP_METHOD_CONNECT,
    };

    class HTTP_factory;
    class HTTP_handler_factory;
    class HTTP_handler;

    class HTTP_parsed_data: public pebble {
    private:
        mutable karoo_mutex sem;
        text request;
        text http_version;
        enum HTTP_method method;
        text path;
        map<text,text>* request_headers;
        stream_buffer* request_body;

        enum HTTP_response_type response_type;
        stream_buffer response_body;
        map<text,text> response_headers;

        stream_socket* parent;
        HTTP_handler_factory* factory;
        HTTP_factory* http_factory;

        bool first;
        bool timeout_pending;
        HTTP_handler* handler;
        uint64_t seq;
    public:
        HTTP_parsed_data(const text& request, map<text,text>* request_headers, char* request_body, size_t request_body_length, HTTP_handler_factory* factory, stream_socket* parent, HTTP_factory* http_factory);
        virtual ~HTTP_parsed_data();

        void run();
        void timedOut();
        text formatResponse();

        log getLogger();

        text getPath() const;
        void setPath(const text& path);
        text getHttpVersion() const;
        void getRequestHeaders(map<text,text>& headers) const;
        void getResponseHeaders(map<text,text>& headers) const;
        enum HTTP_method getMethod() const;
        bool getRequestHeader(const text& key, text& val) const;
        bool getResponseHeader(const text& key, text& val) const;
        void setResponseHeader(const text& key, const text& val);
        void setResponseType(enum HTTP_response_type response_type);
        void writeResponseBody(const char* data, size_t sz);
        void writeResponseBody(const text& str);
        size_t getRequestBodySize() const;
        size_t readRequestBody(char* data, size_t sz);
        void readRequestParams(vector<pair<text,text> >& params);
        ssize_t skipToNextMultipartFormData(const text& boundary);
        HTTP_handler* getHandler();
        uint64_t getSequence() const { return seq; }
        text describe() const;
        //void close();
    };

    class HTTP_handler: public referable {
    protected:
        HTTP_parsed_data* data;
    public:
        HTTP_handler(HTTP_parsed_data* data);
        virtual ~HTTP_handler();

        virtual bool run() = 0; // return true if done

        virtual void timedOut() {} // gtes called if the handler times out.
    };

    class HTTP_handler_factory: public referable {
    protected:
    public:
        HTTP_handler_factory() {}
        virtual ~HTTP_handler_factory() {}

        virtual HTTP_handler* create_HTTP_handler(HTTP_parsed_data* data) = 0;
    };

    class HTTP_parser: public referable {
    private:
        mutable karoo_mutex sem;
        stream_buffer buffer;
        exeque* run_queue;
        enum HTTP_parser_state state;
        text request;
        map<text,text>* headers;
        text key;
        size_t content_length;
        stream_socket* parent;
        HTTP_handler_factory* factory;
        HTTP_factory* http_factory;
    public:
        HTTP_parser(exeque* run_queue, HTTP_handler_factory* factory, HTTP_factory* http_factory);
        virtual ~HTTP_parser();

        bool write(const char* data, size_t sz);

        void setSocket(stream_socket* parent);
        virtual text describe() const;
    };

    class HTTP_factory: public stream_factory {
    private:
        HTTP_handler_factory* handler_factory;
        exeque* run_queue;
        time_t http_timeout;
        khashe http_datas;
        mutable uint64_t http_seq;
        mutable karoo_mutex http_seq_sem;
    public:
        HTTP_factory(exeque* run_queue, stream_service* parent, HTTP_handler_factory* handler_factory, time_t http_timeout);
        virtual ~HTTP_factory();
    
        stream_socket* createSocket();
        exeque* getQueue() { return run_queue; }
        time_t getHTTPTimeout() const { return http_timeout; }
        void pushHTTPData(int64_t seq, HTTP_parsed_data* http_data);
        void eraseHTTPData(int64_t seq);
        bool getHTTPData(int64_t seq, HTTP_parsed_data*& http_data);
        uint64_t getHTTPSequence() const;
    };

    class HTTP_socket: public stream_socket
    {
        HTTP_parser* parser;
    public:
        HTTP_socket(stream_service* parent, HTTP_parser* parser);
        HTTP_socket(stream_service* parent, HTTP_parser* parser, const text& host, int port);
        virtual ~HTTP_socket();
        
        virtual text describe() const;
        //virtual void scheduleClose();
        void run();
    };

    class HTTP_timeout: public pebble {
    private:
        HTTP_factory* parent;
        uint64_t seq;
    public:
        HTTP_timeout(HTTP_factory* parent, uint64_t seq)
        {
            this->parent = parent;
            this->seq = seq;
            dereferenceAfterRun();
        }
        virtual ~HTTP_timeout()
        {
        }
        
        virtual void run()
        {
            HTTP_parsed_data* h;
            if (parent->getHTTPData(seq, h)) {
                h->timedOut();
            }
        }
    };

}

#endif

---