surf/surf.h

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

  This is part of the Multithreading 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

--------------------------------------------------------------------------*/
#include "queue.h"
#include "rock.h"
#include "stream.h"
#include "fab.h"
#include "util.h"
#include <syslog.h>
#include <iostream>
#include <stdlib.h>
#include <sstream>
#include <fstream>

using namespace karoo;
using std::ifstream;

#define SURF_POLL_MESSAGE               10

class surf : public rock {
public:
    class key_val : public referable
    {
    public:
        text key;
        text val;
        key_val(const text& key, const text& val) { this->key = key; this->val = val; }
        virtual ~key_val() {}
    };
    class key_vals : public referable
    {
    public:
        text key;
        vector<text> vals;
        key_vals(const text& key, const text& val) { this->key = key; vals.push_back(val); }
        void add(const text& val) { vals.push_back(val); }
        bool contains(const text& val) { for (vector<text>::const_iterator it = vals.begin(); it != vals.end(); ++it) { if (*it == val) return true; } return false; }
        virtual ~key_vals() {}
    };
private:
    mutable karoo_mutex sem;
    text uri_scheme;
    int http_port;
    time_t http_timeout;
    time_t close_linger_time;
    time_t close_inactivity_time;
    stream_service* service;
    HTTP_factory* factory;
    stream_socket* sock;
    exeque* listenq;
    exeque* streamq;
    exeque* httpq;
    HTTP_handler_factory* handler_factory;
    khashe mime_types;
    khashe env;
    text root;
    text document_root;
    mutable khashe document_roots;
    mutable khashe cave_rocks;
    mutable khashe cave_rock_types;
    text ssl_certificate;
    text ssl_private_key;
    bool ssl_ans1;
    text surf_root;
    text default_document;
    mutable khashe default_documents;
    vector<text> post_response_messages;
    
public:
    surf(vector<text>& opts, int argc, char *argv[]);
    virtual ~surf();
    void traverse(fab* conf, nxml_data_t* p, nxml_data_t* ctx);
    void configure();
    void startup();
    void stop(bool block = true);
    text getRoot() const { return root; }
    text getDocumentRoot(HTTP_parsed_data* data = NULL) const;
    text getDocumentRoot(const text& host) const;
    text getSSLCertificate() const;
    text getSSLPrivateKey() const;
    text getDefaultDocument(HTTP_parsed_data* data = NULL) const;
    text getSurfRoot() const;
    text getType() const { return "surf"; }
    text& writePostResponseMessages(text& s);
    void pushResponseMessage(const text& str);
    bool getMimeType(const text& ext, text& type, const text& host);
    bool getEnv(const text& key, text& val, const text& host);
    HTTP_factory* getFactory() { return factory; }
    bool checkRockName(const text& rock_name, const text& host);
    bool checkRockType(const text& rock_type, const text& host);
};

class surf_handler: public rock_datagram_message_handler
{
public:
    surf_handler(surf* parent) : rock_datagram_message_handler(parent) {}
    virtual ~surf_handler() {}
    bool handle(uint32_t msg_type, uint64_t seq, int32_t port, const text& name, rock_datagram_message* msg);
};

class surf_HTTP_handler_factory: public HTTP_handler_factory {
private:
    surf* parent;
public:
    surf_HTTP_handler_factory(surf* parent) { this->parent = parent; }
    virtual ~surf_HTTP_handler_factory() {}

    HTTP_handler* create_HTTP_handler(HTTP_parsed_data* data);
};

class surf_HTTP_handler: public HTTP_handler {
private:
    surf* parent;
    bool head;
public:
    surf_HTTP_handler(HTTP_parsed_data* data, surf* parent, bool head = false);
    virtual ~surf_HTTP_handler();

    bool run();
};

class surf_HTTP_fab_handler: public HTTP_handler {
public:
    enum active_type {
        ACTIVE_TYPE_PARAMETER = 0,
        ACTIVE_TYPE_CONDUIT,
    };

    class active : public referable
    {
    public:
        enum active_type type;
        surf_HTTP_fab_handler* parent;
        active(surf_HTTP_fab_handler* parent, enum active_type type);
        virtual ~active();
    };

    class active_parameter : public active
    {
    public:
        text key;
        text val;
        active_parameter(surf_HTTP_fab_handler* parent, const text& key, const text& val);
        virtual ~active_parameter();
    };
    
    class active_conduit : public active
    {
    public:
        text id;
        text rock_name;
        text rock_type;
        text service_id;
        vector<surf_HTTP_fab_handler::active_parameter*> params;

        active_conduit(surf_HTTP_fab_handler* parent, const text& id, const text& rock_name, const text& rock_type, const text& service_id);
        virtual ~active_conduit();
    };

private:
    karoo_mutex sem;
    surf* parent;
    bool head;
    fab* source;
    text url;
    nxml_data_t* root;
    nxml_t* nxml;
    bool first_pass;
    khashe conduits;
    khashe params;
    text proxy_host;
    bool traverseFirstPass(nxml_data_t* node, active* context);
    void traverseFinalPass(nxml_data_t* node);
    nxml_data_t* setColumnValues(nxml_data_t* node, khashe* columns);
    bool traverseCheckComplete(nxml_data_t* node);
    text traverseGetErrors(nxml_data_t* node);
    bool _isComplete();
public:
    surf_HTTP_fab_handler(HTTP_parsed_data* data, surf* parent, bool head = false);
    virtual ~surf_HTTP_fab_handler();

    bool run();
    void timedOut();

    void setParametersSuccess(const text& conduit_id, size_t sql_count, size_t row_count);
    void setParametersNull(const text& conduit_id, size_t sql_count, size_t row_count);
    void setParametersDone(const text& conduit_id, size_t sql_count, size_t row_count);
    void setParameter(const text& conduit_id, size_t sql_count, size_t row_count, khashe& columns);
    bool isComplete();
};

class surf_HTTP_post_handler: public HTTP_handler {
private:
    surf* parent;
    text proxy_host;
public:
    surf_HTTP_post_handler(HTTP_parsed_data* data, surf* parent);
    virtual ~surf_HTTP_post_handler();

    bool run();
};


class surf_HTTP_form_data_handler: public HTTP_handler {
private:
    surf* parent;
    text boundary;
    text proxy_host;

    class form_data {
    public:
        text file_name;
        text param_name;
        char* data;
        size_t sz;
        form_data() { data = NULL; sz = 0; }
        ~form_data() {}
    };

public:
    surf_HTTP_form_data_handler(HTTP_parsed_data* data, surf* parent, const text& boundary);
    virtual ~surf_HTTP_form_data_handler();

    virtual bool run();

};

---