evpsgi.c

/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *  evpsgi - PSGI server with libevent httpd
 *
 *       http://github.com/sekimura/evpsgi
 *
 *  Copyright 2009 Masayoshi Sekimura.  All rights reserved.
 *
 *  This library is free software; you can redistribute it and/or modify it
 *  under the same terms as Perl itself
 *
 *  Authors:
 *      Masayoshi Sekimura <sekimura@gmail.com>
 */


#include "evpsgi.h"

#include <sys/queue.h>

#include <err.h>
#include <sys/time.h>
#include <stdlib.h>
#include <stdarg.h>
#include <sysexits.h>
    
#include <netdb.h>
    
#include <event.h>
#include <evhttp.h>

#include <EXTERN.h>
#include <perl.h>
#include <XSUB.h>

/* generated by perl −MExtUtils::Embed −e xsinit −− −o xsinit.c */
EXTERN_C void xs_init (pTHX);

EXTERN_C void boot_DynaLoader (pTHX_ CV* cv);

XS(EvPSGI_Input_read);
XS(EvPSGI_Input_read)
{
    dXSARGS;
    SV *self = ST(0);
    SV *buf = ST(1);
    size_t len = SvIV(ST(2));
    size_t offset = items >= 4 ? SvIV(ST(3)) : 0;
    SV *ret;
    struct evhttp_request *req;
    const char *in;
    size_t inlen;
    dXSTARG;

    ret = (SV *)sv_2mortal((SV *)newSVpv("",0));

    req = (struct evhttp_request *) mg_find(SvRV(self), PERL_MAGIC_ext)->mg_obj;
    in = (const char *) EVBUFFER_DATA( req->input_buffer );
    inlen = EVBUFFER_LENGTH( req->input_buffer );

    if (inlen > 0 && inlen > offset) {
        in += offset;
        sv_catpvn(ret, in, len > inlen ? inlen : len);
    }

    sv_setsv(buf, ret);
    ST(0) = sv_2mortal(newSViv(SvCUR(buf)));
    XSRETURN(1);
}

XS(EvPSGI_Errors_print);
XS(EvPSGI_Errors_print)
{
    dXSARGS;
    SV *self = ST(0);
    SV *msg  = ST(1);
    dXSTARG;
    
    warnx("ERROR: %s", SvPV_nolen(msg));

    ST(0) = (SV *)sv_2mortal((SV *)newSViv(1));
    XSRETURN(1);
}

EXTERN_C void
xs_init(pTHX)
{
    char *file = __FILE__;
    dXSUB_SYS;

    /* DynaLoader is a special case */
    newXS("DynaLoader::boot_DynaLoader", boot_DynaLoader, file);
    newXSproto("EvPSGI::Input::read", EvPSGI_Input_read, file, "$$$;$");
    newXSproto("EvPSGI::Errors::print", EvPSGI_Errors_print, file, "$$");
}

/** exported globals **/
static PerlInterpreter *perlinterp = NULL;
struct settings settings;
SV *app;

