Current section

Files

Jump to
radish src radish.erl
Raw

src/radish.erl

-module(radish).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([start/3, shutdown/1, keys/3, scan/4, scan_pattern/5, scan_with_type/5, scan_pattern_with_type/6, exists/3, get/3, mget/3, append/4, set/4, set_new/4, set_existing/4, mset/3, del/3, incr/3, incr_by/4, incr_by_float/4, decr/3, decr_by/4, random_key/2, key_type/3, rename/4, renamenx/4, persist/3, expire/4, expire_if/5, publish/4, subscribe/5, subscribe_to_patterns/5]).
-export_type([start_option/0, key_type/0, expire_condition/0, next/0]).
-type start_option() :: {timeout, integer()} |
{auth, binary()} |
{auth_with_username, binary(), binary()}.
-type key_type() :: set | list | z_set | hash | string | stream.
-type expire_condition() :: nx | xx | gt | lt.
-type next() :: continue |
unsubscribe_from_all |
{unsubscribe_from, list(binary())}.
-spec start(binary(), integer(), list(start_option())) -> {ok,
gleam@erlang@process:subject(radish@client:message())} |
{error, gleam@otp@actor:start_error()}.
start(Host, Port, Options) ->
{Timeout@1, Options@1} = case gleam@list:pop_map(
Options,
fun(Item) -> case Item of
{timeout, Timeout} ->
{ok, Timeout};
_ ->
{error, nil}
end end
) of
{ok, Result} ->
Result;
{error, nil} ->
{1024, Options}
end,
gleam@result:then(
radish@client:start(Host, Port, Timeout@1),
fun(Client) ->
Options@2 = gleam@list:map(Options@1, fun(Item@1) -> case Item@1 of
{auth, Password} ->
{auth, Password};
{auth_with_username, Username, Password@1} ->
{auth_with_username, Username, Password@1};
{timeout, _} ->
{auth_with_username, <<""/utf8>>, <<""/utf8>>}
end end),
gleam@result:then(
begin
_pipe = radish@utils:execute(
Client,
radish@command:hello(3, Options@2),
Timeout@1
),
gleam@result:map_error(_pipe, fun(Error) -> case Error of
{server_error, Error@1} ->
{init_failed, {abnormal, Error@1}};
_ ->
{init_failed,
{abnormal,
<<"Failed to say hello"/utf8>>}}
end end)
end,
fun(_) -> {ok, Client} end
)
end
).
-spec shutdown(gleam@erlang@process:subject(radish@client:message())) -> nil.
shutdown(Client) ->
gleam@erlang@process:send(Client, shutdown).
-spec keys(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer()
) -> {ok, list(binary())} | {error, radish@error:error()}.
keys(Client, Pattern, Timeout) ->
_pipe = radish@command:keys(Pattern),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{array, Array}] ->
gleam@list:try_map(Array, fun(Item) -> case Item of
{bulk_string, Value@1} ->
{ok, Value@1};
_ ->
{error, resp_error}
end end);
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec scan(
gleam@erlang@process:subject(radish@client:message()),
integer(),
integer(),
integer()
) -> {ok, {list(binary()), integer()}} | {error, radish@error:error()}.
scan(Client, Cursor, Count, Timeout) ->
_pipe = radish@command:scan(Cursor, Count),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{array, [{bulk_string, New_cursor_str}, {array, Keys}]}] ->
case gleam@int:parse(New_cursor_str) of
{ok, New_cursor} ->
gleam@result:then(
gleam@list:try_map(
Keys,
fun(Item) -> case Item of
{bulk_string, Value@1} ->
{ok, Value@1};
_ ->
{error, resp_error}
end end
),
fun(Array) -> {ok, {Array, New_cursor}} end
);
{error, nil} ->
{error, resp_error}
end;
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec scan_pattern(
gleam@erlang@process:subject(radish@client:message()),
integer(),
binary(),
integer(),
integer()
) -> {ok, {list(binary()), integer()}} | {error, radish@error:error()}.
scan_pattern(Client, Cursor, Pattern, Count, Timeout) ->
_pipe = radish@command:scan_pattern(Cursor, Pattern, Count),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{array, [{bulk_string, New_cursor_str}, {array, Keys}]}] ->
case gleam@int:parse(New_cursor_str) of
{ok, New_cursor} ->
gleam@result:then(
gleam@list:try_map(
Keys,
fun(Item) -> case Item of
{bulk_string, Value@1} ->
{ok, Value@1};
_ ->
{error, resp_error}
end end
),
fun(Array) -> {ok, {Array, New_cursor}} end
);
{error, nil} ->
{error, resp_error}
end;
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec scan_with_type(
gleam@erlang@process:subject(radish@client:message()),
integer(),
key_type(),
integer(),
integer()
) -> {ok, {list(binary()), integer()}} | {error, radish@error:error()}.
scan_with_type(Client, Cursor, Key_type, Count, Timeout) ->
_pipe = case Key_type of
set ->
radish@command:scan_with_type(Cursor, <<"set"/utf8>>, Count);
list ->
radish@command:scan_with_type(Cursor, <<"list"/utf8>>, Count);
z_set ->
radish@command:scan_with_type(Cursor, <<"zset"/utf8>>, Count);
hash ->
radish@command:scan_with_type(Cursor, <<"hash"/utf8>>, Count);
string ->
radish@command:scan_with_type(Cursor, <<"string"/utf8>>, Count);
stream ->
radish@command:scan_with_type(Cursor, <<"stream"/utf8>>, Count)
end,
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{array, [{bulk_string, New_cursor_str}, {array, Keys}]}] ->
case gleam@int:parse(New_cursor_str) of
{ok, New_cursor} ->
gleam@result:then(
gleam@list:try_map(
Keys,
fun(Item) -> case Item of
{bulk_string, Value@1} ->
{ok, Value@1};
_ ->
{error, resp_error}
end end
),
fun(Array) -> {ok, {Array, New_cursor}} end
);
{error, nil} ->
{error, resp_error}
end;
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec scan_pattern_with_type(
gleam@erlang@process:subject(radish@client:message()),
integer(),
key_type(),
binary(),
integer(),
integer()
) -> {ok, {list(binary()), integer()}} | {error, radish@error:error()}.
scan_pattern_with_type(Client, Cursor, Key_type, Pattern, Count, Timeout) ->
_pipe = case Key_type of
set ->
radish@command:scan_pattern_with_type(
Cursor,
<<"set"/utf8>>,
Pattern,
Count
);
list ->
radish@command:scan_pattern_with_type(
Cursor,
<<"list"/utf8>>,
Pattern,
Count
);
z_set ->
radish@command:scan_pattern_with_type(
Cursor,
<<"zset"/utf8>>,
Pattern,
Count
);
hash ->
radish@command:scan_pattern_with_type(
Cursor,
<<"hash"/utf8>>,
Pattern,
Count
);
string ->
radish@command:scan_pattern_with_type(
Cursor,
<<"string"/utf8>>,
Pattern,
Count
);
stream ->
radish@command:scan_pattern_with_type(
Cursor,
<<"stream"/utf8>>,
Pattern,
Count
)
end,
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{array, [{bulk_string, New_cursor_str}, {array, Keys}]}] ->
case gleam@int:parse(New_cursor_str) of
{ok, New_cursor} ->
gleam@result:then(
gleam@list:try_map(
Keys,
fun(Item) -> case Item of
{bulk_string, Value@1} ->
{ok, Value@1};
_ ->
{error, resp_error}
end end
),
fun(Array) -> {ok, {Array, New_cursor}} end
);
{error, nil} ->
{error, resp_error}
end;
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec exists(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
exists(Client, Keys, Timeout) ->
_pipe = radish@command:exists(Keys),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec get(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
get(Client, Key, Timeout) ->
_pipe = radish@command:get(Key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{simple_string, Str}] ->
{ok, Str};
[{bulk_string, Str}] ->
{ok, Str};
[null] ->
{error, not_found};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec mget(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
integer()
) -> {ok, list(binary())} | {error, radish@error:error()}.
mget(Client, Keys, Timeout) ->
_pipe = radish@command:mget(Keys),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{array, Array}] ->
gleam@list:try_map(Array, fun(Item) -> case Item of
{bulk_string, Str} ->
{ok, Str};
null ->
{error, not_found};
_ ->
{error, resp_error}
end end);
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec append(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
append(Client, Key, Value, Timeout) ->
_pipe = radish@command:append(Key, Value),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec set(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
set(Client, Key, Value, Timeout) ->
_pipe = radish@command:set(Key, Value, []),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{simple_string, Str}] ->
{ok, Str};
[{bulk_string, Str}] ->
{ok, Str};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec set_new(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
set_new(Client, Key, Value, Timeout) ->
_pipe = radish@command:set(Key, Value, [nx]),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{simple_string, Str}] ->
{ok, Str};
[{bulk_string, Str}] ->
{ok, Str};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec set_existing(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
set_existing(Client, Key, Value, Timeout) ->
_pipe = radish@command:set(Key, Value, [xx, get]),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{simple_string, Str}] ->
{ok, Str};
[{bulk_string, Str}] ->
{ok, Str};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec mset(
gleam@erlang@process:subject(radish@client:message()),
list({binary(), binary()}),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
mset(Client, Kv_list, Timeout) ->
_pipe = radish@command:mset(Kv_list),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{simple_string, Str}] ->
{ok, Str};
[{bulk_string, Str}] ->
{ok, Str};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec del(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
del(Client, Keys, Timeout) ->
_pipe = radish@command:del(Keys),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec incr(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
incr(Client, Key, Timeout) ->
_pipe = radish@command:incr(Key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, New}] ->
{ok, New};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec incr_by(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
incr_by(Client, Key, Value, Timeout) ->
_pipe = radish@command:incr_by(Key, Value),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{integer, New}] ->
{ok, New};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec incr_by_float(
gleam@erlang@process:subject(radish@client:message()),
binary(),
float(),
integer()
) -> {ok, float()} | {error, radish@error:error()}.
incr_by_float(Client, Key, Value, Timeout) ->
_pipe = radish@command:incr_by_float(Key, Value),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@3 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{bulk_string, New}] ->
_pipe@2 = gleam@float:parse(New),
gleam@result:replace_error(_pipe@2, resp_error);
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@3).
-spec decr(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
decr(Client, Key, Timeout) ->
_pipe = radish@command:decr(Key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, New}] ->
{ok, New};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec decr_by(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
decr_by(Client, Key, Value, Timeout) ->
_pipe = radish@command:decr_by(Key, Value),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value@1) -> case Value@1 of
[{integer, New}] ->
{ok, New};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec random_key(
gleam@erlang@process:subject(radish@client:message()),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
random_key(Client, Timeout) ->
_pipe = radish@command:random_key(),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{bulk_string, Str}] ->
{ok, Str};
[null] ->
{error, not_found};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec key_type(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer()
) -> {ok, key_type()} | {error, radish@error:error()}.
key_type(Client, Key, Timeout) ->
_pipe = radish@command:key_type(Key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{simple_string, Str}] ->
case Str of
<<"set"/utf8>> ->
{ok, set};
<<"list"/utf8>> ->
{ok, list};
<<"zset"/utf8>> ->
{ok, z_set};
<<"hash"/utf8>> ->
{ok, hash};
<<"string"/utf8>> ->
{ok, string};
<<"stream"/utf8>> ->
{ok, stream};
_ ->
{error, resp_error}
end;
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec rename(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, binary()} | {error, radish@error:error()}.
rename(Client, Key, New_key, Timeout) ->
_pipe = radish@command:rename(Key, New_key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{simple_string, Str}] ->
{ok, Str};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec renamenx(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
renamenx(Client, Key, New_key, Timeout) ->
_pipe = radish@command:renamenx(Key, New_key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec persist(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
persist(Client, Key, Timeout) ->
_pipe = radish@command:persist(Key),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec expire(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
expire(Client, Key, Ttl, Timeout) ->
_pipe = radish@command:expire(Key, Ttl, none),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec expire_if(
gleam@erlang@process:subject(radish@client:message()),
binary(),
integer(),
expire_condition(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
expire_if(Client, Key, Ttl, Condition, Timeout) ->
_pipe = case Condition of
nx ->
radish@command:expire(Key, Ttl, {some, <<"NX"/utf8>>});
xx ->
radish@command:expire(Key, Ttl, {some, <<"XX"/utf8>>});
gt ->
radish@command:expire(Key, Ttl, {some, <<"GT"/utf8>>});
lt ->
radish@command:expire(Key, Ttl, {some, <<"LT"/utf8>>})
end,
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec publish(
gleam@erlang@process:subject(radish@client:message()),
binary(),
binary(),
integer()
) -> {ok, integer()} | {error, radish@error:error()}.
publish(Client, Channel, Message, Timeout) ->
_pipe = radish@command:publish(Channel, Message),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
_pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
[{integer, N}] ->
{ok, N};
_ ->
{error, resp_error}
end end),
gleam@result:flatten(_pipe@2).
-spec unsubscribe(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe(Client, Channels, Timeout) ->
_pipe = radish@command:unsubscribe(Channels),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
gleam@result:map(
_pipe@1,
fun(Value) ->
gleam@list:all(
Value,
fun(Item) ->
{push,
[{bulk_string, <<"unsubscribe"/utf8>>},
{bulk_string, _},
{integer, N}]} = case Item of
{push,
[{bulk_string, <<"unsubscribe"/utf8>>},
{bulk_string, _},
{integer, _}]} -> Item;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"radish"/utf8>>,
function => <<"unsubscribe"/utf8>>,
line => 673})
end,
N > 0
end
)
end
).
-spec unsubscribe_from_all(
gleam@erlang@process:subject(radish@client:message()),
integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe_from_all(Client, Timeout) ->
_pipe = radish@command:unsubscribe_from_all(),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
gleam@result:map(
_pipe@1,
fun(Value) ->
Value,
gleam@list:all(
Value,
fun(Item) ->
{push,
[{bulk_string, <<"unsubscribe"/utf8>>},
{bulk_string, _},
{integer, N}]} = case Item of
{push,
[{bulk_string, <<"unsubscribe"/utf8>>},
{bulk_string, _},
{integer, _}]} -> Item;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"radish"/utf8>>,
function => <<"unsubscribe_from_all"/utf8>>,
line => 689})
end,
N > 0
end
)
end
).
-spec subscribe(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
fun((binary(), integer()) -> nil),
fun((binary(), binary()) -> next()),
integer()
) -> nil.
subscribe(Client, Channels, Init_handler, Message_handler, Timeout) ->
_ = begin
_pipe = radish@command:subscribe(Channels),
_pipe@1 = radish@utils:execute_blocking(Client, _pipe, Timeout),
gleam@result:map(
_pipe@1,
fun(Value) -> gleam@list:each(Value, fun(Item) -> case Item of
{push,
[{bulk_string, <<"subscribe"/utf8>>},
{bulk_string, Channel},
{integer, N}]} ->
{ok, Init_handler(Channel, N)};
_ ->
{error, resp_error}
end end) end
)
end,
radish@utils:receive_forever(
Client,
Timeout,
fun(Value@1) -> case Value@1 of
{ok,
[{push,
[{bulk_string, <<"message"/utf8>>},
{bulk_string, Channel@1},
{bulk_string, Message}]}]} ->
case Message_handler(Channel@1, Message) of
continue ->
true;
unsubscribe_from_all ->
_ = unsubscribe_from_all(Client, Timeout),
false;
{unsubscribe_from, Channels@1} ->
case unsubscribe(Client, Channels@1, Timeout) of
{ok, Result} ->
Result;
{error, _} ->
false
end
end;
_ ->
false
end end
).
-spec unsubscribe_from_patterns(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe_from_patterns(Client, Patterns, Timeout) ->
_pipe = radish@command:unsubscribe_from_patterns(Patterns),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
gleam@result:map(
_pipe@1,
fun(Value) ->
gleam@list:all(
Value,
fun(Item) ->
{push,
[{bulk_string, <<"punsubscribe"/utf8>>},
{bulk_string, _},
{integer, N}]} = case Item of
{push,
[{bulk_string, <<"punsubscribe"/utf8>>},
{bulk_string, _},
{integer, _}]} -> Item;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"radish"/utf8>>,
function => <<"unsubscribe_from_patterns"/utf8>>,
line => 704})
end,
N > 0
end
)
end
).
-spec unsubscribe_from_all_patterns(
gleam@erlang@process:subject(radish@client:message()),
integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe_from_all_patterns(Client, Timeout) ->
_pipe = radish@command:unsubscribe_from_all_patterns(),
_pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
gleam@result:map(
_pipe@1,
fun(Value) ->
gleam@list:all(
Value,
fun(Item) ->
{push,
[{bulk_string, <<"punsubscribe"/utf8>>},
{bulk_string, _},
{integer, N}]} = case Item of
{push,
[{bulk_string, <<"punsubscribe"/utf8>>},
{bulk_string, _},
{integer, _}]} -> Item;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"radish"/utf8>>,
function => <<"unsubscribe_from_all_patterns"/utf8>>,
line => 719})
end,
N > 0
end
)
end
).
-spec subscribe_to_patterns(
gleam@erlang@process:subject(radish@client:message()),
list(binary()),
fun((binary(), integer()) -> nil),
fun((binary(), binary(), binary()) -> next()),
integer()
) -> nil.
subscribe_to_patterns(Client, Patterns, Init_handler, Message_handler, Timeout) ->
_ = begin
_pipe = radish@command:subscribe_to_patterns(Patterns),
_pipe@1 = radish@utils:execute_blocking(Client, _pipe, Timeout),
gleam@result:map(
_pipe@1,
fun(Value) -> gleam@list:each(Value, fun(Item) -> case Item of
{push,
[{bulk_string, <<"psubscribe"/utf8>>},
{bulk_string, Channel},
{integer, N}]} ->
Init_handler(Channel, N);
_ ->
nil
end end) end
)
end,
radish@utils:receive_forever(
Client,
Timeout,
fun(Value@1) -> case Value@1 of
{ok,
[{push,
[{bulk_string, <<"pmessage"/utf8>>},
{bulk_string, Pattern},
{bulk_string, Channel@1},
{bulk_string, Message}]}]} ->
case Message_handler(Pattern, Channel@1, Message) of
continue ->
true;
unsubscribe_from_all ->
_ = unsubscribe_from_all_patterns(Client, Timeout),
false;
{unsubscribe_from, Patterns@1} ->
case unsubscribe_from_patterns(
Client,
Patterns@1,
Timeout
) of
{ok, Result} ->
Result;
{error, _} ->
false
end
end;
_ ->
false
end end
).