/ 
Net-Async-Redis-6.006
River stage one • 2 direct dependents • 2 total dependents
/ Net::Async::Redis

NAME

Net::Async::Redis - talk to Redis servers via IO::Async

SYNOPSIS

use Net::Async::Redis;
use Future::AsyncAwait;
use IO::Async::Loop;
my $loop = IO::Async::Loop->new;
$loop->add(my $redis = Net::Async::Redis->new);
my $value = await $redis->get('some_key');
$value ||= await $redis->set(some_key => 'some_value');
print "Value: $value\n";

# You can also use ->then chaining, see L<Future> for more details
$redis->connect->then(sub {
    $redis->get('some_key')
})->then(sub {
    my $value = shift;
    return Future->done($value) if $value;
    $redis->set(some_key => 'some_value')
})->on_done(sub {
    print "Value: " . shift;
})->get;

DESCRIPTION

Provides client access for dealing with Redis servers.

6ee Net::Async::Redis::Commands for the full list of commands, this list is autogenerated from the official documentation here:

https://redis.io/commands

This is intended to be a near-complete low-level client module for asynchronous Redis support. See Net::Async::Redis::Server for a (limited) Perl server implementation.

This is an unofficial Perl port, and not endorsed by the Redis server maintainers in any way.

Supported features

Current features include:

Connecting

As with any other IO::Async::Notifier-based module, you'll need to add this to an IO::Async::Loop:

my $loop = IO::Async::Loop->new;
$loop->add(
    my $redis = Net::Async::Redis->new
);

then connect to the server:

use Future::AsyncAwait;
await $redis->connect;
# You could achieve a similar result by passing client_name in
# constructor or ->connect parameters
await $redis->client_setname("example client");

Key-value handling

One of the most common Redis scenarios is as a key/value store. The "get" and "set" methods are typically used here:

use Future::AsyncAwait;
await $redis->connect;
$redis->set(some_key => 'some value');
my ($value) = await $redis->get('some_key');
print "Read back value [$value]\n";

See the next section for more information on what these methods are actually returning.

Requests and responses

Requests are implemented as methods on the Net::Async::Redis object. These typically return a Future which will resolve once ready:

my $future = $redis->incr("xyz")
    ->on_done(sub {
        print "result of increment was " . shift . "\n"
    });

For synchronous code, call ->get on that Future:

print "Database has " . $redis->dbsize->get . " total keys\n";

This means you can end up with ->get being called on the result of ->get, note that these are two very different methods:

$redis
 ->get('some key') # this is being called on $redis, and is issuing a GET request
 ->get # this is called on the returned Future, and blocks until the value is ready

Typical async code would not be expected to use the "get" in Future method extensively; often only calling it in one place at the top level in the code.

RESP3 and RESP2 compatibility

In RESP3 some of the responses are structured differently from RESP2. Net::Async::Redis guarantees the same structure unless you have explicitly requested the new types using the "configure" hashrefs option, which is disabled by default.

Generally RESP3 is recommended if you have Redis version 6 or later installed: it allows subscription operations to share the same connection as regular Redis traffic.

Error handling

Since Future is used for deferred results, failure is indicated by a failing Future with failure category of redis.

The "catch" in Future feature may be useful for handling these:

$redis->lpush(key => $value)
    ->catch(
        redis => sub { warn "probably an incorrect type, cannot push value"; Future->done }
    )->get;

Note that this module uses Future::AsyncAwait internally.

CONSTANTS

OPENTRACING_ENABLED

Defaults to false, this can be controlled by the USE_OPENTRACING environment variable. This provides a way to set the default opentracing mode for all Net::Async::Redis instances - you can enable/disable for a specific instance via "configure":

$redis->configure(opentracing => 1);

When enabled, this will create a span for every Redis request. See OpenTracing::Any for details.

OPENTELEMETRY_ENABLED

Defaults to false, this can be controlled by the USE_OPENTELEMETRY environment variable. This provides a way to set the default opentelemetry mode for all Net::Async::Redis instances - you can enable/disable for a specific instance via "configure":

$redis->configure(opentelemetry => 1);

When enabled, this will create a span for every Redis request. See OpenTelemetry or https://opentelemetry.io for details.

METHODS

NOTE: For a full list of the Redis methods supported by this module, please see Net::Async::Redis::Commands.

configure