SV* new_env(struct evhttp_request *req) {
    AV *version;
    HV *env;
    SV *input, *errors;
    struct evkeyval *header;
    const char *qstring, *p;
    char *decoded_uri, *method;
    char http_version[9];

    decoded_uri = evhttp_decode_uri(req->uri);

    env = (HV *)sv_2mortal((SV *)newHV());

    /* REQUEST_METHOD */
    switch (req->type) {
    case EVHTTP_REQ_GET:
        method = "GET";
        break;
    case EVHTTP_REQ_POST:
        method = "POST";
        break;
    case EVHTTP_REQ_HEAD:
        method = "HEAD";
        break;
    default:
        method = NULL;
        break;
    }
    hv_store(env, "REQUEST_METHOD", 14, newSVpv(method, 0), 0);

    /* REQUEST_URI */
    hv_store(env, "REQUEST_URI", 11, newSVpv(req->uri, 0), 0);

    /* SCRIPT_NAME */
    /* PATH_INFO */
    if (settings.alias_is_root) {
        hv_store(env, "SCRIPT_NAME", 11, newSVpv("", 0), 0);
        hv_store(env, "PATH_INFO", 9, newSVpv(decoded_uri, strcspn(decoded_uri, "?")), 0);
    }
    else {
        hv_store(env, "SCRIPT_NAME", 11, newSVpv(settings.alias, 0), 0);
        p = decoded_uri + strlen(settings.alias);
        hv_store(env, "PATH_INFO", 9, newSVpv(p, strcspn(p, "?")), 0);
    }

    /* QUERY_STRING */
    qstring = strchr(decoded_uri, '?');
    if (qstring == NULL) {
        qstring = "";
    } else {
        qstring++;
    } 
    hv_store(env, "QUERY_STRING", 12, newSVpv(qstring, 0), 0);

    /* SERVER_NAME */
    /* SERVER_PORT */
    hv_store(env, "SERVER_NAME", 11, newSVpv(settings.server_name, 0), 0);
    hv_store(env, "SERVER_PORT", 11, newSViv(settings.port), 0);

    /* SERVER_PROTOCOL */
    sprintf(http_version, "HTTP/%1d.%1d", req->major, req->minor);
    hv_store(env, "SERVER_PROTOCOL", 15, newSVpv(http_version, 0), 0);

    /* HTTP_ valiables */
    TAILQ_FOREACH(header, req->input_headers, next) {
        char header_key[strlen(header->key) + 1 ];
        char *prefix = "HTTP_";
        char buf[strlen(prefix) + strlen(header->key) + 1 ];
        char *ptr;

        strncpy(header_key, header->key, sizeof(header_key));
        ptr = header_key;
        while(*ptr){
            *ptr = toupper(*ptr);
            if (*ptr == '-') *ptr = '_';
            ptr++;
        }

        if (strcmp(header_key, "CONTENT_LENGTH") == 0 ||
            strcmp(header_key, "CONTENT_TYPE") == 0 ) {
            prefix = "";
        }

        strncpy(buf, prefix, sizeof(buf));
        strncat(buf, header_key, sizeof(buf) - strlen(buf) - 1);
        hv_store(env, buf, strlen(buf), newSVpv(header->value, 0), 0);
        /*
         * TODO:
         * If there are multiple header lines sent with the same key, the server
         * should treat them as if they're sent in one line, i.e. combine them
         * with C<, > as in RFC 2616.
         */
    }

    /* psgi.version */
    version = newAV();
    av_push(version, newSViv(1));
    av_push(version, newSViv(0));
    hv_store(env, "psgi.version", 12, newRV_noinc((SV *) version), 0);

    /* psgi.url_scheme */
    hv_store(env, "psgi.url_scheme", 15, newSVpv("http", 0), 0);

    /* psgi.input */
    input = newRV_noinc(newSV(0));
    sv_magic(SvRV(input), NULL, PERL_MAGIC_ext, NULL, 0);
    mg_find(SvRV(input), PERL_MAGIC_ext)->mg_obj = (void *) req;
    sv_bless(input, gv_stashpv("EvPSGI::Input", 1));
    hv_store(env, "psgi.input", 10, input, 0);

    /* psgi.errors */
    errors = newRV_noinc(newSV(0));
    sv_magic(SvRV(errors), NULL, PERL_MAGIC_ext, NULL, 0);
    sv_bless(errors, gv_stashpv("EvPSGI::Errors", 1));
    hv_store(env, "psgi.errors", 11, errors, 0);

    /* psgi.multithread */
    hv_store(env, "psgi.multithread", 16, newSViv(0), 0);

    /* psgi.multiprocess */
    hv_store(env, "psgi.multiprocess", 17, newSViv(0), 0);

    /* psgi.run_once */
    hv_store(env, "psgi.run_once", 13, newSViv(1), 0);

    /* psgi.nonblocking */
    hv_store(env, "psgi.nonblocking", 16, newSViv(1), 0);

    free(decoded_uri);
    return newRV_inc((SV *) env);
}

static SV *run_app(SV *env)
{
    dTHX;
    int count;
    SV *res;
    dSP;
    ENTER;
    SAVETMPS;
    PUSHMARK(SP);
    XPUSHs(sv_2mortal(env));
    PUTBACK;

    hv_store(GvHV(PL_envgv), "EVHTTP_PSGI", 8, newSVpv("1", 0), 0);

    count = call_sv(app, G_EVAL|G_SCALAR|G_KEEPERR);
    SPAGAIN;
    if (SvTRUE(ERRSV)) {
        res = NULL;
        fprintf(stderr, "FATAL: %s", SvPV_nolen(ERRSV));
        /* CLEAR_ERRSV() is only available 5.8.9 or later */
        if (SvMAGICAL(ERRSV)) {
            mg_free(ERRSV);
            mg_clear(ERRSV);
        }
        sv_setpvn_mg(ERRSV,"",0);
        POPs;
    } else if (count > 0) {
        res = POPs;
        SvREFCNT_inc(res);
    } else {
        res = NULL;
    }
    PUTBACK;
    FREETMPS;
    LEAVE;
    return res;
}

static SV * delayed_res(SV *cb)
{
    SV *respond, *res;
    char *code;
    I32 ax;
    int count;

    dSP;

    /* TODO: support the writer object */ 
    code = "sub { my $res = shift; $res;}";
    respond = eval_pv(code, TRUE);

    PUSHMARK(SP);
    XPUSHs(respond);
    PUTBACK;

    count = call_sv(cb, G_SCALAR);

    SPAGAIN;
    SP -= count;
    ax = (SP - PL_stack_base) + 1;
    res = ST(0);
    PUTBACK;

    return res;
}

