Handle timeouts.

Pass timeout down to port level, and if the communication times out,
kill the port and let the supervisor restart it.

Also stub in select_count and param_query, conveniently used to test
timing out.

4 files changed, 53 insertions, 16 deletions

diff --git a/src/mysqlerl.c b/src/mysqlerl.c
index 5cec054..79731b7 100644
--- a/src/mysqlerl.c
+++ b/src/mysqlerl.c
@@ -13,7 +13,8 @@
 #include <string.h>
 
 const char *QUERY_MSG = "sql_query";
-const char *PARAM_QUERY_MSG = "param_query_msg";
+const char *PARAM_QUERY_MSG = "sql_param_query";
+const char *SELECT_COUNT_MSG = "sql_select_count";
 
 void
 usage()
@@ -138,7 +139,7 @@ handle_mysql_result(MYSQL_RES *result)
 }
 
 void
-handle_sql_query(MYSQL *dbh, ETERM *cmd)
+handle_query(MYSQL *dbh, ETERM *cmd)
 {
   ETERM *query, *resp;
   char *q;
@@ -174,8 +175,35 @@ handle_sql_query(MYSQL *dbh, ETERM *cmd)
 }
 
 void
-handle_sql_param_query(MYSQL *dbh, ETERM *cmd)
+handle_param_query(MYSQL *dbh, ETERM *msg)
 {
+  ETERM *query, *params;
+  char *q;
+
+  query = erl_element(2, msg);
+  params = erl_element(3, msg);
+  q = erl_iolist_to_string(query);
+  erl_free_term(query);
+  erl_free_term(params);
+
+  logmsg("DEBUG: got param query msg: %s.", q);
+
+  erl_free(q);
+}
+
+void
+handle_select_count(MYSQL *dbh, ETERM *msg)
+{
+  ETERM *query;
+  char *q;
+
+  query = erl_element(2, msg);
+  q = erl_iolist_to_string(query);
+  erl_free_term(query);
+
+  logmsg("DEBUG: got select count msg: %s.", q);
+
+  erl_free(q);
 }
 
 void
