subcribe modify delete script

Author: wallyatgithub

common_lua.cc

@@ -18,6 +18,8 @@ extern "C" {
 #include "lauxlib.h"
 }
 #include "common_lua.h"
+#include "url-parser/url_parser.h"
+#include "h2load_utils.h"
 
 
 
@@ -207,13 +209,70 @@ void register_3rd_party_lib_func_to_lua(lua_State* L)
     lua_settop(L, 0);
 }
 
+int parse_uri(lua_State* L)
+{
+    const std::string scheme_header = ":scheme";
+    const std::string path_header = ":path";
+    const std::string authority_header = ":authority";
+
+    std::string uri;
+    if ((lua_gettop(L) == 1))
+    {
+        size_t len;
+        const char* str = lua_tolstring(L, -1, &len);
+        uri.assign(str, len);
+        lua_pop(L, 1);
+    }
+    else
+    {
+        std::cerr << __FUNCTION__ << " invalid arguments" << std::endl;
+        lua_settop(L, 0);
+    }
+    if (uri.size())
+    {
+        http_parser_url u {};
+        if (http_parser_parse_url(uri.c_str(), uri.size(), 0, &u) == 0)
+        {
+            std::string path = get_reqline(uri.c_str(), u);
+            std::string schema;
+            std::string host;
+            if (util::has_uri_field(u, UF_SCHEMA) && util::has_uri_field(u, UF_HOST))
+            {
+                schema = util::get_uri_field(uri.c_str(), u, UF_SCHEMA).str();
+                host = util::get_uri_field(uri.c_str(), u, UF_HOST).str();
+                if (util::has_uri_field(u, UF_PORT))
+                {
+                    host.append(":").append(util::utos(u.port));
+                }
+            }
+            lua_createtable(L, 0, schema.empty() ? 1 : 3);
+            lua_pushlstring(L, path_header.c_str(), path_header.size());
+            lua_pushlstring(L, path.c_str(), path.size());
+            lua_rawset(L, -3);
+            if (schema.size())
+            {
+                lua_pushlstring(L, scheme_header.c_str(), scheme_header.size());
+                lua_pushlstring(L, schema.c_str(), schema.size());
+                lua_rawset(L, -3);
+                lua_pushlstring(L, authority_header.c_str(), authority_header.size());
+                lua_pushlstring(L, host.c_str(), host.size());
+                lua_rawset(L, -3);
+            }
+            return 1;
+        }
+    }
+    return 0;
+}
+
+
 void init_new_lua_state_with_common_apis(lua_State* L)
 {
     lua_register(L, "time_since_epoch", time_since_epoch);
     lua_register(L, "store_value", store_value);
     lua_register(L, "get_value", get_value);
     lua_register(L, "delete_value", delete_value);
     lua_register(L, "generate_uuid_v4", generate_uuid);
+    lua_register(L, "parse_uri", parse_uri);
     register_3rd_party_lib_func_to_lua(L);
 }
 

common_lua.h

@@ -20,6 +20,8 @@ extern "C"
 
     int generate_uuid(lua_State* L);
 
+    int parse_uri(lua_State* L);
+
 
     // from pb.so
     LUALIB_API int luaopen_pb_io(lua_State* L);

examples/sub_create_update_notify_delete.lua

@@ -0,0 +1,106 @@
+local server_id = start_server("maock_subscribe_update_delete.json")
+
+register_service_handler(server_id, "subscribe", "handle_subscribe", 20)
+
+register_service_handler(server_id, "subs-update", "handle_subscribe_update", 20)
+
+register_service_handler(server_id, "subs-del", "handle_ubsubscribe", 20)
+
+local location_header_with_subs_id = {"/some-hardcoded-api-root/some-service/v2/some-operation", "/", "subs_id_placeholder"}
+
+local timestamp_reporting_stat = 0
+
+local pseudo_uuid_as_worker_thread_id_for_stats_output = generate_uuid_v4()
+
+local number_of_notify_request_sent = 0
+
+local number_of_notify_request_sent_last_second = 0
+
+math.randomseed(os.time())
+
+-- utility functions begin
+local function tokenize_path_and_query(path)
+    path = path .. "?"
+    results = string.gmatch(path, '([^?]+)')
+    tokens = {}
+    for res in results do
+        table.insert(tokens, res)
+    end
+    return tokens
+end
+
+local function tokenize_path(path)
+    path = path .. "/"
+    results = string.gmatch(path, '([^/]+)')
+    tokens = {}
+    for res in results do
+        table.insert(tokens, res)
+    end
+    return tokens
+end
+-- utility functions end
+
+function handle_subscribe(response_addr, request_headers, request_payload)
+    subscription_id = generate_uuid_v4()
+    store_value(subscription_id, request_payload)
+    location_header_with_subs_id[2] = subscription_id
+    response_header = {[":status"] = "201", ["location"] = table.concat(location_header_with_subs_id)}
+    send_response(response_addr, response_header, "")
+end
+
+function handle_subscribe_update(response_addr, request_headers, request_payload)
+    path = headers[":path"]
+    tokens = tokenize_path_and_query(path)
+    path_without_query = tokens[1]
+    path_tokens = tokenize_path(path_without_query)
+    subscription_id = path_tokens[table.getn(path_tokens)]
+    subscribe_request_payload = get_value(subscription_id)
+    if (subscribe_request_payload == nil)
+    then
+        response_header = {[":status"] = "404"}
+        send_response(response_addr, response_header, "")
+        return
+    end
+    response_header = {[":status"] = "200"}
+    send_response(response_addr, response_header, '{"status": "success"}')
+
+    -- send notify
+    doc = rapidjson.Document()
+    ok, message = doc:parse(subscribe_request_payload)
+    notify_uri = doc:get("/callback-uri")
+    
+    notify_headers = parse_uri(notify_uri)
+    
+    notify_response_headers, notify_response_payload = send_http_request_and_await_response(notify_headers, "")
+    
+    if (notify_response_headers ~= nil)
+    then
+        number_of_notify_request_sent = number_of_notify_request_sent + 1
+    end
+    now = time_since_epoch()
+    if (now - timestamp_reporting_stat > 1000)
+    then
+        output = string.format("thread: %s, number request sent: %d, tps = %d", pseudo_uuid_as_worker_thread_id_for_stats_output, number_of_notify_request_sent, ((number_of_request_sent - number_of_notify_request_sent_last_second)*1000)/(now - timestamp_reporting_stat))
+        number_of_notify_request_sent_last_second = number_of_notify_request_sent;
+        timestamp_reporting_stat = now
+        print (output)
+    end
+end
+
+function handle_ubsubscribe(response_addr, request_headers, request_payload)
+    path = headers[":path"]
+    tokens = tokenize_path_and_query(path)
+    path_without_query = tokens[1]
+    path_tokens = tokenize_path(path_without_query)
+    subscription_id = path_tokens[table.getn(path_tokens)]
+    subscribe_request_payload = get_value(subscription_id)
+    if (subscribe_request_payload == nil)
+    then
+        response_header = {[":status"] = "404"}
+        send_response(response_addr, response_header, "")
+        return
+    end
+    response_header = {[":status"] = "204"}
+    delete_value(subscription_id)
+    send_response(response_addr, response_header, "")
+end