redis-asio

redis-asio is a Redis client library written in pure Rust based on asynchronous tokio library.

The library provides a base module for low-level request sending and response handling, and a stream module that contains specific interfaces for work with Redis-Stream "https://redis.io/topics/streams-intro".

The library works with binary-safe strings that allows users to serialize their message structures and send via RESP protocol "https://redis.io/topics/protocol".

Use in project

Depend on the redis-asio in project via Cargo:

toml [dependencies] redis-async = "0.1"

Resolve the crate interfaces:

rust extern crate redis_asio;

Motivating examples

SET, GET value from cache

```rust use std::net::SocketAddr; use futures::Future; use redisasio::{RedisCoreConnection, RedisValue, fromredis_value};

let address = &"127.0.0.1:6379".parse::().unwrap();

let setreq = command("SET").arg("foo").arg(123); let getreq = command("GET").arg("foo");

let future = RedisCoreConnection::connect(address) .andthen(move |con| { // send "SET foo 123" request con.send(setreq) }) .andthen(|(con, response)| { // check if the value has been set asserteq!(RedisValue::Ok, response); // send "GET foo" request con.send(getreq) }) .map(move |(, response)| // check if the received value is the same asserteq!(123, fromredisvalue(&response).unwrap())) .maperr(|_| unreachable!()); // start the Tokio runtime using the future tokio::run(future); ```

Subscribe to a Redis Stream

Subscribe to a Redis stream and process all incoming entries. Redis Streams requires to send XREAD/XREADGROUP requests every time the client receives a response on the previous, in other words Redis Streams does not provide an interface to subscribe to a Redis stream.

In the Crate the subscription is possible by hidden requests sending within the Crate engine.

Request that will be sent to get new entries in the following example: "XREADGROUP GROUP mygroup Bob BLOCK 0 STREAMS mystream <"

```rust use std::net::SocketAddr; use futures::{Future, Stream}; use redis_asio::stream::{RedisStream, SubscribeOptions, StreamEntry, RedisGroup};

let address = &"127.0.0.1:6379".parse::().unwrap(); // create options to pass it into RedisStream::subscribe() let groupinfo = RedisGroup::new("mygroup".tostring(), "Bob".tostring()); let subscribeoptions = SubscribeOptions::withgroup(vec!["mystream".tostring()], group_info);

let future = RedisStream::connect(address) .andthen(move |stream: RedisStream| { stream.subscribe(subscribeoptions) }) .andthen(|subscribe| /*:Subscribe*/ { // pass the closure that will be called on each incoming entries subscribe.foreach(|entries: Vec| { for entry in entries.intoiter() { println!("Received: {:?}", entry); } Ok(()) }) }) .maperr(|err| eprintln!("something went wrong: {}", err)); // start the Tokio runtime using the future tokio::run(future); ```

Send an entry into a Redis Stream

Send an entry into a Redis stream. Request that will be sent to push a specified entry in the following example: "XADD mystream * type 3 data \"Hello, world\""

```rust use std::net::SocketAddr; use std::collections::HashMap; use futures::Future; use redisasio::{RedisArgument, IntoRedisArgument}; use redisasio::stream::{RedisStream, SendEntryOptions, EntryId};

let address = &"127.0.0.1:6379".parse::().unwrap(); // create options to pass it into RedisStream::sendentry() let sendoptions = SendEntryOptions::new("mystream".to_string());

// create key-value pairs let mut request: HashMap = HashMap::new(); request.insert("type".tostring(), 3i32.intoredisargument()); request.insert("data".tostring(), "Hello, world!".intoredisargument());

let future = RedisStream::connect(address) .andthen(move |stream: RedisStream| { // HashMap satisfies the // HashMap stream.sendentry(sendoptions, request) }) .map(|(, insertedentryid): (RedisStream, EntryId)| { println!("{:?} has sent", insertedentryid.tostring()); }) .maperr(|err| eprintln!("something went wrong: {}", err)); tokio::run(future); ```