Given the class Thing, which can be represented in JSON:

public class Thing {

   private String name;

   public Thing() {
   }

   public Thing(String name) {
      this.name = name;
   }
   ...
}

the method postThingList() in the following is a valid resource method:

@POST
@Path("post/thing/list")
@Produces(MediaType.APPLICATION_JSON)
@Stream
public Flowable<List<Thing>> postThingList(String s) {
   return buildFlowableThingList(s, 2, 3);
}

static Flowable<List<Thing>> buildFlowableThingList(String s, int listSize, int elementSize) {
   return Flowable.create(
      new FlowableOnSubscribe<List<Thing>>() {

         @Override
         public void subscribe(FlowableEmitter<List<Thing>> emitter) throws Exception {
            for (int i = 0; i < listSize; i++) {
               List<Thing> list = new ArrayList<Thing>();
               for (int j = 0; j < elementSize; j++) {
                  list.add(new Thing(s));
               }
               emitter.onNext(list);
            }
            emitter.onComplete();
         }
      },
      BackpressureStrategy.BUFFER);
}

The somewhat imposing method buildFlowableThingList() probably deserves some explanation. First,

Flowable<List<Thing>> Flowable.create(FlowableOnSubscribe<List<Thing>> source, BackpressureStrategy mode);

creates a Flowable<List<Thing>> by describing what should happen when the Flowable<List<Thing>> is subscribed to. FlowableEmitter<List<Thing>> extends io.reactivex.Emitter<List<Thing>>:

/**
 * Base interface for emitting signals in a push-fashion in various generator-like source
 * operators (create, generate).
 *
 * @param <T> the value type emitted
 */
public interface Emitter<T> {

    /**
     * Signal a normal value.
     * @param value the value to signal, not null
     */
    void onNext(@NonNull T value);

    /**
     * Signal a Throwable exception.
     * @param error the Throwable to signal, not null
     */
    void onError(@NonNull Throwable error);

    /**
     * Signal a completion.
     */
    void onComplete();
}

and FlowableOnSubscribe uses a FlowableEmitter to send out values from the Flowable<List<Thing>>:

/**
 * A functional interface that has a {@code subscribe()} method that receives
 * an instance of a {@link FlowableEmitter} instance that allows pushing
 * events in a backpressure-safe and cancellation-safe manner.
 *
 * @param <T> the value type pushed
 */
public interface FlowableOnSubscribe<T> {

    /**
     * Called for each Subscriber that subscribes.
     * @param e the safe emitter instance, never null
     * @throws Exception on error
     */
    void subscribe(@NonNull FlowableEmitter<T> e) throws Exception;
}

So, what will happen when a subscription to the Flowable<List<Thing>> is created is, the FlowableEmitter.onNext() will be called, once for each <List<Thing>> created, followed by a call to FlowableEmitter.onComplete() to indicate that the sequence has ended. Under the covers, RESTEasy subscribes to the Flowable<List<Thing>> and handles each element passed in by way of onNext().

On the client side, JAX-RS 2.1 supports extensions for reactive classes by adding the method

/**
 * Access a reactive invoker based on a {@link RxInvoker} subclass provider. Note
 * that corresponding {@link RxInvokerProvider} must be registered in the client runtime.
 * 
 * This method is an extension point for JAX-RS implementations to support other types
 * representing asynchronous computations.
 *
 * @param clazz {@link RxInvoker} subclass.
 * @return reactive invoker instance.
 * @throws IllegalStateException when provider for given class is not registered.
 * @see javax.ws.rs.client.Client#register(Class)
 * @since 2.1
 */
public <T extends RxInvoker> T rx(Class<T> clazz);

to interface javax.ws.rs.client.Invocation.Builder. Resteasy module resteasy-rxjava2 adds support for classes:

which allow accessing Singles, Observables, and Flowables on the client side.

For example, given the resource method postThingList() above, a Flowable<List<Thing>> can be retrieved from the server by calling

@SuppressWarnings("unchecked")
@Test
public void testPostThingList() throws Exception {
   CountDownLatch latch = new CountdownLatch(1);
   FlowableRxInvoker invoker = client.target(generateURL("/post/thing/list")).request().rx(FlowableRxInvoker.class);
   Flowable<List<Thing>> flowable = (Flowable<List<Thing>>) invoker.post(Entity.entity("a", MediaType.TEXT_PLAIN_TYPE), new GenericType<List<Thing>>() {});
   flowable.subscribe(
         (List<?> l) -> thingListList.add(l),
         (Throwable t) -> latch.countDown(),
         () -> latch.countDown());
   latch.await();
   Assert.assertEquals(aThingListList, thingListList);
}

where aThingListList is

