SmallRye Mutiny Vert.x Bindingsを使って、Redisにアクセスしてみる（＋AssertSubscriberを試す）

$ java --version
openjdk 11.0.11 2021-04-20
OpenJDK Runtime Environment (build 11.0.11+9-Ubuntu-0ubuntu2.20.04)
OpenJDK 64-Bit Server VM (build 11.0.11+9-Ubuntu-0ubuntu2.20.04, mixed mode, sharing)


$ mvn --version
Apache Maven 3.8.2 (ea98e05a04480131370aa0c110b8c54cf726c06f)
Maven home: $HOME/.sdkman/candidates/maven/current
Java version: 11.0.11, vendor: Ubuntu, runtime: /usr/lib/jvm/java-11-openjdk-amd64
Default locale: ja_JP, platform encoding: UTF-8
OS name: "linux", version: "5.4.0-84-generic", arch: "amd64", family: "unix"

Redisは6.2.5を使い、172.17.0.2のIPアドレスで動作しているものとします。また、パスワードはredispassで設定している
ものとします。

準備

Maven依存関係はこちら。

    <dependencies>
        <dependency>
            <groupId>io.smallrye.reactive</groupId>
            <artifactId>smallrye-mutiny-vertx-redis-client</artifactId>
            <version>2.13.0</version>
        </dependency>

        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <version>5.7.2</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-engine</artifactId>
            <version>5.7.2</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.assertj</groupId>
            <artifactId>assertj-core</artifactId>
            <version>3.20.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.22.2</version>
            </plugin>
        </plugins>
    </build>

今回の主は、こちらのsmallrye-mutiny-vertx-redis-clientですね。

        <dependency>
            <groupId>io.smallrye.reactive</groupId>
            <artifactId>smallrye-mutiny-vertx-redis-client</artifactId>
            <version>2.13.0</version>
        </dependency>

よく見ると、SmallRye Mutinyって1.0.0になっていたんですね。

        <mutiny.version>1.0.0</mutiny.version>

https://github.com/smallrye/smallrye-mutiny-vertx-bindings/blob/2.13.0/pom.xml#L50

Release 1.0.0 · smallrye/smallrye-mutiny · GitHub

テストコードの雛形

確認は、テストコードで行います。まずは雛形を。

src/test/java/org/littlewings/smallrye/mutiny/RedisClientTest.java

package org.littlewings.smallrye.mutiny;

import java.util.List;

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.helpers.test.AssertSubscriber;
import io.smallrye.mutiny.helpers.test.UniAssertSubscriber;
import io.vertx.mutiny.core.Vertx;
import io.vertx.mutiny.redis.client.Redis;
import io.vertx.mutiny.redis.client.RedisAPI;
import io.vertx.mutiny.redis.client.Response;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;

public class RedisClientTest {

    // ここに、テストを書く！！
}

import文がちょっとポイントですが、Vert.xのクラスを含め、SmallRye Mutinyのものを使います。

import io.vertx.mutiny.core.Vertx;
import io.vertx.mutiny.redis.client.Redis;
import io.vertx.mutiny.redis.client.RedisAPI;
import io.vertx.mutiny.redis.client.Response;

簡単にset／get／delしてみる

SmallRye Mutiny Vert.x Bindingsを使って、Redisにアクセスしてみます。簡単にset／get／delしてみましょう。

    @Test
    public void gettingStarted() {
        Vertx vertx = Vertx.vertx();

        Redis redis = Redis.createClient(vertx, "redis://:redispass@172.17.0.2:6379/0");
        RedisAPI redisApi = RedisAPI.api(redis);

        Uni<Response> setResponse = redisApi.set(List.of("key1", "value1"));
        Uni<Response> getResponse =
                setResponse
                        .onItem()
                        .transformToUni(r -> redisApi.get("key1"));

        UniAssertSubscriber<Response> getAssertSubscriber =
                getResponse
                        .subscribe()
                        .withSubscriber(UniAssertSubscriber.create());

        Response actualGetResponse =
                getAssertSubscriber
                        .awaitItem()
                        .assertCompleted()
                        .getItem();

        assertThat(actualGetResponse).asString().isEqualTo("value1");

        Uni<Response> delResponse = redisApi.del(List.of("key1"));
        Uni<Response> emptyResponse =
                delResponse
                        .onItem()
                        .transformToUni(r -> redisApi.get("key1"));

        UniAssertSubscriber<Response> emptyAssertSubscriber =
                emptyResponse
                        .subscribe()
                        .withSubscriber(UniAssertSubscriber.create());

        Response actualEmptyResponse =
                emptyAssertSubscriber
                        .awaitItem()
                        .assertCompleted()
                        .getItem();

        assertThat(actualEmptyResponse).isNull();

        redisApi.close();

        vertx.closeAndAwait();
    }

set、get。

        Uni<Response> setResponse = redisApi.set(List.of("key1", "value1"));
        Uni<Response> getResponse =
                setResponse
                        .onItem()
                        .transformToUni(r -> redisApi.get("key1"));

getの結果はUniで返ってくるので、確認にはUniAssertSubscriberを使いましょう。

        UniAssertSubscriber<Response> getAssertSubscriber =
                getResponse
                        .subscribe()
                        .withSubscriber(UniAssertSubscriber.create());

        Response actualGetResponse =
                getAssertSubscriber
                        .awaitItem()
                        .assertCompleted()
                        .getItem();

        assertThat(actualGetResponse).asString().isEqualTo("value1");