void add_headers(struct evhttp_request *req,  SV *headers )
{
    AV * headers_av;
    SV *key_sv, *val_sv;
    char *key, *val;

    headers_av = (AV *) SvRV(headers);
    while (av_len(headers_av) > -1) {
        key_sv = av_shift(headers_av);
        val_sv = av_shift(headers_av);
        if (key_sv == NULL || val_sv == NULL) break;
        key = SvPV_nolen(key_sv);
        val = SvPV_nolen(val_sv);
        evhttp_add_header(req->output_headers, key, val);
        SvREFCNT_dec(key_sv);
        SvREFCNT_dec(val_sv);
    }
}

int add_body_iterator(struct evbuffer *buf, struct evhttp_request *req,  SV *io )
{
    dTHX;
    SV *buf_sv;
    STRLEN len;
    dSP;
    char *line;
    int count;

    ENTER;
    SAVETMPS;

    while (1) {
        PUSHMARK(SP);
        XPUSHs(io);
        PUTBACK;
        count = call_method("getline", G_SCALAR);
        SPAGAIN;
        buf_sv = POPs;
        if (!SvOK(buf_sv))
            break;
        line = SvPV(buf_sv, len);
        evbuffer_add(buf, line, len);
        count += len;
    }
    PUSHMARK(SP);
    XPUSHs(io);
    PUTBACK;
    call_method("close", G_DISCARD);
    SPAGAIN;
    PUTBACK;
    FREETMPS;
    LEAVE;
    return count;
}

int add_body_av(struct evbuffer *buf, struct evhttp_request *req,  AV *body_av )
{
    SV *b;
    I32 i, lastidx;
    STRLEN len;
    char *line;
    int ret;

    lastidx = av_len(body_av);
    for (i = 0; i <= lastidx; i++) {
        b = (SV *) *(av_fetch(body_av, i, 0));
        if (SvOK(b)) {
            line = SvPV(b, len);
            evbuffer_add(buf, line, len);
            ret += len;
        }
    }
    return ret;
}
int add_body(struct evbuffer *buf, struct evhttp_request *req,  SV *body )
{
    int ret, type;

    switch (type = SvTYPE(SvRV(body))) {
        case SVt_PVAV:
            add_body_av(buf, req, (AV *) SvRV(body));
            break;
        case SVt_PVHV:
            if (!sv_derived_from(body, "IO::Handle::Iterator")) {
                err(1, "response body must be an array reference or IO::Handle like object");
            }
        case SVt_PVGV:
            if (!sv_derived_from(body, "IO::Handle")) {
                require_pv("IO/Handle.pm");
            }
            add_body_iterator(buf, req, body);
            break;
        default:
            err(1, "response body must be an array reference or IO::Handle like object");
            ret = 0;
            break;
    }

    return ret;
}


void psgi_handler(struct evhttp_request *req, void *arg)
{
    struct evbuffer *evb;
    const char *ptr;
    SV *env, *res, *status;
    AV *res_av;

    ENTER;
    SAVETMPS;

    evb = evbuffer_new();

    if (evb == NULL)
        err(1, "failed to create response buffer");

    ptr = req->uri;

    if (!settings.alias_is_root)
        ptr += strlen(settings.alias);

    if (strncmp(req->uri, settings.alias, strlen(settings.alias)) != 0) {
        evhttp_send_error(req, HTTP_NOTFOUND, "File Not Found.");
        evbuffer_free(evb);
        FREETMPS;
        LEAVE;
        return;
    }
    else if (!(*ptr == '/' || *ptr == '?' || *ptr == '#')) {
        evhttp_send_error(req, HTTP_NOTFOUND, "File Not Found");
        evbuffer_free(evb);
        FREETMPS;
        LEAVE;
        return;
    }

    env = new_env(req);

    res = run_app(env);
    if (res == NULL) {
        evhttp_send_error(req, 500, "Internal Server Error");
        evbuffer_free(evb);
        FREETMPS;
        LEAVE;
        return;
    }

    if ( SvTYPE(SvRV(res)) == SVt_PVCV ) {
        res = delayed_res(res);
        SvREFCNT_inc(res);
    }

    res_av = (AV *) SvRV(res);
 
    /* response status code */
    status = (SV *) *(av_fetch(res_av, 0, 0));

    /* response header */
    add_headers(req, *(av_fetch(res_av, 1, 0)));

    /* response body */
    add_body(evb, req, *(av_fetch(res_av, 2, 0)));

    evhttp_send_reply(req, SvIV(status), "OK", evb);

    evbuffer_free(evb);
    SvREFCNT_dec(res); /* since it was increaced in run_app() */
    FREETMPS;
    LEAVE;
}