Applies configuration parameters - currently supports:

  • host

  • port

  • auth

  • database

  • pipeline_depth

  • stream_read_len

  • stream_write_len

  • on_disconnect

  • client_name

  • opentracing

  • opentelemetry

  • protocol - either 'resp2' or 'resp3', default is autodetect

  • hashrefs - RESP3 (Redis 6.0+) supports more data types, currently the only difference this makes to us is that it now supports hashrefs for key/value pairs. This is disabled by default to ensure compatibility across newer+older versions.

Note that enabling hashrefs will cause connections to fail if the server does not support RESP3.

host

Returns the host or IP address for the Redis server.

port

Returns the port used for connecting to the Redis server.

database

Returns the database index used when connecting to the Redis server.

See the "select" in Net::Async::Redis::Commands method for details.

uri

Returns the Redis endpoint URI instance.

stream_read_len

Returns the buffer size when reading from a Redis connection.

Defaults to 1MB, reduce this if you're dealing with a lot of connections and want to minimise memory usage. Alternatively, if you're reading large amounts of data and spend too much time in needless epoll_wait calls, try a larger value.

stream_write_len

Returns the buffer size when writing to Redis connections, in bytes. Defaults to 1MB.

See "stream_read_len".

client_name

Returns the name used for this client when connecting.

METHODS - Connection

connect

Connects to the Redis server.

Will use the "configure"d parameters if available, but as a convenience can be passed additional parameters which will then be applied as if you had called "configure" with those beforehand. This also means that they will be preserved for subsequent "connect" calls.

connected

Establishes a connection if needed, otherwise returns an immediately-available Future instance.

endpoint

The string describing the remote endpoint.

local_endpoint

A string describing the local endpoint, usually host:port.

METHODS - Subscriptions

See https://redis.io/topics/pubsub for more details on this topic. There's also more details on the internal implementation in Redis here: https://making.pusher.com/redis-pubsub-under-the-hood/.

NOTE: On Redis versions prior to 6.0, you will need a separate connection for subscriptions; you cannot share a connection for regular requests once any of the "subscribe" or "psubscribe" methods have been called on an existing connection.

With Redis 6.0, a newer protocol version (RESP3) is used by default, and this is quite happy to support pubsub activity on the same connection as other traffic.

psubscribe

Subscribes to a pattern.

Example:

# Subscribe to 'info::*' channels, i.e. any message
# that starts with the C<info::> prefix, and prints them
# with a timestamp.
$redis_connection->psubscribe('info::*')
   ->then(sub {
       my $sub = shift;
       $sub->map('payload')
           ->each(sub {
            print localtime . ' ' . $_ . "\n";
           })->retain
   })->get;
# this will block until the subscribe is confirmed. Note that you can't publish on
# a connection that's handling subscriptions due to Redis protocol restrictions.
$other_redis_connection->publish('info::example', 'a message here')->get;

Returns a Future which resolves to a Net::Async::Redis::Subscription instance.

subscribe

Subscribes to one or more channels.

Returns a Future which resolves to a Net::Async::Redis::Subscription instance.

Example:

# Subscribe to 'notifications' channel,
# print the first 5 messages, then unsubscribe
$redis->subscribe('notifications')
   ->then(sub {
       my $sub = shift;
       $sub->events
           ->map('payload')
           ->take(5)
           ->say
           ->completed
   })->then(sub {
       $redis->unsubscribe('notifications')
   })->get

ssubscribe

Subscribes to one or more sharded channels.

This behaves similarly to "subscribe", but applies to messages received on a specific shard. This is mostly relevant in a cluster context, where subscriptions can be localised to one shard (group of nodes) in the cluster to improve performance.

More details are in the sharded pubsub documentation.

Returns a Future which resolves to a Net::Async::Redis::Subscription instance.

Example:

# Subscribe to 'notifications' channel,
# print the first 5 messages, then unsubscribe
$redis->subscribe('notifications')
   ->then(sub {
       my $sub = shift;
       $sub->events
           ->map('payload')
           ->take(5)
           ->say
           ->completed
   })->then(sub {
       $redis->unsubscribe('notifications')
   })->get

METHODS - Transactions

multi

Executes the given code in a Redis MULTI transaction.

This will cause each of the requests to be queued on the server, then applied in a single atomic transaction.

Note that the commands will resolve only after the transaction is committed: for example, when the "set" command is issued, Redis will return QUEUED. This information is not used as the result - we only pass through the immediate response if there was an error. The Future representing the response will be marked as done once the EXEC command is applied and we have the results back.

Example:

