Redis Streams in Action - Part 2 (Rust app to consume from the Twitter Streaming API)
Tags: redis, rust, database, programming
Published: 2021-05-31 13:18:28

Welcome to this series of blog posts which covers [Redis Streams](https://redis.io/topics/streams-intro) with the help of a practical example. We will use a sample application to make Twitter data available for search and query in real-time. [RediSearch](https://redisearch.io/) and [Redis Streams](https://redis.io/topics/streams-intro) serve as the backbone of this solution that consists of several co-operating components, each of which will we covered in a dedicated blog post. 

> The code is available in this GitHub repo - https://github.com/abhirockzz/redis-streams-in-action

In this part, we look at the service which interacts with the Twitter Streaming API to consume tweets and move them on to the next part in the processing pipeline.

![](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/p77ceg4yz8z0ybag3jj3.jpg)

Our end goal is to be able to process tweets and make them available for search and queries via `RediSearch`. One could write a "do it all" service to consume tweets and directly store them in `RediSearch`. But, in order to scale to handle the volume of tweets, we need a service to act as a buffer and decouple our producer (the application we will focus in this blog) and consumer (covered in next blog). 

This is exactly what our first component facilitates - it consumes streaming Twitter data and forwards it to Redis Streams. We will deploy it to [Azure Container Instances](https://docs.microsoft.com/azure/container-instances/container-instances-overview?WT.mc_id=data-17927-abhishgu), validate it's functionality and also walk-through how it works along with the code.

> As you will see in the later parts of this series, this also provides a foundation for scale-out architecture.

All in all, this blog post is short and simple! It lays down the ground work for other parts of the solution which will be covered in subsequent posts. Please don't worry about the fact that the service is written in Rust (in case you don't know it already). The logic can be easily ported over to your favorite programming language.

## Pre-requisites

Start by getting a [free Azure account](https://azure.microsoft.com/free/?WT.mc_id=data-17927-abhishgu) if you don't have one already and [install the Azure CLI](https://docs.microsoft.com/cli/azure/install-azure-cli?WT.mc_id=data-17927-abhishgu) as well.

We will be deploying the tweets consumer application to [Azure Container Instances](https://docs.microsoft.com/azure/container-instances/container-instances-overview?WT.mc_id=data-17927-abhishgu) using regular Docker CLI commands. This capability is enabled by [integration between Docker and Azure](https://docs.docker.com/engine/context/aci-integration/). Just ensure that you have Docker Desktop version 2.3.0.5 or later, for [Windows](https://desktop.docker.com/win/edge/Docker%20Desktop%20Installer.exe), [macOS](https://desktop.docker.com/mac/edge/Docker.dmg), or install the [Docker ACI Integration CLI for Linux](https://docs.docker.com/engine/context/aci-integration/#install-the-docker-aci-integration-cli-on-linux).

To use the Twitter Streaming API, you will also need a Twitter developer account. If you don't have one already, please [follow these instructions](https://developer.twitter.com/en/apply-for-access) on how to set it up.

## Deploy the app to Azure Container Instances

To start off, setup the Enterprise tier of Azure Cache for Redis, [using this quickstart](https://docs.microsoft.com/azure/azure-cache-for-redis/quickstart-create-redis-enterprise?WT.mc_id=data-17927-abhishgu).Once you finish this step, ensure that you save the following information: the Redis host name and Access key

![](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/gq0ogkhc9hr9co8e04zh.png)

The tweets consumer application is available as a [Docker container](https://hub.docker.com/r/abhirockzz/tweets-redis-streams-producer-rust) - the easiest way is to simply re-use it. If you wish to build you own image, please use the [`Dockerfile` available in the GitHub repo](https://github.com/abhirockzz/redis-streams-in-action/blob/master/tweets-consumer/Dockerfile). 

You will now see how convenient it is to deploy it to Azure Container Instances, that allows you to run Docker containers on-demand in a managed, serverless Azure environment. 

First, [create an Azure context](https://docs.microsoft.com/azure/container-instances/quickstart-docker-cli?WT.mc_id=data-17927-abhishgu#create-azure-context) to associate Docker with an Azure subscription and resource group so you can create and manage container instances.

```bash
docker login azure
docker context create aci aci-context
docker context use aci-context
```

Set the environment variables - make sure to update Redis host and credentials as per your account:

```bash
export REDIS_HOSTNAME=<redis host port e.g. my-redis-host:10000>
export IS_TLS=true
export REDIS_PASSWORD=<redis access key (password)>

# don't forget your twitter api credentials
export TWITTER_API_KEY=<api key>
export TWITTER_API_KEY_SECRET=<api key secret>
export TWITTER_ACCESS_TOKEN=<access token>
export TWITTER_ACCESS_TOKEN_SECRET=<access token secret>
```

Just execute the good old `docker run`:

```bash
docker run -d --name redis-streams-producer \
-e REDIS_HOSTNAME=$REDIS_HOSTNAME \
-e IS_TLS=$IS_TLS \
-e REDIS_PASSWORD=$REDIS_PASSWORD \
-e TWITTER_API_KEY=$TWITTER_API_KEY \
-e TWITTER_API_KEY_SECRET=$TWITTER_API_KEY_SECRET \
-e TWITTER_ACCESS_TOKEN=$TWITTER_ACCESS_TOKEN \
-e TWITTER_ACCESS_TOKEN_SECRET=$TWITTER_ACCESS_TOKEN_SECRET \
abhirockzz/tweets-redis-streams-producer-rust
```

A container should now be created in Azure and you should see an output similar to this:

```bash
[+] Running 2/2
 ⠿ Group redis-streams-producer  Created                                                                             4.2s
 ⠿ redis-streams-producer        Created                                                                            15.8s
```

Validate this using the Azure portal:

![](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/c8a4is8xpv4i6u7w04n2.png)

To check the container logs:

```bash
docker logs redis-streams-producer
```

## So, does it work?

Well, it should! To confirm, connect to the Redis instance using `redis-cli`:

```bash
redis-cli -h <redis cache host> -p <redis port> -a <access key> --tls
```

... and run the [XRANGE](https://redis.io/commands/xrange) command to introspect Redis Streams:

```bash
XRANGE tweets_stream - + COUNT 5
```

This will return the first five tweets. You can change the `COUNT` as per your requirements.

> The `-` and `+` special IDs mean respectively the minimum ID possible and the maximum ID possible inside the stream

That's all you need to confirm that our application is able to consume tweets and add them to Redis Streams. As mentioned before, the rest of the components in our solution will build on top of this foundation.

You can either pause the app for now or delete it:

```bash
#to pause
docker stop redis-streams-producer

#to delete
docker rm redis-streams-producer
```

Now that you've seen the application in action, let's quickly walk through "how" things work. If you're interested in exploring some Rust code, you will find it useful.

## Code walk through

> You can [refer to the code here](https://github.com/abhirockzz/redis-streams-in-action/tree/master/tweets-consumer/src)

The app uses the following libraries:

- A Rust library for accessing the [Twitter Streaming API](https://github.com/tesaguri/twitter-stream-rs) (uses [tokio](https://github.com/tokio-rs/tokio))
- [redis-rs](https://github.com/mitsuhiko/redis-rs), a Rust library for Redis with both high and low-level APIs
- [serde](https://github.com/serde-rs/serde) and [serde json](https://github.com/serde-rs/json)

It starts by connecting to Redis and Twitter:

```rust
fn connect_redis() -> redis::Connection {
    println!("Connecting to Redis");
    let redis_host_name =
        env::var("REDIS_HOSTNAME").expect("missing environment variable REDIS_HOSTNAME");
    let redis_password = env::var("REDIS_PASSWORD").unwrap_or_default();

    //if Redis server needs secure connection
    let uri_scheme = match env::var("IS_TLS") {
        Ok(_) => "rediss",
        Err(_) => "redis",
    };

    let redis_conn_url = format!("{}://:{}@{}", uri_scheme, redis_password, redis_host_name);
    println!("redis_conn_url {}", redis_conn_url);

    let client = redis::Client::open(redis_conn_url).expect("check Redis connection URL");
    client.get_connection().expect("failed to connect to Redis")
}
```

Rather than follow a specific set of keywords or a user, we simply connect to the [Twitter sample stream](https://developer.twitter.com/en/docs/twitter-api/tweets/sampled-stream/api-reference/get-tweets-sample-stream), which provides access to about 1% of all Tweets in real-time:

```rust
    let token = twitter_token();

    TwitterStream::sample(&token)
        .try_flatten_stream()
        .try_for_each(|json| {
            let msg: model::StreamMessage =
                serde_json::from_str(&json).expect("failed to convert tweet JSON to struct");
            process(msg, c.clone());
            future::ok(())
        })
        .await
        .expect("error connecting to Twitter stream!");
```

Bulk of the logic is encapsulated in the `process` function. Let's go through that bit by bit.
`twitter-stream` crate returns each tweet in raw JSON form. It is converted into a `model::StreamMessage` which is a struct that's modeled as per the data we intend to extract from the raw tweet.

We use `serde_json` to get this done:

```rust
serde_json::from_str(&json).expect("json to struct conversion failed");
```

It is then passed to the `process` function along with a [redis::Connection](https://docs.rs/redis/0.20.0/redis/struct.Connection.html).

```rust
let conn = connect_redis();
let c = Arc::new(Mutex::new(conn));
...
fn process(msg: model::StreamMessage, conn: Arc<Mutex<redis::Connection>>) {
    //omitted
}
```

But why wrap it within an [Arc](https://doc.rust-lang.org/std/sync/struct.Arc.html) of [Mutex](https://doc.rust-lang.org/std/sync/struct.Mutex.html)?

That's because we need to pass the `redis::Connection` to a `FnMut` closure. It [moves](https://doc.rust-lang.org/book/ch04-01-what-is-ownership.html) the connection, thus we need to use a shared reference, which `Arc` provides. But `Arc` is not enough since we are not allowed to mutate the data. Thus, we use to use a `Mutex` to lock the connection object - the Rust compiler can be confident that only one thread can access it at a time (preserve immutability)

The processing part is relatively simple. It's all about using the [xadd_map](https://docs.rs/redis/0.20.0/redis/trait.Commands.html#method.xadd_map) function to add the tweet to a Redis Stream. It accepts a [BTreeMap](https://doc.rust-lang.org/std/collections/struct.BTreeMap.html), which we create from info in `model::StreamMessage` - the tweet text, twitter user (screen) name, ID, location and hashtags (if any). Ultimately, the goal is to be able to index these in `RediSearch` and query them flexibly.


```rust
let mut stream_entry: BTreeMap<String, String> = BTreeMap::new();
                stream_entry.insert("id".to_string(), tweet.id.to_string());
                stream_entry.insert("user".to_string(), tweet.user.screen_name);
                stream_entry.insert("text".to_string(), tweet.text);
                stream_entry.insert("location".to_string(), tweet.user.location);
```

That's all for this part.

## Moving on to the next one...

We're just getting started! This was the first component in our service that lays the foundation for processing the tweets and making them queryable via `RediSearch`. In the upcoming blog, we will dive into how to consume and process tweets from Redis Streams using a Java based application. Stay tuned!