void generic_handler(struct evhttp_request *req, void *arg)
{
    struct evbuffer *buf;
    buf = evbuffer_new();

    if (buf == NULL)
        err(1, "failed to create response buffer");

    evbuffer_add(buf, "Hi there\n", 9);
    evhttp_send_reply(req, HTTP_OK, "OK", buf);
    evbuffer_free(buf);
}

static void settings_init(void) {
    settings.inter = "0.0.0.0";
    settings.port = 80;
    settings.file =  NULL;
    settings.alias = "/";
    settings.alias_is_root = TRUE;
    settings.server_name = NULL;
}

static void usage(void) {
    printf("-p <num>      TCP port number to listen on (default: 80)\n"
           "-f <file>     PSGI script file\n"
           "-a <mask>     alias\n"
           "-l <ip_addr>  interface to listen on (default: INADDR_ANY, all addresses)\n"
    );
    return;
}

int main(int argc, char **argv, char**env)
{
    int c;
    struct evhttp *httpd;
    struct event_base *ev;
    char *embedding[] = { "", "-e", "0" };
    char code[1024];
    struct addrinfo *ai;
    struct addrinfo *next;
    struct addrinfo hints = { .ai_flags = AI_PASSIVE,
                              .ai_family = AF_UNSPEC,
                              .ai_socktype = SOCK_STREAM, };
    char port_buf[NI_MAXSERV];
    int error;
    int rc;
    char server_name_buf[NI_MAXHOST];

    /* init settings */
    settings_init();

    /* process arguments */
    while (-1 != (c = getopt(argc, argv,
          "a:"  /* alias */
          "p:"  /* TCP port number to listen on */
          "f:"  /* psgi script file */
          "l:"  /* interface to listen on */
          "h"   /* usage */
        ))) {
        switch (c) {
        case 'a':
            settings.alias = optarg;
            break;
        case 'p':
            settings.port = atoi(optarg);
            break;
        case 'f':
            settings.file = optarg;
            break;
        case 'l':
            settings.inter= optarg;
            break;
        case 'h':
            usage();
            exit(EXIT_SUCCESS);
        default:
            warnx("Illegal argument \"%c\"", c);
            return 1;
        }
    }

    if (settings.file == NULL)
        errx(1, "No psgi file specified");

    settings.alias_is_root = (strcmp(settings.alias, "/") == 0)
        ? TRUE : FALSE;

    snprintf(port_buf, sizeof(port_buf), "%d", settings.port);
    error= getaddrinfo(settings.inter, port_buf, &hints, &ai);
    if (error != 0) {
        if (error != EAI_SYSTEM)
          warnx("getaddrinfo(): %s", gai_strerror(error));
        else
          err(1, "getaddrinfo()");
        return 1;
    }

    for (next= ai; next; next= next->ai_next) {
        rc = getnameinfo((struct sockaddr *) next->ai_addr, next->ai_addrlen,
                    server_name_buf, sizeof(server_name_buf),
                    NULL, 0,
                    0);
        if (rc != 0) {
            warnx("getnameinfo(): %s", gai_strerror(rc));
            continue;
        }
        settings.server_name = server_name_buf; 
        break;
    }

    PERL_SYS_INIT3(&argc,&argv,&env);
    perlinterp = perl_alloc();
    PERL_SET_CONTEXT(perlinterp); /* not my_perl */
    perl_construct(perlinterp);
    perl_parse(perlinterp, xs_init, 3, embedding, NULL);
    PL_exit_flags |= PERL_EXIT_DESTRUCT_END;

    sprintf(code, "do \"%s\" or die $@", settings.file);
    app = eval_pv(code, TRUE); /* croak_on_error */
    warnx("eval_pv(\"%s\"): ", code);

    /* start httpd */
    ev = event_init();
    httpd = evhttp_start(settings.inter, settings.port);
    if (httpd == NULL) {
        err(1, "evhttp_start(\"%s\", %d) failed: ", settings.inter, settings.port);
    }
    else {
        warnx("evhttp_start(\"%s\", %d): ", settings.inter, settings.port);
    }

    /* Set a callback for all requests. */
    evhttp_set_gencb(httpd, psgi_handler, NULL);

    event_dispatch();

    /* Not reached in this code as it is now. */
    PL_perl_destruct_level = 1;
    perl_destruct(perlinterp);
    perl_free(perlinterp);
    PERL_SYS_TERM();
    evhttp_free(httpd);
    event_base_free(ev);

    return 0;
}