For example to write a client that streams data from the controller defined in the section of the documentation you can simply define a client that returns an unbound Publisher such as a RxJava or Reactor :

HeadlineClient.java

HeadlineClient.java

import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Get
import io.micronaut.http.client.annotation.Client
import io.reactivex.Flowable
@Client("/streaming")
interface HeadlineClient {
    @Get(value = "/headlines", processes = MediaType.APPLICATION_JSON_STREAM) (1)
    Flowable<Headline> streamHeadlines() (2)
}

HeadlineClient.java

import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Get
import io.micronaut.http.client.annotation.Client
import io.reactivex.Flowable
@Client("/streaming")
interface HeadlineClient {
    @Get(value = "/headlines", processes = [MediaType.APPLICATION_JSON_STREAM]) (1)
    fun streamHeadlines(): Flowable<Headline>  (2)

The following example shows how the previously defined HeadlineClient can be invoked from a JUnit test:

Streaming HeadlineClient

    @Test
    public void testClientAnnotationStreaming() throws Exception {
        try( EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer.class) ) {
            HeadlineClient headlineClient = embeddedServer
                                                .getApplicationContext()
                                                .getBean(HeadlineClient.class); (1)
            Headline headline = firstHeadline.blockingGet(); (3)
            assertNotNull( headline );
            assertTrue( headline.getText().startsWith("Latest Headline") );
        }

Streaming HeadlineClient

    void "test client annotation streaming"() throws Exception {
        when:
        HeadlineClient headlineClient = embeddedServer.getApplicationContext()
                                            .getBean(HeadlineClient.class) (1)
        Maybe<Headline> firstHeadline = headlineClient.streamHeadlines().firstElement() (2)
        Headline headline = firstHeadline.blockingGet() (3)
        then:
        null != headline
        headline.getText().startsWith("Latest Headline")
    }

Streaming HeadlineClient

The example defined in the previous section expects the server to respond with a stream of JSON objects and the content type to be application/x-json-stream. For example:

{"title":"The Stand"}
{"title":"The Shining"}

The reason for this is simple, a sequence of JSON object is not, in fact, valid JSON and hence the response content type cannot be application/json. For the JSON to be valid it would have to return an array:

A JSON Array

[
    {"title":"The Stand"},
    {"title":"The Shining"}
]

Micronaut’s client does however support streaming of both individual JSON objects via application/x-json-stream and also JSON arrays defined with application/json.

If the server returns application/json and a non-single Publisher is returned (such as an or a Reactor Flux) then the client with stream the array elements as they become available.

When streaming responses from servers, the underlying HTTP client will not apply the default readTimeout setting (which defaults to 10 seconds) of the HttpClientConfiguration since the delay between reads for streaming responses may differ from normal reads.

Instead the read-idle-timeout setting (which defaults to 60 seconds) is used to dictate when a connection should be closed after becoming idle.

If you are streaming data from a server that defines a longer delay than 60 seconds between items being sent to the client you should adjust the readIdleTimeout. The following configuration in application.yml demonstrates how:

Adjusting the readIdleTimeout

micronaut:
    http:
        client:
            read-idle-timeout: 5m

The above example sets the readIdleTimeout to 5 minutes.

You can use this client to stream SSE events from any server that emits them.

The annotation also supports consuming SSE streams. For example, consider the following controller method that produces a stream of SSE events:

SSE Controller

    @Get(value = "/headlines", processes = MediaType.TEXT_EVENT_STREAM) (1)
    Flowable<Event<Headline>> streamHeadlines() {
        return Flowable.<Event<Headline>>create((emitter) -> {  (2)
            emitter.onNext(Event.of(headline));
            emitter.onComplete();
        }, BackpressureStrategy.BUFFER).repeat(100) (3)
          .delay(1, TimeUnit.SECONDS); (4)
    }

SSE Controller

SSE Controller

    @Get(value = "/headlines", processes = [MediaType.TEXT_EVENT_STREAM]) (1)
    internal fun streamHeadlines(): Flowable<Event<Headline>> {
        return Flowable.create<Event<Headline>>( {  (2)
            emitter ->
            val headline = Headline()
            headline.text = "Latest Headline at " + ZonedDateTime.now()
            emitter.onNext(Event.of(headline))
            emitter.onComplete()
        }, BackpressureStrategy.BUFFER).repeat(100) (3)
                .delay(1, TimeUnit.SECONDS) (4)
    }

Notice that the return type of the controller is also Event and that the Event.of method is used to create events to stream to the client.

To define a client that consumes the events you simply have to define a method that processes MediaType.TEXT_EVENT_STREAM:

SSE Client

@Client("/streaming/sse")
public interface HeadlineClient {
    @Get(value = "/headlines", processes = MediaType.TEXT_EVENT_STREAM)
    Flowable<Event<Headline>> streamHeadlines();
}

SSE Client

@Client("/streaming/sse")
interface HeadlineClient {
    @Get(value = "/headlines", processes = MediaType.TEXT_EVENT_STREAM)
    Flowable<Event<Headline>> streamHeadlines()
}

@Client("/streaming/sse")
interface HeadlineClient {
    @Get(value = "/headlines", processes = [MediaType.TEXT_EVENT_STREAM])
    fun streamHeadlines(): Flowable<Event<Headline>>
}

The generic type of the or Flowable can be either an , in which case you will receive the full event object, or a POJO, in which case you will receive only the data contained within the event converted from JSON.