this repo has no description
Add mutable transaction API
+96
+96
lib/hobbes.ex
+96
lib/hobbes.ex
···
1
1
defmodule Hobbes do
2
+
alias Hobbes.Transaction
3
+
alias Hobbes.Transaction.TxnState
4
+
5
+
@txn_key :_hobbes_active_transaction
6
+
7
+
defp get_txn, do: Process.get(@txn_key)
8
+
defp put_txn(%TxnState{} = txn), do: Process.put(@txn_key, txn)
9
+
defp delete_txn, do: Process.delete(@txn_key)
10
+
11
+
defp fetch_txn! do
12
+
case get_txn() do
13
+
nil -> raise "This function must be called within a transaction"
14
+
%TxnState{} = txn -> txn
15
+
end
16
+
end
17
+
18
+
@spec transaction(binary, (-> result), keyword) :: result when result: term
19
+
def transaction(cluster_name, fun, opts \\ []) do
20
+
case get_txn() do
21
+
nil -> put_txn(begin(cluster_name, opts))
22
+
# Nested transaction, do nothing
23
+
%TxnState{} -> :noop
24
+
end
25
+
26
+
try do
27
+
fun.()
28
+
commit(get_txn())
29
+
after
30
+
delete_txn()
31
+
end
32
+
end
33
+
34
+
defp begin(cluster_name, opts) do
35
+
case Transaction.new(cluster_name, opts) do
36
+
{:ok, %TxnState{} = txn} -> txn
37
+
{:error, _err} = error -> raise "Failed to begin transaction: #{inspect(error)}"
38
+
end
39
+
end
40
+
41
+
defp commit(%TxnState{} = txn) do
42
+
case Transaction.commit(txn) do
43
+
{:ok, %TxnState{} = _txn} -> :ok
44
+
{:error, _err} = error -> raise "Failed to commit transaction: #{inspect(error)}"
45
+
end
46
+
end
47
+
48
+
@spec read(binary) :: binary | nil
49
+
@spec read([binary]) :: %{binary => binary | nil}
50
+
def read(key_or_keys) do
51
+
case Transaction.read(fetch_txn!(), key_or_keys) do
52
+
{:ok, {result, %TxnState{} = txn}} ->
53
+
put_txn(txn)
54
+
result
55
+
{:error, _err} = error ->
56
+
raise "Failed to read: #{inspect(error)}"
57
+
end
58
+
end
59
+
60
+
@spec read_range(binary, binary) :: [{binary, binary}]
61
+
def read_range(start_key, end_key) do
62
+
case Transaction.read_range(fetch_txn!(), start_key, end_key) do
63
+
{:ok, {result, %TxnState{} = txn}} ->
64
+
put_txn(txn)
65
+
result
66
+
{:error, _err} = error ->
67
+
raise "Failed to read: #{inspect(error)}"
68
+
end
69
+
end
70
+
71
+
@spec write(binary, binary) :: :ok
72
+
def write(key, value) do
73
+
%TxnState{} = txn = Transaction.write(fetch_txn!(), key, value)
74
+
put_txn(txn)
75
+
:ok
76
+
end
77
+
78
+
@spec write([{binary, binary}]) :: :ok
79
+
def write(pairs) when is_list(pairs) do
80
+
%TxnState{} = txn = Transaction.write(fetch_txn!(), pairs)
81
+
put_txn(txn)
82
+
:ok
83
+
end
84
+
85
+
@spec clear(binary) :: :ok
86
+
def clear(key) do
87
+
%TxnState{} = txn = Transaction.clear(fetch_txn!(), key)
88
+
put_txn(txn)
89
+
:ok
90
+
end
91
+
92
+
@spec clear_range(binary, binary) :: :ok
93
+
def clear_range(start_key, end_key) do
94
+
%TxnState{} = txn = Transaction.clear_range(fetch_txn!(), start_key, end_key)
95
+
put_txn(txn)
96
+
:ok
97
+
end
2
98
end