[[Thing[a], Thing[a], Thing[a]], [Thing[a], Thing[a], Thing[a]]]

Note the call to Flowable.suscribe(). On the server side, RESTEasy subscribes to a returning Flowable in order to receive its elements and send them over the wire. On the client side, the user subscribes to the Flowable in order to receive its elements and do whatever it wants to with them. In this case, three lambdas determine what should happen 1) for each element, 2) if a Throwable is thrown, and 3) when the Flowable is done passing elements.

Neither Reactive Streams nor JAX-RS have anything to say about representing reactive types on the network. RESTEasy offers a number of representations, each suitable for different circumstances. The wire protocol is determined by 1) the presence or absence of the @Stream annotation on the resource method, and 2) the value of the value field in the @Stream annotation:

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Stream
{
   public enum MODE {RAW, GENERAL};
   public String INCLUDE_STREAMING_PARAMETER = "streaming";
   public MODE value() default MODE.GENERAL;
   public boolean includeStreaming() default false;
}

Note that MODE.GENERAL is the default value, so @Stream is equivalent to @Stream(Stream.MODE.GENERAL).

Example 1.

@POST
@Path("post/thing/list")
@Produces(MediaType.APPLICATION_JSON)
@Stream(Stream.MODE.GENERAL)
public Flowable<List<Thing>> postThingList(String s) {
   return buildFlowableThingList(s, 2, 3);
}
...
@SuppressWarnings("unchecked")
@Test
public void testPostThingList() throws Exception {
   CountDownLatch latch = new CountdownLatch(1);
   FlowableRxInvoker invoker = client.target(generateURL("/post/thing/list")).request().rx(FlowableRxInvoker.class);
   Flowable<List<Thing>> flowable = (Flowable<List<Thing>>) invoker.post(Entity.entity("a", MediaType.TEXT_PLAIN_TYPE), new GenericType<List<Thing>>() {});
   flowable.subscribe(
         (List<?> l) -> thingListList.add(l),
         (Throwable t) -> latch.countDown(),
         () -> latch.countDown());
   latch.await();
   Assert.assertEquals(aThingListList, thingListList);
}

This is the example given previously, except that the mode in the @Stream annotation (which defaults to MODE.GENERAL) is given explicitly. In this scenario, the Flowable emits <List<Thing>> elements on the server, they are transmitted over the wire as SSE events:

data: [{"name":"a"},{"name":"a"},{"name":"a"}]
data: [{"name":"a"},{"name":"a"},{"name":"a"}]

and the FlowableRxInvoker reconstitutes a Flowable on the client side.

Example 2.

@POST
@Path("post/thing/list")
@Produces(MediaType.APPLICATION_JSON)
public Flowable<List<Thing>> postThingList(String s) {
   return buildFlowableThingList(s, 2, 3);
}
...
@Test
public void testPostThingList() throws Exception {
   Builder request = client.target(generateURL("/post/thing/list")).request();
   List<List<Thing>> list = request.post(Entity.entity("a", MediaType.TEXT_PLAIN_TYPE), new GenericType<List<List<Thing>>>() {});
   Assert.assertEquals(aThingListList, list);
}  

In this scenario, in which the resource method has no @Stream annotation, the Flowable emits stream elements which are accumulated by the server until the Flowable is done, at which point the entire JSON list is transmitted over the wire:

[[{"name":"a"},{"name":"a"},{"name":"a"}],[{"name":"a"},{"name":"a"},{"name":"a"}]]

and the list is reconstituted on the client side by an ordinary invoker.

Example 3.

@GET
@Path("get/bytes")
@Produces(MediaType.APPLICATION_OCTET_STREAM)
@Stream(Stream.MODE.RAW)
public Flowable<byte[]> getBytes() {
   return Flowable.create(
      new FlowableOnSubscribe<byte[]>() {

         @Override
         public void subscribe(FlowableEmitter<byte[]> emitter) throws Exception {
            for (int i = 0; i < 3; i++) {
               byte[] b = new byte[10];
               for (int j = 0; j < 10; j++) {
                  b[j] = (byte) (i + j);
               }
               emitter.onNext(b);
            }
            emitter.onComplete();
         }
      },
      BackpressureStrategy.BUFFER);
}
...
@Test
public void testGetBytes() throws Exception {
   Builder request = client.target(generateURL("/get/bytes")).request();
   InputStream is = request.get(InputStream.class);
   int n = is.read();
   while (n > -1) {
      System.out.print(n);
      n = is.read();
   }
}

Here, the byte arrays are written to the network as they are created by the Flowable. On the network, they are concatenated, so the client sees one stream of bytes.

