Removed query queue, save only one query at the time

Author: dankmolot

source/async_postgres.hpp

@@ -89,7 +89,7 @@ namespace async_postgres {
 
     struct Connection {
         pg::conn conn;
-        std::queue<Query> queries;
+        std::optional<Query> query;
         std::optional<ResetEvent> reset_event;
         GLua::AutoReference on_notify;
 
@@ -113,7 +113,7 @@ namespace async_postgres {
     void process_notifications(GLua::ILuaInterface* lua, Connection* state);
 
     // query.cpp
-    void process_queries(GLua::ILuaInterface* lua, Connection* state);
+    void process_query(GLua::ILuaInterface* lua, Connection* state);
 
     // result.cpp
     void create_result_table(GLua::ILuaInterface* lua, PGresult* result);

source/main.cpp

@@ -23,7 +23,7 @@ namespace async_postgres::lua {
 
             async_postgres::process_reset(lua, state);
             async_postgres::process_notifications(lua, state);
-            async_postgres::process_queries(lua, state);
+            async_postgres::process_query(lua, state);
         }
 
         return 0;
@@ -47,14 +47,18 @@ namespace async_postgres::lua {
         lua->CheckType(3, GLua::Type::Function);
 
         auto state = lua_connection_state();
+        if (state->query) {
+            throw std::runtime_error("query already in progress");
+        }
 
         async_postgres::SimpleCommand command = {lua->GetString(2)};
         async_postgres::Query query = {std::move(command)};
-        if (!lua->IsType(3, GLua::Type::Nil)) {
+
+        if (lua->IsType(3, GLua::Type::Function)) {
             query.callback = GLua::AutoReference(lua, 3);
         }
 
-        state->queries.push(std::move(query));
+        state->query = std::move(query);
         return 0;
     }
 
@@ -65,15 +69,21 @@ namespace async_postgres::lua {
         lua->CheckType(4, GLua::Type::Function);
 
         auto state = lua_connection_state();
+        if (state->query) {
+            throw std::runtime_error("query already in progress");
+        }
 
         async_postgres::ParameterizedCommand command = {
             lua->GetString(2),
             async_postgres::array_to_params(lua, 3),
         };
-
         async_postgres::Query query = {std::move(command)};
-        query.callback = GLua::AutoReference(lua, 4);
-        state->queries.push(std::move(query));
+
+        if (lua->IsType(4, GLua::Type::Function)) {
+            query.callback = GLua::AutoReference(lua, 4);
+        }
+
+        state->query = std::move(query);
 
         return 0;
     }
@@ -82,7 +92,7 @@ namespace async_postgres::lua {
         lua->CheckType(1, async_postgres::connection_meta);
 
         GLua::AutoReference callback;
-        if (!lua->IsType(2, GLua::Type::Nil)) {
+        if (lua->IsType(2, GLua::Type::Function)) {
             callback = GLua::AutoReference(lua, 2);
         }
 

source/notifications.cpp

@@ -13,8 +13,7 @@ void async_postgres::process_notifications(GLua::ILuaInterface* lua,
         return;
     }
 
-    if (state->queries.empty() &&
-        check_socket_status(state->conn.get()).read_ready &&
+    if (!state->query && check_socket_status(state->conn.get()).read_ready &&
         PQconsumeInput(state->conn.get()) == 0) {
         // we consumed input
         // but there was some error

source/query.cpp

@@ -18,11 +18,23 @@ inline bool send_query(PGconn* conn, Query& query) {
     return false;
 }
 
-void query_failed(GLua::ILuaInterface* lua, PGconn* conn, Query& query) {
+// This function will remove the query from the connection state
+// and call the callback with the error message
+void query_failed(GLua::ILuaInterface* lua, Connection* state) {
+    if (!state->query) {
+        return;
+    }
+
+    auto query = std::move(*state->query);
+    lua->Msg("[async_postgres] query failed: current query %d\n",
+             state->query.has_value());
+
+    state->query.reset();
+
     if (query.callback) {
         query.callback.Push();
         lua->PushBool(false);
-        lua->PushString(PQerrorMessage(conn));
+        lua->PushString(PQerrorMessage(state->conn.get()));
         pcall(lua, 2, 0);
     }
 }
@@ -54,42 +66,37 @@ bool bad_result(PGresult* result) {
            status == PGRES_FATAL_ERROR;
 }
 
-void async_postgres::process_queries(GLua::ILuaInterface* lua,
-                                     Connection* state) {
-    if (state->queries.empty()) {
+void async_postgres::process_query(GLua::ILuaInterface* lua,
+                                   Connection* state) {
+    if (!state->query || state->reset_event) {
         // no queries to process
-        return;
-    }
-
-    if (state->reset_event) {
         // don't process queries while reconnecting
         return;
     }
 
-    auto& query = state->queries.front();
+    auto& query = state->query.value();
     if (!query.sent) {
         if (!send_query(state->conn.get(), query)) {
-            query_failed(lua, state->conn.get(), query);
-            state->queries.pop();
-            return process_queries(lua, state);
+            query_failed(lua, state);
+            return process_query(lua, state);
         }
 
         query.sent = true;
         query.flushed = PQflush(state->conn.get()) == 0;
     }
 
     if (!poll_query(state->conn.get(), query)) {
-        query_failed(lua, state->conn.get(), query);
-        state->queries.pop();
-        return process_queries(lua, state);
+        query_failed(lua, state);
+        return process_query(lua, state);
     }
 
     while (PQisBusy(state->conn.get()) == 0) {
         auto result = pg::getResult(state->conn);
         if (!result) {
             // query is done
-            state->queries.pop();
-            return process_queries(lua, state);
+            // TODO: remove query if we have a final result
+            state->query.reset();
+            return process_query(lua, state);
         }
 
         if (query.callback) {