On this page

Rxjava3

1 Introduction

Adds support for RxJava 3 to a Micronaut application. The following integrations are included:

Converters for RxJava 3 types so that RxJava 3 types can be used in controllers and clients
Instrumentation for RxJava 3 types so that tracing works with RxJava 3 types
A version of the Http client that supports RxJava 3

1.1 Release History

For this project, you can find a list of releases (with release notes) here:

https://github.com/micronaut-projects/micronaut-rxjava3/releases

2 Quick Start

Add the following dependency to your Micronaut application:

implementation("io.micronaut.rxjava3:micronaut-rxjava3")

<dependency>
    <groupId>io.micronaut.rxjava3</groupId>
    <artifactId>micronaut-rxjava3</artifactId>
</dependency>

To use the HTTP client add the following dependency:

implementation("io.micronaut.rxjava3:micronaut-rxjava3-http-client")

<dependency>
    <groupId>io.micronaut.rxjava3</groupId>
    <artifactId>micronaut-rxjava3-http-client</artifactId>
</dependency>

To use the RxJava 3 variation of the Micronaut HTTP client inject the Rx3HttpClient interface (or one of the other variants). For example:

Injecting the RxJava 3 clients

import io.micronaut.http.client.annotation.Client;
import io.micronaut.rxjava3.http.client.Rx3HttpClient;
import io.micronaut.rxjava3.http.client.Rx3StreamingHttpClient;
import io.micronaut.rxjava3.http.client.Rx3SseClient;
import jakarta.inject.Inject;

    @Inject @Client("/") Rx3HttpClient httpClient; // 
    @Inject @Client("/") Rx3SseClient sseClient; // 
    @Inject @Client("/") Rx3StreamingHttpClient streamingClient; //

from typing import Annotated

from jakarta.inject import Inject
from micronaut.http.client.annotation import Client
from micronaut.rxjava3.http.client import Rx3HttpClient, Rx3SseClient, Rx3StreamingHttpClient

    http_client: Annotated[Rx3HttpClient, Inject, Client("/")]  # 
    sse_client: Annotated[Rx3SseClient, Inject, Client("/")]  # 
    streaming_client: Annotated[Rx3StreamingHttpClient, Inject, Client("/")]  #

import io.micronaut.http.client.annotation.Client
import io.micronaut.rxjava3.http.client.Rx3HttpClient
import io.micronaut.rxjava3.http.client.Rx3StreamingHttpClient
import io.micronaut.rxjava3.http.client.Rx3SseClient
import jakarta.inject.Inject

    @Inject @field:Client("/") lateinit var httpClient: Rx3HttpClient // 
    @Inject @field:Client("/") lateinit var sseClient: Rx3SseClient // 
    @Inject @field:Client("/") lateinit var streamingClient: Rx3StreamingHttpClient //

import io.micronaut.http.client.annotation.Client
import io.micronaut.rxjava3.http.client.Rx3HttpClient
import io.micronaut.rxjava3.http.client.Rx3StreamingHttpClient
import io.micronaut.rxjava3.http.client.Rx3SseClient
import jakarta.inject.Inject

    @Inject @Client("/") Rx3HttpClient httpClient // 
    @Inject @Client("/") Rx3SseClient sseClient // 
    @Inject @Client("/") Rx3StreamingHttpClient streamingClient //

The regular client, Rx3HttpClient
The server sent events client, Rx3SseClient
The streaming client, Rx3StreamingHttpClient

The methods of these clients return RxJava 3 types (Flowable, Single, Maybe) instead of Reactive Streams Publisher instances:

Using the RxJava 3 client

String greeting = httpClient.retrieve("/hello").blockingFirst();

greeting = self.http_client.retrieve("/hello").blockingFirst()

val greeting = httpClient.retrieve("/hello").blockingFirst()

String greeting = httpClient.retrieve("/hello").blockingFirst()

3 Repository

You can find the source code of this project in this repository:

https://github.com/micronaut-projects/micronaut-rxjava3