ちょっとハマったのがawaitItemで、これをつけないと動かなかったのですが…なにがマズいんでしょうね？

オマケ：Multiで試す

UniAssertSubscriberでちょっとハマったので、Multiも扱ってみることにしました。

Redis Clientで直接Multiを返すものはなかったので、mgetの結果をMultiに変換して扱うことにします。
まあ、これはこれでハマる理由になったのですが。

作成したテストコードは、こちら。

    @Test
    public void assertMulti() {
        Vertx vertx = Vertx.vertx();

        Redis redis = Redis.createClient(vertx, "redis://:redispass@172.17.0.2:6379/0");
        RedisAPI redisApi = RedisAPI.api(redis);

        Uni<Response> msetResponse = redisApi.mset(List.of("key1", "value1", "key2", "value2", "key3", "value3"));
        Uni<Response> mgetResponse =
                msetResponse.onItem().transformToUni(r -> redisApi.mget(List.of("key1", "key2", "key3")));

        Multi<String> multiMgetResponse =
                mgetResponse
                        .onItem()
                        .transformToMulti(item -> item.toMulti()).onItem().transform(item -> item.toString());

        AssertSubscriber<String> mgetAssertSubscriber =
                multiMgetResponse
                        .subscribe()
                        .withSubscriber(AssertSubscriber.create(10));

        List<String> actualMgetResponse =
                mgetAssertSubscriber
                        .awaitItems(3)
                        .assertCompleted()
                        .getItems();

        assertThat(actualMgetResponse).hasSize(3).containsExactly("value1", "value2", "value3");

        Uni<Response> delResponse = redisApi.del(List.of("key1", "key2", "key3"));
        Uni<Response> emptyResponse =
                delResponse.onItem().transformToUni(r -> redisApi.mget(List.of("key1", "key2", "key3")));

        UniAssertSubscriber<Response> emptyAssertSubscriber =
                emptyResponse
                        .subscribe()
                        .withSubscriber(UniAssertSubscriber.create());

        Response actualEmptyResponse =
                emptyAssertSubscriber
                        .awaitItem()
                        .assertCompleted()
                        .getItem();

        assertThat(actualEmptyResponse)
                .hasSize(3)
                .containsOnlyNulls();

        redisApi.close();

        vertx.closeAndAwait();
    }

mset／mgetして、mgetの結果は個々の要素をバラしてUniからMultiに変換。

        Uni<Response> msetResponse = redisApi.mset(List.of("key1", "value1", "key2", "value2", "key3", "value3"));
        Uni<Response> mgetResponse =
                msetResponse.onItem().transformToUni(r -> redisApi.mget(List.of("key1", "key2", "key3")));

        Multi<String> multiMgetResponse =
                mgetResponse
                        .onItem()
                        .transformToMulti(item -> item.toMulti()).onItem().transform(item -> item.toString());

Multiの場合は、AssertSubscriberを使います。

        AssertSubscriber<String> mgetAssertSubscriber =
                multiMgetResponse
                        .subscribe()
                        .withSubscriber(AssertSubscriber.create(10));

        List<String> actualMgetResponse =
                mgetAssertSubscriber
                        .awaitItems(3)
                        .assertCompleted()
                        .getItems();

        assertThat(actualMgetResponse).hasSize(3).containsExactly("value1", "value2", "value3");

ポイントはAssertSubscriber#createとAssertSubscriber#awaitItemsです。

AssertSubscriber#createに与える値は、少なくとも期待する要素数以上を指定する必要があります。今回はMultiに
3つの要素が含まれるので、3以上であればOKです。

                        .withSubscriber(AssertSubscriber.create(10));

AssertSubscriber#awaitItemsについては、期待する数をきっちり指定する必要があります。今回は、3以外だとうまくいかない
ことになります。

                mgetAssertSubscriber
                        .awaitItems(3)

ところで、AssertSubscriber#awaitItemsが必要な理由もやっぱりよくわかりませんでした…。

delしてmget。このコードの場合、データは削除した後なのでmgetの結果はすべてnullになるのですが、この状態の
UniをMultiに変換しようとするとうまくいかない（nullを「失敗」と見なす模様）のでそのままUniにすることに
しました…。

        Uni<Response> delResponse = redisApi.del(List.of("key1", "key2", "key3"));
        Uni<Response> emptyResponse =
                delResponse.onItem().transformToUni(r -> redisApi.mget(List.of("key1", "key2", "key3")));

        UniAssertSubscriber<Response> emptyAssertSubscriber =
                emptyResponse
                        .subscribe()
                        .withSubscriber(UniAssertSubscriber.create());

        Response actualEmptyResponse =
                emptyAssertSubscriber
                        .awaitItem()
                        .assertCompleted()
                        .getItem();

        assertThat(actualEmptyResponse)
                .hasSize(3)
                .containsOnlyNulls();

これはこれで、ずいぶんハマりましたが…。

まとめ

SmallRye Mutiny Vert.x Bindingsの最近の情報を調べつつ、Redis Clientで遊び、AssertSubscriberでハマってみました。

いつの間にか、SmallRye Mutinyも1.0.0になっていたんですね。

SmallRye Mutiny Vert.x Bindingsの方も含めて、実装、ドキュメントともにいろいろ増えてきて良いなと思います。
もうちょっと扱って慣れていかないと…。