Python: fixing ASGI receive() issues.

The receive() call never blocks for a GET request and always returns the same
empty body message.  The Starlette framework creates a separate task when
receive() is called in a loop until an 'http.disconnect' message is received.

The 'http.disconnect' message was previously issued after the response header
had been sent.  However, the correct behavior is to respond with
'http.disconnect' after sending the response is complete.

This closes #564 issue on GitHub.

2 files changed, 61 insertions, 33 deletions

diff --git a/docs/changes.xml b/docs/changes.xml
index dd8fb731..68db823d 100644
--- a/docs/changes.xml
+++ b/docs/changes.xml
@@ -65,6 +65,13 @@ a full-form IPv6 in a listener address.
 </para>
 </change>
 
+<change type="bugfix">
+<para>
+compatibility issues with some Python ASGI apps, notably based on Starlette
+framework.
+</para>
+</change>
+
 </changes>
 
 
diff --git a/src/python/nxt_python_asgi_http.c b/src/python/nxt_python_asgi_http.c
index d88c4b00..3074c09f 100644
--- a/src/python/nxt_python_asgi_http.c
+++ b/src/python/nxt_python_asgi_http.c
@@ -23,10 +23,11 @@ typedef struct {
     PyObject                 *send_future;
     uint64_t                 content_length;
     uint64_t                 bytes_sent;
-    int                      complete;
-    int                      closed;
     PyObject                 *send_body;
     Py_ssize_t               send_body_off;
+    uint8_t                  complete;
+    uint8_t                  closed;
+    uint8_t                  empty_body_received;
 } nxt_py_asgi_http_t;
 
 
@@ -37,6 +38,7 @@ static PyObject *nxt_py_asgi_http_response_start(nxt_py_asgi_http_t *http,
     PyObject *dict);
 static PyObject *nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http,
     PyObject *dict);
+static void nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http);
 static PyObject *nxt_py_asgi_http_done(PyObject *self, PyObject *future);
 
 
@@ -94,10 +96,11 @@ nxt_py_asgi_http_create(nxt_unit_request_info_t *req)
         http->send_future = NULL;
         http->content_length = -1;
         http->bytes_sent = 0;
-        http->complete = 0;
-        http->closed = 0;
         http->send_body = NULL;
         http->send_body_off = 0;
+        http->complete = 0;
+        http->closed = 0;
+        http->empty_body_received = 0;
     }
 
     return (PyObject *) http;
@@ -117,7 +120,7 @@ nxt_py_asgi_http_receive(PyObject *self, PyObject *none)
 
     nxt_unit_req_debug(req, "asgi_http_receive");
 
-    if (nxt_slow_path(http->closed || nxt_unit_response_is_sent(req))) {
+    if (nxt_slow_path(http->closed || http->complete )) {
         msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str);
 
     } else {
@@ -171,6 +174,14 @@ nxt_py_asgi_http_read_msg(nxt_py_asgi_http_t *http)
         size = nxt_py_asgi_http_body_buf_size;
     }
 
+    if (size == 0) {
+        if (http->empty_body_received) {
+            Py_RETURN_NONE;
+        }
+
+        http->empty_body_received = 1;
+    }
+
     if (size > 0) {
         body = PyBytes_FromStringAndSize(NULL, size);
         if (nxt_slow_path(body == NULL)) {
@@ -442,6 +453,8 @@ nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict)
 
     if (more_body == NULL || more_body == Py_False) {
         http->complete = 1;
+
+        nxt_py_asgi_http_emit_disconnect(http);
     }
 
     Py_INCREF(http);
@@ -449,6 +462,41 @@ nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict)
 }
 
 
+static void
+nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http)
+{
+    PyObject  *msg, *future, *res;
+
+    if (http->receive_future == NULL) {
+        return;
+    }
+
+    msg = nxt_py_asgi_new_msg(http->req, nxt_py_http_disconnect_str);
+    if (nxt_slow_path(msg == NULL)) {
+        return;
+    }
+
+    if (msg == Py_None) {
+        Py_DECREF(msg);
+        return;
+    }
+
+    future = http->receive_future;
+    http->receive_future = NULL;
+
+    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL);
+    if (nxt_slow_path(res == NULL)) {
+        nxt_unit_req_alert(http->req, "'set_result' call failed");
+        nxt_python_print_exception();
+    }
+
+    Py_XDECREF(res);
+    Py_DECREF(future);
+
+    Py_DECREF(msg);
+}
+
+
 void
 nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req)
 {
@@ -573,7 +621,6 @@ fail:
 void
 nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req)
 {
-    PyObject            *msg, *future, *res;
     nxt_py_asgi_http_t  *http;
 
     http = req->data;
@@ -582,33 +629,7 @@ nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req)
 
     http->closed = 1;
 
-    if (http->receive_future == NULL) {
-        return;
-    }
-
-    msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str);
-    if (nxt_slow_path(msg == NULL)) {
-        return;
-    }
-
-    if (msg == Py_None) {
-        Py_DECREF(msg);
-        return;
-    }
-
-    future = http->receive_future;
-    http->receive_future = NULL;
-
-    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL);
-    if (nxt_slow_path(res == NULL)) {
-        nxt_unit_req_alert(req, "'set_result' call failed");
-        nxt_python_print_exception();
-    }
-
-    Py_XDECREF(res);
-    Py_DECREF(future);
-
-    Py_DECREF(msg);
+    nxt_py_asgi_http_emit_disconnect(http);
 }