Since general streaming mode and SSE share minor variants of the same wire protocol, they are, modulo the SSE restriction to character data, interchangeable. That is, an SSE client can connect to a resource method that returns a Flowable or an Observable, and a FlowableRxInvoker, for example, can connect to an SSE resource method.

Note. SSE requires a @Produces("text/event-stream") annotation, so, unlike the cases of raw and general streaming, the element media type cannot be derived from the @Produces annotation. To solve this problem, Resteasy introduces the

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface SseElementType
{
   public String value();
}

annotation, from which the element media type is derived.

Example 1.

@GET
@Path("eventStream/thing")
@Produces("text/event-stream")
@SseElementType("application/json")
public void eventStreamThing(@Context SseEventSink eventSink, @Context Sse sse) {
   new ScheduledThreadPoolExecutor(5).execute(() -> {
      try (SseEventSink sink = eventSink) {
         OutboundSseEvent.Builder  builder = sse.newEventBuilder();
         eventSink.send(builder.data(new Thing("e1")).build());
         eventSink.send(builder.data(new Thing("e2")).build());
         eventSink.send(builder.data(new Thing("e3")).build());
      }
   });
}
...
@SuppressWarnings("unchecked")
@Test
public void testFlowableToSse() throws Exception {
   CountDownLatch latch = new CountDownLatch(1);
   final AtomicInteger errors = new AtomicInteger(0);
   FlowableRxInvoker invoker = client.target(generateURL("/eventStream/thing")).request().rx(FlowableRxInvoker.class);
   Flowable<Thing> flowable = (Flowable<Thing>) invoker.get(Thing.class);
   flowable.subscribe(
      (Thing t) -> thingList.add(t),
      (Throwable t) -> errors.incrementAndGet(),
      () -> latch.countDown());
   boolean waitResult = latch.await(30, TimeUnit.SECONDS);
   Assert.assertTrue("Waiting for event to be delivered has timed out.", waitResult);
   Assert.assertEquals(0, errors.get());
   Assert.assertEquals(eThingList, thingList);
}  

Here, a FlowableRxInvoker is connecting to an SSE resource method. On the network, the data looks like

data: {"name":"e1"}
data: {"name":"e2"}
data: {"name":"e3"}

Note that the character data is suitable for an SSE resource method.

Also, note that the eventStreamThing() method in this example induces the media type

    text/event-stream;element-type="application/json"

Example 2.

@GET
@Path("flowable/thing")
@Produces("text/event-stream")
@SseElementType("application/json")
public Flowable<Thing> flowableSSE() {
   return Flowable.create(
      new FlowableOnSubscribe<Thing>() {

         @Override
         public void subscribe(FlowableEmitter<Thing> emitter) throws Exception {
            emitter.onNext(new Thing("e1"));
            emitter.onNext(new Thing("e2"));
            emitter.onNext(new Thing("e3"));
            emitter.onComplete();
         }
      },
      BackpressureStrategy.BUFFER);
}
...
@Test
public void testSseToFlowable() throws Exception {
   final CountDownLatch latch = new CountDownLatch(3);
   final AtomicInteger errors = new AtomicInteger(0);
   WebTarget target = client.target(generateURL("/flowable/thing"));
   SseEventSource msgEventSource = SseEventSource.target(target).build();
   try (SseEventSource eventSource = msgEventSource)
   {
      eventSource.register(
         event -> {thingList.add(event.readData(Thing.class, MediaType.APPLICATION_JSON_TYPE)); latch.countDown();},
         ex -> errors.incrementAndGet());
      eventSource.open();

      boolean waitResult = latch.await(30, TimeUnit.SECONDS);
      Assert.assertTrue("Waiting for event to be delivered has timed out.", waitResult);
      Assert.assertEquals(0, errors.get());
      Assert.assertEquals(eThingList, thingList);
   }
}

Here, an SSE client is connecting to a resource method that returns a Flowable. Again, the server is sending character data, which is suitable for the SSE client, and the data looks the same on the network.

Whether or not it is appropriate to stream a list of values is a judgment call. Certainly, if the list is unbounded, then it isn't practical, or even possible, perhaps, to collect the entire list and send it at once. In other cases, the decision is less obvious.

Case 1. Suppose that all of the elements are producible quickly. Then the overhead of sending them independently is probably not worth it.

Case 2. Suppose that the list is bounded but the elements will be produced over an extended period of time. Then returning the initial elements when they become available might lead to a better user experience.

Case 3. Suppose that the list is bounded and the elements can be produced in a relatively short span of time but only after some delay. Here is a situation that illustrates the fact that asynchronous reactive processing and streaming over the network are independent concepts. In this case it's worth considering having the resource method return something like CompletionStage<List<Thing>> rather than Flowable<List<Thing>>. This has the benefit of creating the list asynchronously but, once it is available, sending it to the client in one piece.