Add ability to subscribe to table updates.

This was implemented based on
[this](https://www.sqlite.org/c3ref/update_hook.html) document.
This commit is contained in:
Connor Rigby
2018-11-29 13:29:15 -08:00
parent ab25c2d2b6
commit 8fa063a3f1
4 changed files with 307 additions and 187 deletions

View File

@@ -35,6 +35,7 @@ static ErlNifResourceType *esqlite_statement_type = NULL;
typedef struct { typedef struct {
ErlNifTid tid; ErlNifTid tid;
ErlNifThreadOpts* opts; ErlNifThreadOpts* opts;
ErlNifPid notification_pid;
sqlite3 *db; sqlite3 *db;
queue *commands; queue *commands;
@@ -50,6 +51,8 @@ typedef struct {
typedef enum { typedef enum {
cmd_unknown, cmd_unknown,
cmd_open, cmd_open,
cmd_update_hook_set,
cmd_notification,
cmd_exec, cmd_exec,
cmd_changes, cmd_changes,
cmd_prepare, cmd_prepare,
@@ -75,6 +78,8 @@ typedef struct {
static ERL_NIF_TERM atom_esqlite3; static ERL_NIF_TERM atom_esqlite3;
static ERL_NIF_TERM push_command(ErlNifEnv *env, esqlite_connection *conn, esqlite_command *cmd);
static ERL_NIF_TERM static ERL_NIF_TERM
make_atom(ErlNifEnv *env, const char *atom_name) 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"); 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 static ERL_NIF_TERM
@@ -641,6 +694,8 @@ evaluate_command(esqlite_command *cmd, esqlite_connection *conn)
switch(cmd->type) { switch(cmd->type) {
case cmd_open: case cmd_open:
return do_open(cmd->env, conn, cmd->arg); 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: case cmd_exec:
return do_exec(cmd->env, conn, cmd->arg); return do_exec(cmd->env, conn, cmd->arg);
case cmd_changes: case cmd_changes:
@@ -692,6 +747,8 @@ esqlite_connection_run(void *arg)
if(cmd->type == cmd_stop) { if(cmd->type == cmd_stop) {
continue_running = 0; continue_running = 0;
} else if(cmd->type == cmd_notification) {
enif_send(NULL, &db->notification_pid, cmd->env, cmd->arg);
} else { } else {
enif_send(NULL, &cmd->pid, cmd->env, make_answer(cmd, evaluate_command(cmd, db))); 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); 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 * 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[] = { static ErlNifFunc nif_funcs[] = {
{"start", 0, esqlite_start}, {"start", 0, esqlite_start},
{"open", 4, esqlite_open}, {"open", 4, esqlite_open},
{"set_update_hook", 4, set_update_hook},
{"exec", 4, esqlite_exec}, {"exec", 4, esqlite_exec},
{"changes", 3, esqlite_changes}, {"changes", 3, esqlite_changes},
{"prepare", 4, esqlite_prepare}, {"prepare", 4, esqlite_prepare},

View File

@@ -22,6 +22,7 @@
%% higher-level export %% higher-level export
-export([open/1, open/2, -export([open/1, open/2,
set_update_hook/2, set_update_hook/3,
exec/2, exec/3, exec/2, exec/3,
changes/1, changes/2, changes/1, changes/2,
insert/2, insert/2,
@@ -69,6 +70,22 @@ open(Filename, Timeout) ->
Error Error
end. 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. %% @doc Execute a sql statement, returns a list with tuples.
-spec q(sql(), connection()) -> list(tuple()) | {error, term()}. -spec q(sql(), connection()) -> list(tuple()) | {error, term()}.
q(Sql, Connection) -> q(Sql, Connection) ->

View File

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

View File

@@ -20,6 +20,19 @@ open_multiple_different_databases_test() ->
{ok, _C2} = esqlite3:open("test2.db"), {ok, _C2} = esqlite3:open("test2.db"),
ok. 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() -> simple_query_test() ->
{ok, Db} = esqlite3:open(":memory:"), {ok, Db} = esqlite3:open(":memory:"),
ok = esqlite3:exec("begin;", Db), ok = esqlite3:exec("begin;", Db),
@@ -367,4 +380,3 @@ garbage_collect_test() ->