Added basic HTTP request processing in router.

- request to connection mapping in engine;
- requests queue in connection;
- engine port creation;
- connected ports hash for each process;
- engine port data messages processing (app responses);

1 files changed, 112 insertions, 0 deletions

diff --git a/src/nxt_event_engine.c b/src/nxt_event_engine.c
index d636dc25..9171bcd6 100644
--- a/src/nxt_event_engine.c
+++ b/src/nxt_event_engine.c
@@ -544,3 +544,115 @@ nxt_event_engine_start(nxt_event_engine_t *engine)
         nxt_timer_expire(engine, now);
     }
 }
+
+
+static nxt_int_t
+nxt_req_conn_test(nxt_lvlhsh_query_t *lhq, void *data)
+{
+    return NXT_OK;
+}
+
+static const nxt_lvlhsh_proto_t  lvlhsh_req_conn_proto  nxt_aligned(64) = {
+    NXT_LVLHSH_DEFAULT,
+    nxt_req_conn_test,
+    nxt_lvlhsh_alloc,
+    nxt_lvlhsh_free,
+};
+
+
+void
+nxt_event_engine_request_add(nxt_event_engine_t *engine,
+    nxt_req_conn_link_t *rc)
+{
+    nxt_lvlhsh_query_t  lhq;
+
+    lhq.key_hash = nxt_murmur_hash2(&rc->req_id, sizeof(rc->req_id));
+    lhq.key.length = sizeof(rc->req_id);
+    lhq.key.start = (u_char *) &rc->req_id;
+    lhq.proto = &lvlhsh_req_conn_proto;
+    lhq.replace = 0;
+    lhq.value = rc;
+    lhq.pool = engine->mem_pool;
+
+    switch (nxt_lvlhsh_insert(&engine->requests, &lhq)) {
+
+    case NXT_OK:
+        break;
+
+    default:
+        nxt_thread_log_error(NXT_LOG_WARN, "req %08uxD to conn add failed",
+                             rc->req_id);
+        break;
+    }
+}
+
+
+nxt_req_conn_link_t *
+nxt_event_engine_request_find(nxt_event_engine_t *engine, nxt_req_id_t req_id)
+{
+    nxt_lvlhsh_query_t  lhq;
+
+    lhq.key_hash = nxt_murmur_hash2(&req_id, sizeof(req_id));
+    lhq.key.length = sizeof(req_id);
+    lhq.key.start = (u_char *) &req_id;
+    lhq.proto = &lvlhsh_req_conn_proto;
+
+    if (nxt_lvlhsh_find(&engine->requests, &lhq) == NXT_OK) {
+        return lhq.value;
+    }
+
+    return NULL;
+}
+
+
+void
+nxt_event_engine_request_remove(nxt_event_engine_t *engine,
+    nxt_req_conn_link_t *rc)
+{
+    nxt_lvlhsh_query_t  lhq;
+
+    lhq.key_hash = nxt_murmur_hash2(&rc->req_id, sizeof(rc->req_id));
+    lhq.key.length = sizeof(rc->req_id);
+    lhq.key.start = (u_char *) &rc->req_id;
+    lhq.proto = &lvlhsh_req_conn_proto;
+    lhq.pool = engine->mem_pool;
+
+    switch (nxt_lvlhsh_delete(&engine->requests, &lhq)) {
+
+    case NXT_OK:
+        break;
+
+    default:
+        nxt_thread_log_error(NXT_LOG_WARN, "req %08uxD to conn remove failed",
+                             rc->req_id);
+        break;
+    }
+}
+
+
+nxt_req_conn_link_t *
+nxt_event_engine_request_find_remove(nxt_event_engine_t *engine,
+    nxt_req_id_t req_id)
+{
+    nxt_lvlhsh_query_t  lhq;
+
+    lhq.key_hash = nxt_murmur_hash2(&req_id, sizeof(req_id));
+    lhq.key.length = sizeof(req_id);
+    lhq.key.start = (u_char *) &req_id;
+    lhq.proto = &lvlhsh_req_conn_proto;
+    lhq.pool = engine->mem_pool;
+
+    switch (nxt_lvlhsh_delete(&engine->requests, &lhq)) {
+
+    case NXT_OK:
+        return lhq.value;
+
+    default:
+        nxt_thread_log_error(NXT_LOG_WARN, "req %08uxD to conn remove failed",
+                             req_id);
+        break;
+    }
+
+    return NULL;
+}
+