$redis->multi(sub {
 my $tx = shift;
 $tx->incr('some::key')->on_done(sub { print "Final value for incremented key was " . shift . "\n"; });
 $tx->set('other::key => 'test data')
})->then(sub {
 my ($success, $failure) = @_;
 return Future->fail("Had $failure failures, expecting everything to succeed") if $failure;
 print "$success succeeded\m";
 return Future->done;
})->retain;

METHODS - Clientside caching

Enable clientside caching by passing a true value for client_side_caching_enabled in "configure" or "new". This is currently experimental, and only operates on "get" in Net::Async::Redis::Commands requests.

See https://redis.io/topics/client-side-caching for more details on this feature.

clientside_cache_events

A Ryu::Source which emits key names as they are invalidated.

With client-side caching enabled, can be used to monitor which keys are changing.

client_side_cache_ready

Returns a Future representing the client-side cache connection status, if there is one.

client_side_cache

Returns the Cache::LRU instance used for the client-side cache.

is_client_side_cache_enabled

Returns true if the client-side cache is enabled.

client_side_cache_size

Returns the current client-side cache size, as a number of entries.

METHODS - Generic

keys

watch_keyspace

A convenience wrapper around the keyspace notifications API.

Provides the necessary setup to establish a PSUBSCRIBE subscription on the __keyspace@*__ namespace, setting the configuration required for this to start emitting events, and then calls $code with each event.

Note that this will switch the connection into pubsub mode on versions of Redis older than 6.0, so it will no longer be available for any other activity. This limitation does not apply on Redis 6 or above.

Use * to listen for all keyspace changes.

Resolves to a Ryu::Source instance.

pipeline_depth

Number of requests awaiting responses before we start queuing. This defaults to an arbitrary value of 100 requests.

Note that this does not apply when in transaction (MULTI) mode.

See https://redis.io/topics/pipelining for more details on this concept.

opentracing

Indicates whether OpenTracing::Any support is enabled.

opentelemetry

Indicates whether OpenTelemetry support is enabled.

METHODS - Deprecated

This are still supported, but no longer recommended.

METHODS - Internal

on_message

Called for each incoming message.

Passes off the work to "handle_pubsub_message" or the next queue item, depending on whether we're dealing with subscriptions at the moment.

next_in_pipeline

Attempt to process next pending request when in pipeline mode.

on_error_message

Called when there's an error response.

handle_pubsub_message

Deal with an incoming pubsub-related message.

stream

Represents the IO::Async::Stream instance for the active Redis connection.

notify_close

Called when the socket is closed.

command_label

Generate a label for the given command list.

span_for_future

See https://opentelemetry.io/docs/specs/semconv/database/redis/ for current semantic conventions around Redis.

execute_command

Queues the given command for execution.

ryu

A Ryu::Async instance for source/sink creation.

future

Factory method for creating new Future instances.

wire_protocol

Returns the Net::Async::Redis::Protocol instance used for encoding and decoding messages.

enable_clientside_cache

Used internally to prepare for client-side caching: subscribes to the invalidation events.

_init

_add_to_loop

retrieve_full_command_list

Iterates through all commands defined in Redis, extracting the information about that command using COMMAND INFO.

The data is formatted for internal use, converting information such as flags into hashrefs for easier lookup.

This information is also used by "extract_keys_from_command".

Returns a hashref, where each key represents a method name (space-separated commands such as CLUSTER NODES are returned as cluster_nodes). The values are a restructured form of https://redis.io/commands/command.

extract_keys_for_command

Given a command arrayref and a definition for the server, this will return a list of any keys found in that command.

Since the logic for this is slightly slow, we are caching the result unless a specific definition is provided: this is an internal implementation detail and not something to rely on.

(the key specification is a relatively new Redis feature - an optimised version of this logic will be added to Net::Async::Redis::XS in due course, which should reduce the need for caching)

Returns a list of keys.

ssl_options

Extracts the SSL-related options as a hashref for passing to $loop->connect.

SEE ALSO

Some other Redis implementations on CPAN:

  • Mojo::Redis2 - nonblocking, using the Mojolicious framework, actively maintained

  • MojoX::Redis - changelog mentions that this was obsoleted by Mojo::Redis, although there have been new versions released since then

  • RedisDB - another synchronous (blocking) implementation, handles pub/sub and autoreconnect

  • Cache::Redis - wrapper around RedisDB

  • Redis::Fast - wraps hiredis, faster than Redis

  • Redis::Jet - also XS-based, docs mention very early development stage but appears to support pipelining and can handle newer commands via ->command.

  • Redis - synchronous (blocking) implementation, handles pub/sub and autoreconnect

  • HiRedis::Raw - another hiredis wrapper

AUTHOR

Tom Molesworth <TEAM@cpan.org>

INHERITED METHODS

Net::Async::Redis::Commands

FT_ADD, FT_AGGREGATE, FT_ALIASADD, FT_ALIASDEL, FT_ALIASUPDATE, FT_ALTER, FT_CONFIG, FT_CREATE, FT_CURSOR, FT_DEL, FT_DICTADD, FT_DICTDEL, FT_DICTDUMP, FT_DROP, FT_DROPINDEX, FT_EXPLAIN, FT_EXPLAINCLI, FT_GET, FT_INFO, FT_MGET, FT_PROFILE, FT_SEARCH, FT_SPELLCHECK, FT_SUGADD, FT_SUGDEL, FT_SUGGET, FT_SUGLEN, FT_SYNADD, FT_SYNDUMP, FT_SYNFORCEUPDATE, FT_SYNUPDATE, FT_TAGVALS, FT__ALIASADDIFNX, FT__ALIASDELIFX, FT__ALTERIFNX, FT__CREATEIFNX, FT__DROPIFX, FT__DROPINDEXIFX, FT__LIST, acl, acl_cat, acl_deluser, acl_dryrun, acl_genpass, acl_getuser, acl_help, acl_list, acl_load, acl_log, acl_save, acl_setuser, acl_users, acl_whoami, append, asking, auth, bf_add, bf_card, bf_debug, bf_exists, bf_info, bf_insert, bf_loadchunk, bf_madd, bf_mexists, bf_reserve, bf_scandump, bgrewriteaof, bgsave, bitcount, bitfield, bitfield_ro, bitop, bitpos, blmove, blmpop, blpop, brpop, brpoplpush, bzmpop, bzpopmax, bzpopmin, cf_add, cf_addnx, cf_compact, cf_count, cf_debug, cf_del, cf_exists, cf_info, cf_insert, cf_insertnx, cf_loadchunk, cf_mexists, cf_reserve, cf_scandump, client, client_caching, client_getname, client_getredir, client_help, client_id, client_info, client_kill, client_list, client_no, client_no_evict, client_no_touch, client_pause, client_reply, client_setinfo, client_setname, client_tracking, client_trackinginfo, client_unblock, client_unpause, cluster, cluster_addslots, cluster_addslotsrange, cluster_bumpepoch, cluster_count, cluster_count_failure, cluster_count_failure_reports, cluster_countkeysinslot, cluster_delslots, cluster_delslotsrange, cluster_failover, cluster_flushslots, cluster_forget, cluster_getkeysinslot, cluster_help, cluster_info, cluster_keyslot, cluster_links, cluster_meet, cluster_myid, cluster_myshardid, cluster_nodes, cluster_replicas, cluster_replicate, cluster_reset, cluster_saveconfig, cluster_set, cluster_set_config, cluster_set_config_epoch, cluster_setslot, cluster_shards, cluster_slaves, cluster_slots, cms_incrby, cms_info, cms_initbydim, cms_initbyprob, cms_merge, cms_query, command, command_count, command_docs, command_getkeys, command_getkeysandflags, command_help, command_info, command_list, config, config_get, config_help, config_resetstat, config_rewrite, config_set, copy, dbsize, debug, decr, decrby, del, discard, dump, echo, eval, eval_ro, evalsha, evalsha_ro, exec, exists, expire, expireat, expiretime, failover, fcall, fcall_ro, flushall, flushdb, function, function_delete, function_dump, function_flush, function_help, function_kill, function_list, function_load, function_restore, function_stats, geoadd, geodist, geohash, geopos, georadius, georadius_ro, georadiusbymember, georadiusbymember_ro, geosearch, geosearchstore, getbit, getdel, getex, getrange, getset, hdel, hello, hexists, hexpire, hexpireat, hexpiretime, hget, hgetall, hincrby, hincrbyfloat, hkeys, hlen, hmget, hmset, hpersist, hpexpire, hpexpireat, hpexpiretime, hpttl, hrandfield, hscan, hset, hsetnx, hstrlen, httl, hvals, incr, incrby, incrbyfloat, info, json_arrappend, json_arrindex, json_arrinsert, json_arrlen, json_arrpop, json_arrtrim, json_clear, json_debug, json_del, json_forget, json_get, json_merge, json_mget, json_mset, json_numincrby, json_nummultby, json_numpowby, json_objkeys, json_objlen, json_resp, json_set, json_strappend, json_strlen, json_toggle, json_type, lastsave, latency, latency_doctor, latency_graph, latency_help, latency_histogram, latency_history, latency_latest, latency_reset, lcs, lindex, linsert, llen, lmove, lmpop, lolwut, lpop, lpos, lpush, lpushx, lrange, lrem, lset, ltrim, memory, memory_doctor, memory_help, memory_malloc, memory_malloc_stats, memory_purge, memory_stats, memory_usage, mget, migrate, module, module_help, module_list, module_load, module_loadex, module_unload, monitor, move, mset, msetnx, object, object_encoding, object_freq, object_help, object_idletime, object_refcount, persist, pexpire, pexpireat, pexpiretime, pfadd, pfcount, pfdebug, pfmerge, pfselftest, ping, psetex, psync, pttl, publish, pubsub, pubsub_channels, pubsub_help, pubsub_numpat, pubsub_numsub, pubsub_shardchannels, pubsub_shardnumsub, punsubscribe, quit, randomkey, readonly, readwrite, rename, renamenx, replconf, replicaof, reset, restore, restore_asking, role, rpop, rpoplpush, rpush, rpushx, sadd, save, scan, scard, script, script_debug, script_exists, script_flush, script_help, script_kill, script_load, sdiff, sdiffstore, search_CLUSTERINFO, search_CLUSTERREFRESH, search_CLUSTERSET, select, set, setbit, setex, setnx, setrange, shutdown, sinter, sintercard, sinterstore, sismember, slaveof, slowlog, slowlog_get, slowlog_help, slowlog_len, slowlog_reset, smembers, smismember, smove, sort, sort_ro, spop, spublish, srandmember, srem, sscan, strlen, substr, sunion, sunionstore, sunsubscribe, swapdb, sync, tdigest_add, tdigest_byrank, tdigest_byrevrank, tdigest_cdf, tdigest_create, tdigest_info, tdigest_max, tdigest_merge, tdigest_min, tdigest_quantile, tdigest_rank, tdigest_reset, tdigest_revrank, tdigest_trimmed_mean, time, timeseries_CLUSTERSET, timeseries_CLUSTERSETFROMSHARD, timeseries_FORCESHARDSCONNECTION, timeseries_HELLO, timeseries_INFOCLUSTER, timeseries_INNERCOMMUNICATION, timeseries_NETWORKTEST, timeseries_REFRESHCLUSTER, topk_add, topk_count, topk_incrby, topk_info, topk_list, topk_query, topk_reserve, touch, ts_add, ts_alter, ts_create, ts_createrule, ts_decrby, ts_del, ts_deleterule, ts_get, ts_incrby, ts_info, ts_madd, ts_mget, ts_mrange, ts_mrevrange, ts_queryindex, ts_range, ts_revrange, ttl, type, unlink, unsubscribe, unwatch, wait, waitaof, watch, xack, xadd, xautoclaim, xclaim, xdel, xgroup, xgroup_create, xgroup_createconsumer, xgroup_delconsumer, xgroup_destroy, xgroup_help, xgroup_setid, xinfo, xinfo_consumers, xinfo_groups, xinfo_help, xinfo_stream, xlen, xpending, xrange, xrevrange, xsetid, xtrim, zadd, zcard, zcount, zdiff, zdiffstore, zincrby, zinter, zintercard, zinterstore, zlexcount, zmpop, zmscore, zpopmax, zpopmin, zrandmember, zrange, zrangebylex, zrangebyscore, zrangestore, zrank, zrem, zremrangebylex, zremrangebyrank, zremrangebyscore, zrevrange, zrevrangebylex, zrevrangebyscore, zrevrank, zscan, zscore, zunion, zunionstore

IO::Async::Notifier

add_child, adopt_future, adopted_futures, can_event, children, configure_unknown, debug_printf, get_loop, invoke_error, invoke_event, loop, make_event_cb, maybe_invoke_event, maybe_make_event_cb, notifier_name, parent, remove_child, remove_from_parent

Object::Pad::UNIVERSAL

BUILDARGS

CONTRIBUTORS

With thanks to the following for contributing patches, bug reports, tests and feedback:

  • BINARY@cpan.org

  • PEVANS@cpan.org

  • @eyadof

  • Nael Alolwani

  • Marc Frank

  • @pnevins

LICENSE

Copyright Tom Molesworth and others 2015-2024. Licensed under the same terms as Perl itself.