@@ -186,10 +214,13 @@ dispatch_db_cmd(MYSQL *dbh, ETERM *msg)
   tag = erl_element(1, msg);
   if (strncmp((char *)ERL_ATOM_PTR(tag),
               QUERY_MSG, strlen(QUERY_MSG)) == 0) {
-    handle_sql_query(dbh, msg);
+    handle_query(dbh, msg);
   } else if (strncmp((char *)ERL_ATOM_PTR(tag),
                      PARAM_QUERY_MSG, strlen(PARAM_QUERY_MSG)) == 0) {
-    handle_sql_param_query(dbh, msg);
+    handle_param_query(dbh, msg);
+  } else if (strncmp((char *)ERL_ATOM_PTR(tag),
+                     SELECT_COUNT_MSG, strlen(SELECT_COUNT_MSG)) == 0) {
+    handle_select_count(dbh, msg);
   } else {
     logmsg("WARNING: message type %s unknown.", (char *)ERL_ATOM_PTR(tag));
     erl_free_term(tag);
diff --git a/src/mysqlerl.erl b/src/mysqlerl.erl
index a5df819..113efee 100644
--- a/src/mysqlerl.erl
+++ b/src/mysqlerl.erl
@@ -115,7 +115,7 @@ first(Ref) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 first(Ref, Timeout) ->
-    gen_server:call(Ref, #sql_first{}, Timeout).
+    conn_fwd(Ref, #sql_first{}, Timeout).
 
 last(Ref) ->
     last(Ref, infinity).
@@ -127,7 +127,7 @@ last(Ref) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 last(Ref, Timeout) ->
-    gen_server:call(Ref, #sql_last{}, Timeout).
+    conn_fwd(Ref, #sql_last{}, Timeout).
 
 next(Ref) ->
     next(Ref, infinity).
@@ -139,7 +139,7 @@ next(Ref) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 next(Ref, Timeout) ->
-    gen_server:call(Ref, #sql_next{}, Timeout).
+    conn_fwd(Ref, #sql_next{}, Timeout).
 
 prev(Ref) ->
     prev(Ref, infinity).
@@ -151,7 +151,7 @@ prev(Ref) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 prev(Ref, Timeout) ->
-    gen_server:call(Ref, #sql_prev{}, Timeout).
+    conn_fwd(Ref, #sql_prev{}, Timeout).
 
 select_count(Ref, SQLQuery) ->
     select_count(Ref, SQLQuery, infinity).
@@ -164,7 +164,7 @@ select_count(Ref, SQLQuery) ->
 %%     {ok, NrRows} | {error, Reason}
 %%     NrRows = n_rows()
 select_count(Ref, SQLQuery, Timeout) ->
-    gen_server:call(Ref, #sql_select_count{q = SQLQuery}, Timeout).
+    conn_fwd(Ref, #sql_select_count{q = SQLQuery}, Timeout).
 
 select(Ref, Pos, N) ->
     select(Ref, Pos, N, infinity).
@@ -177,7 +177,7 @@ select(Ref, Pos, N) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 select(Ref, Pos, N, Timeout) ->
-    gen_server:call(Ref, #sql_select{pos = Pos, n = N}, Timeout).
+    conn_fwd(Ref, #sql_select{pos = Pos, n = N}, Timeout).
 
 param_query(Ref, SQLQuery, Params) ->
     param_query(Ref, SQLQuery, Params, infinity).
@@ -191,8 +191,7 @@ param_query(Ref, SQLQuery, Params) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 param_query(Ref, SQLQuery, Params, Timeout) ->
-    gen_server:call(Ref, #sql_param_query{q = SQLQuery, params = Params},
-                    Timeout).
+    conn_fwd(Ref, #sql_param_query{q = SQLQuery, params = Params}, Timeout).
 
 sql_query(Ref, SQLQuery) ->
     sql_query(Ref, SQLQuery, infinity).
@@ -205,7 +204,10 @@ sql_query(Ref, SQLQuery) ->
 %%     {selected, ColNames, Rows} | {error, Reason}
 %%     Rows = rows()
 sql_query(Ref, SQLQuery, Timeout) ->
-    gen_server:call(Ref, #sql_query{q = SQLQuery}, Timeout).
+    conn_fwd(Ref, #sql_query{q = SQLQuery}, Timeout).
+
+conn_fwd(Ref, Msg, Timeout) ->
+    gen_server:call(Ref, {Msg, Timeout}, infinity).
 
 %% Convert type needs some love! Cover all bases here instead of
 %% fudging.
diff --git a/src/mysqlerl_connection.erl b/src/mysqlerl_connection.erl
index b6015e3..ac0497a 100644
--- a/src/mysqlerl_connection.erl
+++ b/src/mysqlerl_connection.erl
@@ -45,7 +45,7 @@ handle_call(stop, _From, State) ->
     {stop, normal, State};
 handle_call(Request, _From, State) ->
     {reply, gen_server:call(port_ref(State#state.sup),
-                            #req{request = Request}), State}.
+                            #req{request = Request}, infinity), State}.
 
 handle_cast(_Request, State) ->
     {noreply, State}.
diff --git a/src/mysqlerl_port.erl b/src/mysqlerl_port.erl
index 572bc11..3593119 100644
--- a/src/mysqlerl_port.erl
+++ b/src/mysqlerl_port.erl
@@ -32,7 +32,8 @@ terminate(Reason, State) ->
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-handle_call(#req{request = Request}, From, #state{ref = Ref} = State) ->
+handle_call(#req{request = {Request, Timeout}}, From,
+            #state{ref = Ref} = State) ->
     io:format("DEBUG: Sending request: ~p~n", [Request]),
     port_command(Ref, term_to_binary(Request)),
     receive
@@ -46,6 +47,9 @@ handle_call(#req{request = Request}, From, #state{ref = Ref} = State) ->
                                      [Other]),
             gen_server:reply(From, {error, connection_closed}),
             {stop, {unknownreply, Other}, State}
+    after Timeout ->
+            gen_server:reply(From, timeout),
+            {stop, timeout, State}
     end.
 
 handle_cast(_Request, State) ->