add wsgi support

nginx/http/http.pyx

@@ -1,7 +1,29 @@
 from cpython cimport Py_INCREF, Py_DECREF
-from .nginx_core cimport ngx_log_error, NGX_LOG_CRIT, NGX_AGAIN, from_nginx_str
-from .ngx_http cimport ngx_http_request_t, ngx_http_core_run_phases
-from .ngx_http cimport ngx_http_get_module_ctx, ngx_http_set_ctx
+from .nginx_core cimport (
+    ngx_log_error,
+    NGX_LOG_CRIT,
+    NGX_AGAIN,
+    from_nginx_str,
+    ngx_calloc,
+    ngx_free,
+    ngx_memcpy,
+    ngx_module_t,
+    ngx_str_t,
+)
+from .ngx_http cimport (
+    ngx_http_request_t,
+    ngx_http_core_run_phases,
+    ngx_http_get_module_ctx,
+    ngx_http_set_ctx,
+    ngx_http_send_header,
+    ngx_list_push,
+    ngx_table_elt_t,
+    ngx_str_set,
+    ngx_http_output_filter,
+    ngx_chain_t,
+    ngx_buf_t,
+    ngx_calloc_buf,
+)
 
 import traceback
 
@@ -17,6 +39,8 @@ cdef class Request:
         public str extension
         public str unparsed_uri
         public str method_name
+        public str content_type
+        public str content_length
         public str http_protocol
 
     def __init__(self, *args):
@@ -37,17 +61,88 @@ cdef class Request:
             return self.future.result()
         return NGX_AGAIN
 
+    def send_header(self):
+        return ngx_http_send_header(self.request)
+
+    def add_response_header(self, key, value):
+        cdef:
+            ngx_table_elt_t *h
+            char *cstr
+            char *csource
+            bytes key_data, value_data
+        h = ngx_list_push(&self.request.headers_out.headers)
+        if h == NULL:
+            raise MemoryError()
+        h.hash = 1
+
+        key_data = str(key).encode('iso8859-1')
+        cstr = <char *>ngx_calloc(sizeof(char) * len(key_data), self.request.connection.log)
+        h.key.len = len(key_data)
+        csource = key_data
+        ngx_memcpy(cstr, csource, len(key_data))
+        h.key.data = cstr
+
+        value_data = str(value).encode('iso8859-1')
+        cstr = <char *>ngx_calloc(sizeof(char) * len(value_data), self.request.connection.log)
+        h.value.len = len(value_data)
+        csource = value_data
+        ngx_memcpy(cstr, csource, len(value_data))
+        h.value.data = cstr
+
+    def send_response(self, pos):
+        cdef:
+            ngx_chain_t out
+            ngx_buf_t *b
+            bytes data = pos
+            char* cstr = data
+        b = ngx_calloc_buf(self.request.pool)
+        if b == NULL:
+            raise MemoryError
+        b.last_buf = 1
+        b.last_in_chain = 1
+        b.memory = 1
+        b.pos = cstr
+        b.last = b.pos + len(data)
+
+        out.buf = b
+        out.next = NULL
+
+        return ngx_http_output_filter(self.request, &out)
+
+    def get_app_from_config(self):
+        cdef ngx_wsgi_pass_conf_t *conf
+
+        conf = <ngx_wsgi_pass_conf_t *>self.request.loc_conf[ngx_python_module.ctx_index]
+        return from_nginx_str(conf.wsgi_pass)
+
+    property response_status:
+        def __get__(self):
+            return self.request.headers_out.status
+
+        def __set__(self, value):
+            self.request.headers_out.status = value
+
+    property response_content_length:
+        def __get__(self):
+            if self.request.headers_out.content_length:
+                return self.request.headers_out.content_length.value
+
+        def __set__(self, value):
+            self.request.headers_out.content_length.value = value
+
     def __repr__(self):
         return f'Request({self.method_name} {self.uri})'
 
     def __str__(self):
         return f''' request_line: {self.request_line}
-          uri: {self.uri}
-         args: {self.args}
-    extension: {self.extension}
- unparsed_uri: {self.unparsed_uri}
-  method_name: {self.method_name}
-http_protocol: {self.http_protocol}'''
+           uri: {self.uri}
+          args: {self.args}
+     extension: {self.extension}
+  unparsed_uri: {self.unparsed_uri}
+   method_name: {self.method_name}
+  content_type: {self.content_type}
+content_length: {self.content_length}
+ http_protocol: {self.http_protocol}'''
 
     @staticmethod
     cdef Request from_ptr(ngx_http_request_t *request):
@@ -67,6 +162,12 @@ http_protocol: {self.http_protocol}'''
             new_req.unparsed_uri = from_nginx_str(request.unparsed_uri)
             new_req.method_name = from_nginx_str(request.method_name)
             new_req.http_protocol = from_nginx_str(request.http_protocol)
