//// All timeouts are in milliseconds import gleam/int import gleam/list import gleam/float import gleam/option import gleam/result import gleam/otp/actor import gleam/erlang/process import radish/resp import radish/error import radish/client import radish/utils.{execute, execute_blocking, receive_forever} import radish/command pub type StartOption { Timeout(Int) Auth(String) AuthWithUsername(String, String) } pub type KeyType { Set List ZSet Hash String Stream } pub type ExpireCondition { NX XX GT LT } pub fn start(host: String, port: Int, options: List(StartOption)) { let #(timeout, options) = case list.pop_map( options, fn(item) { case item { Timeout(timeout) -> Ok(timeout) _ -> Error(Nil) } }, ) { Ok(result) -> result Error(Nil) -> #(1024, options) } use client <- result.then(client.start(host, port, timeout)) let options = list.map( options, fn(item) { case item { Auth(password) -> command.Auth(password) AuthWithUsername(username, password) -> command.AuthWithUsername(username, password) Timeout(_) -> command.AuthWithUsername("", "") } }, ) use _ <- result.then( execute(client, command.hello(3, options), timeout) |> result.map_error(fn(error) { case error { error.ServerError(error) -> actor.InitFailed(process.Abnormal(error)) _ -> actor.InitFailed(process.Abnormal("Failed to say hello")) } }), ) Ok(client) } pub fn shutdown(client) { process.send(client, client.Shutdown) } /// see [here](https://redis.io/commands/keys)! pub fn keys(client, pattern: String, timeout: Int) { command.keys(pattern) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Array(array)] -> list.try_map( array, fn(item) { case item { resp.BulkString(value) -> Ok(value) _ -> Error(error.RESPError) } }, ) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/scan)! pub fn scan(client, cursor: Int, count: Int, timeout: Int) { command.scan(cursor, count) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Array([resp.BulkString(new_cursor_str), resp.Array(keys)])] -> case int.parse(new_cursor_str) { Ok(new_cursor) -> { use array <- result.then(list.try_map( keys, fn(item) { case item { resp.BulkString(value) -> Ok(value) _ -> Error(error.RESPError) } }, )) Ok(#(array, new_cursor)) } Error(Nil) -> Error(error.RESPError) } _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/scan)! pub fn scan_pattern( client, cursor: Int, pattern: String, count: Int, timeout: Int, ) { command.scan_pattern(cursor, pattern, count) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Array([resp.BulkString(new_cursor_str), resp.Array(keys)])] -> case int.parse(new_cursor_str) { Ok(new_cursor) -> { use array <- result.then(list.try_map( keys, fn(item) { case item { resp.BulkString(value) -> Ok(value) _ -> Error(error.RESPError) } }, )) Ok(#(array, new_cursor)) } Error(Nil) -> Error(error.RESPError) } _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/scan)! pub fn scan_with_type( client, cursor: Int, key_type: KeyType, count: Int, timeout: Int, ) { case key_type { Set -> command.scan_with_type(cursor, "set", count) List -> command.scan_with_type(cursor, "list", count) ZSet -> command.scan_with_type(cursor, "zset", count) Hash -> command.scan_with_type(cursor, "hash", count) String -> command.scan_with_type(cursor, "string", count) Stream -> command.scan_with_type(cursor, "stream", count) } |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Array([resp.BulkString(new_cursor_str), resp.Array(keys)])] -> case int.parse(new_cursor_str) { Ok(new_cursor) -> { use array <- result.then(list.try_map( keys, fn(item) { case item { resp.BulkString(value) -> Ok(value) _ -> Error(error.RESPError) } }, )) Ok(#(array, new_cursor)) } Error(Nil) -> Error(error.RESPError) } _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/scan)! pub fn scan_pattern_with_type( client, cursor: Int, key_type: KeyType, pattern: String, count: Int, timeout: Int, ) { case key_type { Set -> command.scan_pattern_with_type(cursor, "set", pattern, count) List -> command.scan_pattern_with_type(cursor, "list", pattern, count) ZSet -> command.scan_pattern_with_type(cursor, "zset", pattern, count) Hash -> command.scan_pattern_with_type(cursor, "hash", pattern, count) String -> command.scan_pattern_with_type(cursor, "string", pattern, count) Stream -> command.scan_pattern_with_type(cursor, "stream", pattern, count) } |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Array([resp.BulkString(new_cursor_str), resp.Array(keys)])] -> case int.parse(new_cursor_str) { Ok(new_cursor) -> { use array <- result.then(list.try_map( keys, fn(item) { case item { resp.BulkString(value) -> Ok(value) _ -> Error(error.RESPError) } }, )) Ok(#(array, new_cursor)) } Error(Nil) -> Error(error.RESPError) } _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/exists)! pub fn exists(client, keys: List(String), timeout: Int) { command.exists(keys) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/get)! pub fn get(client, key: String, timeout: Int) { command.get(key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] | [resp.BulkString(str)] -> Ok(str) [resp.Null] -> Error(error.NotFound) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/mget)! pub fn mget(client, keys: List(String), timeout: Int) { command.mget(keys) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Array(array)] -> list.try_map( array, fn(item) { case item { resp.BulkString(str) -> Ok(str) resp.Null -> Error(error.NotFound) _ -> Error(error.RESPError) } }, ) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/append)! pub fn append(client, key: String, value: String, timeout: Int) { command.append(key, value) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/set)! pub fn set(client, key: String, value: String, timeout: Int) { command.set(key, value, []) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] | [resp.BulkString(str)] -> Ok(str) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/set)! pub fn set_new(client, key: String, value: String, timeout: Int) { command.set(key, value, [command.NX]) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] | [resp.BulkString(str)] -> Ok(str) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/set)! pub fn set_existing(client, key: String, value: String, timeout: Int) { command.set(key, value, [command.XX, command.GET]) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] | [resp.BulkString(str)] -> Ok(str) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/mset)! pub fn mset(client, kv_list: List(#(String, String)), timeout: Int) { command.mset(kv_list) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] | [resp.BulkString(str)] -> Ok(str) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/del)! pub fn del(client, keys: List(String), timeout: Int) { command.del(keys) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/incr)! pub fn incr(client, key: String, timeout: Int) { command.incr(key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(new)] -> Ok(new) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/incrby)! pub fn incr_by(client, key: String, value: Int, timeout: Int) { command.incr_by(key, value) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(new)] -> Ok(new) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/incrbyfloat)! pub fn incr_by_float(client, key: String, value: Float, timeout: Int) { command.incr_by_float(key, value) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.BulkString(new)] -> float.parse(new) |> result.replace_error(error.RESPError) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/decr)! pub fn decr(client, key: String, timeout: Int) { command.decr(key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(new)] -> Ok(new) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/decrby)! pub fn decr_by(client, key: String, value: Int, timeout: Int) { command.decr_by(key, value) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(new)] -> Ok(new) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/randomkey)! pub fn random_key(client, timeout: Int) { command.random_key() |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.BulkString(str)] -> Ok(str) [resp.Null] -> Error(error.NotFound) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/type)! pub fn key_type(client, key: String, timeout: Int) { command.key_type(key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] -> case str { "set" -> Set "list" -> List "zset" -> ZSet "hash" -> Hash "string" -> String "stream" -> Stream } |> Ok _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/rename)! pub fn rename(client, key: String, new_key: String, timeout: Int) { command.rename(key, new_key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.SimpleString(str)] -> Ok(str) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/renamenx)! pub fn renamenx(client, key: String, new_key: String, timeout: Int) { command.renamenx(key, new_key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/persist)! pub fn persist(client, key: String, timeout: Int) { command.persist(key) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/expire)! pub fn expire(client, key: String, ttl: Int, timeout: Int) { command.expire(key, ttl, option.None) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/expire)! pub fn expire_if( client, key: String, ttl: Int, condition: ExpireCondition, timeout: Int, ) { case condition { NX -> command.expire(key, ttl, option.Some("NX")) XX -> command.expire(key, ttl, option.Some("XX")) GT -> command.expire(key, ttl, option.Some("GT")) LT -> command.expire(key, ttl, option.Some("LT")) } |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } pub type Next { Continue UnsubscribeFromAll UnsubscribeFrom(List(String)) } /// see [here](https://redis.io/commands/publish)! pub fn publish(client, channel: String, message: String, timeout: Int) { command.publish(channel, message) |> execute(client, _, timeout) |> result.map(fn(value) { case value { [resp.Integer(n)] -> Ok(n) _ -> Error(error.RESPError) } }) |> result.flatten } /// see [here](https://redis.io/commands/subscribe)! /// Also see [here](https://redis.io/docs/manual/keyspace-notifications)! pub fn subscribe( client, channels: List(String), init_handler: fn(String, Int) -> Nil, message_handler: fn(String, String) -> Next, timeout: Int, ) { let _ = command.subscribe(channels) |> execute_blocking(client, _, timeout) |> result.map(fn(value) { list.each( value, fn(item) { case item { resp.Push([ resp.BulkString("subscribe"), resp.BulkString(channel), resp.Integer(n), ]) -> Ok(init_handler(channel, n)) _ -> Error(error.RESPError) } }, ) }) use value <- receive_forever(client, timeout) case value { Ok([ resp.Push([ resp.BulkString("message"), resp.BulkString(channel), resp.BulkString(message), ]), ]) -> case message_handler(channel, message) { Continue -> True UnsubscribeFromAll -> { let _ = unsubscribe_from_all(client, timeout) False } UnsubscribeFrom(channels) -> case unsubscribe(client, channels, timeout) { Ok(result) -> result Error(_) -> False } } _ -> False } } /// see [here](https://redis.io/commands/psubscribe)! /// Also see [here](https://redis.io/docs/manual/keyspace-notifications)! pub fn subscribe_to_patterns( client, patterns: List(String), init_handler: fn(String, Int) -> Nil, message_handler: fn(String, String, String) -> Next, timeout: Int, ) { let _ = command.subscribe_to_patterns(patterns) |> execute_blocking(client, _, timeout) |> result.map(fn(value) { list.each( value, fn(item) { case item { resp.Push([ resp.BulkString("psubscribe"), resp.BulkString(channel), resp.Integer(n), ]) -> init_handler(channel, n) _ -> Nil } }, ) }) use value <- receive_forever(client, timeout) case value { Ok([ resp.Push([ resp.BulkString("pmessage"), resp.BulkString(pattern), resp.BulkString(channel), resp.BulkString(message), ]), ]) -> case message_handler(pattern, channel, message) { Continue -> True UnsubscribeFromAll -> { let _ = unsubscribe_from_all_patterns(client, timeout) False } UnsubscribeFrom(patterns) -> { case unsubscribe_from_patterns(client, patterns, timeout) { Ok(result) -> result Error(_) -> False } } } _ -> False } } fn unsubscribe(client, channels: List(String), timeout: Int) { command.unsubscribe(channels) |> execute(client, _, timeout) |> result.map(fn(value) { list.all( value, fn(item) { let assert resp.Push([ resp.BulkString("unsubscribe"), resp.BulkString(_), resp.Integer(n), ]) = item n > 0 }, ) }) } fn unsubscribe_from_all(client, timeout: Int) { command.unsubscribe_from_all() |> execute(client, _, timeout) |> result.map(fn(value) { value list.all( value, fn(item) { let assert resp.Push([ resp.BulkString("unsubscribe"), resp.BulkString(_), resp.Integer(n), ]) = item n > 0 }, ) }) } fn unsubscribe_from_patterns(client, patterns: List(String), timeout: Int) { command.unsubscribe_from_patterns(patterns) |> execute(client, _, timeout) |> result.map(fn(value) { list.all( value, fn(item) { let assert resp.Push([ resp.BulkString("punsubscribe"), resp.BulkString(_), resp.Integer(n), ]) = item n > 0 }, ) }) } fn unsubscribe_from_all_patterns(client, timeout: Int) { command.unsubscribe_from_all_patterns() |> execute(client, _, timeout) |> result.map(fn(value) { list.all( value, fn(item) { let assert resp.Push([ resp.BulkString("punsubscribe"), resp.BulkString(_), resp.Integer(n), ]) = item n > 0 }, ) }) }