Merge pull request #48 from ConnorRigby/notifications

Add ability to subscribe to table updates.

c_src/esqlite3_nif.c

@@ -35,6 +35,7 @@ static ErlNifResourceType *esqlite_statement_type = NULL;
 typedef struct {
     ErlNifTid tid;
     ErlNifThreadOpts* opts;
+    ErlNifPid notification_pid;
 
     sqlite3 *db;
     queue *commands;
@@ -50,6 +51,8 @@ typedef struct {
 typedef enum {
     cmd_unknown,
     cmd_open,
+    cmd_update_hook_set,
+    cmd_notification,
     cmd_exec,
     cmd_changes,
     cmd_prepare,
@@ -75,6 +78,8 @@ typedef struct {
 
 static ERL_NIF_TERM atom_esqlite3;
 
+static ERL_NIF_TERM push_command(ErlNifEnv *env, esqlite_connection *conn, esqlite_command *cmd);
+
 static ERL_NIF_TERM
 make_atom(ErlNifEnv *env, const char *atom_name)
 {
@@ -267,6 +272,54 @@ do_open(ErlNifEnv *env, esqlite_connection *db, const ERL_NIF_TERM arg)
     return make_atom(env, "ok");
 }
 
+void
+update_callback(void *arg, int sqlite_operation_type, char const *sqlite_database, char const *sqlite_table, sqlite3_int64 sqlite_rowid)
+{
+    esqlite_connection *db = (esqlite_connection *)arg;
+    esqlite_command *cmd = NULL;
+    ERL_NIF_TERM type, table, rowid;
+    cmd = command_create();
+
+    if(db == NULL)
+        return;
+
+    if(!cmd)
+	    return;
+
+    rowid = enif_make_int64(cmd->env, sqlite_rowid);
+    table = enif_make_string(cmd->env, sqlite_table, ERL_NIF_LATIN1);
+
+    switch(sqlite_operation_type) {
+        case SQLITE_INSERT:
+            type = make_atom(cmd->env, "insert");
+            break;
+        case SQLITE_DELETE:
+            type = make_atom(cmd->env, "delete");
+            break;
+        case SQLITE_UPDATE:
+            type = make_atom(cmd->env, "update");
+            break;
+        default:
+            return;
+    }
+    cmd->type = cmd_notification;
+    cmd->arg = enif_make_tuple3(cmd->env, type, table, rowid);
+    push_command(cmd->env, db, cmd);
+}
+
+static ERL_NIF_TERM
+do_set_update_hook(ErlNifEnv *env, esqlite_connection *db, const ERL_NIF_TERM arg)
+{
+    if(!enif_get_local_pid(env, arg, &db->notification_pid))
+	    return make_error_tuple(env, "invalid_pid");
+
+    sqlite3_update_hook(db->db, NULL, NULL);
+    if(sqlite3_update_hook(db->db, update_callback, db) != SQLITE_OK)
+        return make_error_tuple(env, "sqlite3_update_hook_fail");
+
+    return make_atom(env, "ok");
+}
+
 /*
  */
 static ERL_NIF_TERM
@@ -641,6 +694,8 @@ evaluate_command(esqlite_command *cmd, esqlite_connection *conn)
     switch(cmd->type) {
     case cmd_open:
 	    return do_open(cmd->env, conn, cmd->arg);
+    case cmd_update_hook_set:
+        return do_set_update_hook(cmd->env, conn, cmd->arg);
     case cmd_exec:
 	    return do_exec(cmd->env, conn, cmd->arg);
     case cmd_changes:
@@ -692,6 +747,8 @@ esqlite_connection_run(void *arg)
 
 	    if(cmd->type == cmd_stop) {
 	        continue_running = 0;
+        } else if(cmd->type == cmd_notification) {
+            enif_send(NULL, &db->notification_pid, cmd->env, cmd->arg);
         } else {
 	        enif_send(NULL, &cmd->pid, cmd->env, make_answer(cmd, evaluate_command(cmd, db)));
         }
@@ -773,6 +830,35 @@ esqlite_open(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
     return push_command(env, db, cmd);
 }
 
+static ERL_NIF_TERM
+set_update_hook(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+    esqlite_connection *db;
+    esqlite_command *cmd = NULL;
+    ErlNifPid pid;
+
+    if(argc != 4)
+	    return enif_make_badarg(env);
+    if(!enif_get_resource(env, argv[0], esqlite_connection_type, (void **) &db))
+	    return enif_make_badarg(env);
+    if(!enif_is_ref(env, argv[1]))
+	    return make_error_tuple(env, "invalid_ref");
+    if(!enif_get_local_pid(env, argv[2], &pid))
+	    return make_error_tuple(env, "invalid_pid");
+
+    cmd = command_create();
+    if(!cmd)
+	    return make_error_tuple(env, "command_create_failed");
+
+    /* command */
+    cmd->type = cmd_update_hook_set;
+    cmd->ref = enif_make_copy(cmd->env, argv[1]);
+    cmd->pid = pid;
+    cmd->arg = enif_make_copy(cmd->env, argv[3]);
+
+    return push_command(env, db, cmd);
+}
+
 /*
  * Execute the sql statement
  */
@@ -1154,6 +1240,7 @@ static int on_upgrade(ErlNifEnv* env, void** priv, void** old_priv_data, ERL_NIF
 static ErlNifFunc nif_funcs[] = {
     {"start", 0, esqlite_start},
     {"open", 4, esqlite_open},
+    {"set_update_hook", 4, set_update_hook},
     {"exec", 4, esqlite_exec},
     {"changes", 3, esqlite_changes},
     {"prepare", 4, esqlite_prepare},

src/esqlite3.erl

@@ -22,6 +22,7 @@
 
 %% higher-level export
 -export([open/1, open/2,
+         set_update_hook/2, set_update_hook/3,
          exec/2, exec/3,
          changes/1, changes/2,
          insert/2,
@@ -69,6 +70,22 @@ open(Filename, Timeout) ->
             Error
     end.
 
+%% @doc Subscribe to database notifications
+%% Messages will come in the shape {action, table, id}
+%% Where action will be insert | update | delete
+%% and table will be a string
+%% and id will be an integer
+%%
+-spec set_update_hook(pid(), connection()) -> ok | {error, term()}.
+set_update_hook(Pid, Connection) ->
+    set_update_hook(Pid, Connection, ?DEFAULT_TIMEOUT).
+
+-spec set_update_hook(pid(), connection(), timeout()) -> ok | {error, term()}.
+set_update_hook(Pid, {connection, _Ref, Connection}, Timeout) ->
+    Ref = make_ref(),
+    ok = esqlite3_nif:set_update_hook(Connection, Ref, self(), Pid),
+    receive_answer(Ref, Timeout).
+
 %% @doc Execute a sql statement, returns a list with tuples.
 -spec q(sql(), connection()) -> list(tuple()) | {error, term()}.
 q(Sql, Connection) ->

src/esqlite3_nif.erl

@@ -23,6 +23,7 @@
 %% low-level exports
 -export([start/0,
          open/4,
+         set_update_hook/4,
          exec/4,
          changes/3,
          insert/4,
@@ -62,6 +63,9 @@ start() ->
 open(_Db, _Ref, _Dest, _Filename) ->
     erlang:nif_error(nif_library_not_loaded).
 
+set_update_hook(_Db, _Ref, _Dest, _Pid) ->
+    erlang:nif_error(nif_library_not_loaded).
+
 %% @doc Exec the query.
 %%
 %% Sends an asynchronous exec command over the connection and returns

test/esqlite_test.erl

@@ -20,6 +20,19 @@ open_multiple_different_databases_test() ->
     {ok, _C2} = esqlite3:open("test2.db"),
     ok.
 
+update_hook_test() ->
+    {ok, Db} = esqlite3:open(":memory:"),
+    ok = esqlite3:set_update_hook(self(), Db),
+    ok = esqlite3:exec("CREATE TABLE test (id INTEGER PRIMARY KEY, val STRING);", Db),
+    ok = esqlite3:exec("INSERT INTO test (val) VALUES ('this is a test');", Db),
+    ok = receive {insert, "test", 1} -> ok after 150 -> no_message end,
+    ok = esqlite3:exec("UPDATE test SET val = 'a new test' WHERE id = 1;", Db),
+    ok = receive {update, "test", 1} -> ok after 150 -> no_message end,
+    ok = esqlite3:exec("DELETE FROM test WHERE id = 1;", Db),
+    ok = receive {delete, "test", 1} -> ok after 150 -> no_message end,
+    ok.
+
+
 simple_query_test() ->
     {ok, Db} = esqlite3:open(":memory:"),
     ok = esqlite3:exec("begin;", Db),
@@ -367,4 +380,3 @@ garbage_collect_test() ->
 
 
 
-