+            if request.headers_in.content_type:
+                new_req.content_type = from_nginx_str(
+                    request.headers_in.content_type.value)
+            if request.headers_in.content_length:
+                new_req.content_length = from_nginx_str(
+                    request.headers_in.content_length.value)
 
             ngx_http_set_ctx(request, <void *>new_req, ngx_python_module)
             return new_req

nginx/http/module.py

@@ -1,4 +1,7 @@
 import asyncio
+import io
+import sys
+import functools
 
 from . import log
 from .._nginx import run_phases
@@ -13,14 +16,50 @@ def __init__(self):
     def init_process(self):
         self.loop = asyncio.get_event_loop()
 
+    def start_response(self, request, status, response_headers, exc_info=None):
+        request.response_status = int(status.split(maxsplit=1)[0])
+        for key, value in response_headers:
+            request.add_response_header(key, value)
+
     def post_read(self, request):
-        if request._started():
-            log.debug('post_read end')
-            return request._result()
+        environ = {
+            'REQUEST_METHOD': request.method_name,
+            'SCRIPT_NAME': None,
+            'PATH_INFO': request.uri,
+            'QUERY_STRING': request.args,
+            'CONTENT_TYPE': request.content_type,
+            'CONTENT_LENGTH': request.content_length,
+            'SERVER_NAME': 'localhost',
+            'SERVER_PORT': '8080',
+            'SERVER_PROTOCOL': request.http_protocol,
+            'wsgi.input': io.BytesIO(),
+            'wsgi.errors': sys.stderr,
+            'wsgi.version': (1, 0),
+            'wsgi.multithread': False,
+            'wsgi.multiprocess': True,
+            'wsgi.run_once': True,
+        }
+        if environ.get('HTTPS', 'off') in ('on', '1'):
+            environ['wsgi.url_scheme'] = 'https'
         else:
-            log.debug('post_read request:\n%s', request)
-            return request._start(self.loop.create_task(
-                self._post_read_async(request)))
+            environ['wsgi.url_scheme'] = 'http'
+        app_name = request.get_app_from_config()
+        module_name, method_name = app_name.split(':')
+        app = getattr(__import__(module_name), method_name)
+        resp = app(environ, functools.partial(
+            self.start_response, request))
+        request.send_header()
+        rv = 404
+        for pos in resp:
+            rv = request.send_response(pos)
+        return rv
+        # if request._started():
+        #     log.debug('post_read end')
+        #     return request._result()
+        # else:
+        #     log.debug('post_read request:\n%s', request)
+        #     return request._start(self.loop.create_task(
+        #         self._post_read_async(request)))
 
     async def _post_read_async(self, request):
         try:

nginx/nginx.pyx

@@ -4,14 +4,16 @@ import traceback
 from enum import IntEnum
 
 from .nginx_config cimport ngx_int_t
-from .nginx_core cimport ngx_module_t, ngx_cycle_t
+from .nginx_core cimport ngx_module_t, ngx_cycle_t, ngx_str_t
 from .nginx_core cimport NGX_OK, NGX_ERROR, NGX_DECLINED, NGX_AGAIN
 from .nginx_core cimport NGX_LOG_DEBUG, NGX_LOG_CRIT
 from .nginx_core cimport ngx_log_error
 
 
 cdef extern from "ngx_python_module.h":
     ngx_module_t ngx_python_module
+    ctypedef struct ngx_wsgi_pass_conf_t:
+        ngx_str_t  wsgi_pass
 
 
 class ReturnCode(IntEnum):

nginx/nginx_core.pxd

@@ -26,6 +26,7 @@ cdef extern from "ngx_core.h":
         char *data
 
     ctypedef struct ngx_module_t:
+        ngx_uint_t ctx_index
         pass
 
     ctypedef struct ngx_log_t:
@@ -40,6 +41,7 @@ cdef extern from "ngx_core.h":
 
     void *ngx_calloc(size_t size, ngx_log_t *log)
     void ngx_free(void *p)
+    void *ngx_memcpy(void *dst, const void *src, size_t n)
     void ngx_log_error(ngx_uint_t level,
                        ngx_log_t *log,
                        ngx_err_t err,

nginx/ngx_http.pxd

@@ -1,7 +1,39 @@
-from .nginx_core cimport ngx_str_t, ngx_module_t, ngx_log_t
+from .nginx_core cimport ngx_str_t, ngx_module_t, ngx_log_t, ngx_uint_t, ngx_int_t
 
 
 cdef extern from "ngx_http.h":
+
+    ctypedef struct ngx_table_elt_t:
+        ngx_uint_t hash
+        ngx_str_t key
+        ngx_str_t value
+
+    ctypedef struct ngx_list_t:
+        pass
+
+    ctypedef struct ngx_http_headers_in_t:
+        ngx_table_elt_t *content_type
+        ngx_table_elt_t *content_length
+
+    ctypedef struct ngx_http_headers_out_t:
+        ngx_uint_t status
+        ngx_table_elt_t *content_length
+        ngx_list_t headers
+
+    ctypedef struct ngx_chain_t:
+        ngx_buf_t *buf
+        ngx_chain_t *next
+
+    ctypedef struct ngx_buf_t:
+        unsigned last_buf
+        unsigned last_in_chain
+        unsigned memory
+        char *pos
+        char *last
+
+    ctypedef struct ngx_pool_t:
+        pass
+
     ctypedef struct ngx_connection_t:
         ngx_log_t *log
 
@@ -14,9 +46,25 @@ cdef extern from "ngx_http.h":
         ngx_str_t unparsed_uri
         ngx_str_t method_name
         ngx_str_t http_protocol
+        ngx_pool_t *pool
+        ngx_http_headers_in_t headers_in
+        ngx_http_headers_out_t headers_out
+        void **loc_conf
 
     void ngx_http_core_run_phases(ngx_http_request_t *request)
     void *ngx_http_get_module_ctx(ngx_http_request_t *request,
                                   ngx_module_t module)
     void ngx_http_set_ctx(ngx_http_request_t *request, void *ctx,
                           ngx_module_t module)
+
+    void ngx_http_send_header(ngx_http_request_t *r)
+
+    ngx_table_elt_t *ngx_list_push(ngx_list_t *list)
+
+    void ngx_str_set(ngx_str_t *str, char *text)
+
+    ngx_int_t ngx_http_output_filter(ngx_http_request_t *r, ngx_chain_t *input)
+
+    ngx_buf_t *ngx_calloc_buf(ngx_pool_t *pool)
+
+

nginx/ngx_python_module.c

@@ -3,6 +3,7 @@
 #include <ngx_http.h>
 #include <Python.h>
 #include "nginx.h"
+#include "ngx_python_module.h"
 
 
 static ngx_int_t ngx_python_init_process(ngx_cycle_t *cycle);
@@ -16,9 +17,6 @@ static void *ngx_http_wsgi_create_loc_conf(ngx_conf_t *cf);
 static char *ngx_handle_app(ngx_conf_t *cf, ngx_command_t *cmd, void *conf);
 static ngx_conf_post_t  ngx_wsgi_pass_post = { ngx_handle_app };
 
-typedef struct {
-    ngx_str_t  wsgi_pass;
-} ngx_wsgi_pass_conf_t;
 
 static ngx_command_t  ngx_wsgi_commands[] = {
     { ngx_string("wsgi_pass"),
@@ -109,7 +107,7 @@ ngx_python_postconfiguration(ngx_conf_t *cf) {
 
     cmcf = ngx_http_conf_get_module_main_conf(cf, ngx_http_core_module);
 
-    h = ngx_array_push(&cmcf->phases[NGX_HTTP_POST_READ_PHASE].handlers);
+    h = ngx_array_push(&cmcf->phases[NGX_HTTP_PRECONTENT_PHASE].handlers);
     if (h == NULL) {
         return NGX_ERROR;
     }

nginx/ngx_python_module.h

@@ -1 +1,5 @@
 extern ngx_module_t ngx_python_module;
+
+typedef struct {
+    ngx_str_t  wsgi_pass;
+} ngx_wsgi_pass_conf_t;