Returning a stream from a Spring REST Controller

Question

I am curios if it is possible to return a Stream from a Spring RestController

@RestController
public class X {
  @RequestMapping(...)
  public Stream<?> getAll() { ... }
}

Is it ok to do something like this? I tried and Spring returns something else other than the values of a stream.

Shall I keep returning a List<?>?

Answer 1

This can also be accomplished with Spring MVC Controller, but there are a few concerns: limitations in Spring Data JPA Repository, whether the database supports Holdable Cursors (ResultSet Holdability) and the version of Jackson.

The key concept, I struggled to appreciate, is that a Java 8 Stream returns a series of functions which execute in a terminal operation, and therefore the database has to be accessible in the context executing the terminal operation.

Spring Data JPA Limitations

I found the Spring Data JPA documentation does not provide enough detail for Java 8 Streams. It looks like you can simply declare Stream<MyObject> readAll(), but I needed to annotate the method with @Query to make it work. I was also not able to use a JPA criteria API Specification. So I had to settle for a hard-coded query like:

@Query("select mo from MyObject mo where mo.foo.id in :fooIds")
Stream<MyObject> readAllByFooIn(@Param("fooIds") Long[] fooIds);

Holdable Cursor

If you have a database supporting Holdable Cursors, the result set is accessible after the transaction is committed. This is important since we typically annotate our @Service class methods with @Transactional, so if your database supports holdable cursors the ResultSet can be accessed after the service method returns, i.e. in the @Controller method. If the database does not support holdable cursors, e.g. MySQL, you'll need to add the @Transaction annotation to the controller's @RequestMapping method.

So now the ResultSet is accessible outside the @Service method, right? That again depends on holdability. For MySQL, it's only accessible within the @Transactional method, so the following will work (though defeats the whole purpose of using Java 8 Streams):

@Transaction @RequestMapping(...)
public List<MyObject> getAll() {
   try(Stream<MyObject> stream = service.streamAll) {
        return stream.collect(Collectors.toList())
    };
}

but not

@Transaction @RequestMapping
public Stream<MyObject> getAll() {
    return service.streamAll;
}

because the terminal operator is not in your @Controller it happens in Spring after the controller method returns.

Serializing a stream to JSON without Holdable Cursor support

To serialize the stream to JSON without a holdable cursor, add HttpServletResponse response to the controller method, get the output stream and use ObjectMapper to write the stream. With FasterXML 3.x, you can call ObjectMapper().writeValue(writer, stream), but with 2.8.x you have to use the stream's iterator:

@RequestMapping(...)
@Transactional
public void getAll(HttpServletResponse response) throws IOException {
    try(final Stream<MyObject> stream = service.streamAll()) {
        final Writer writer = new BufferedWriter(new OutputStreamWriter(response.getOutputStream()));
        new ObjectMapper().writerFor(Iterator.class).writeValue(writer, stream.iterator());
    }
}

Next steps

My next steps are to attempt refactor this within a Callable WebAsyncTask and to move the JSON serialization into a service.

References

• Be sure to read Marko Topolnik's blog post https://www.airpair.com/java/posts/spring-streams-memory-efficiency, without which I wouldn't have known where to start.
• MySQL >5.0.2 now supports cursors, so you can add useCursorFetch=true to the connection string -- https://dev.mysql.com/doc/connector-j/5.1/en/connector-j-reference-configuration-properties.html
• FasterXml serialization of Java 8 Stream -- https://stackoverflow.com/a/37979665/2562746

Answer 2

You can stream entities in Spring 5.0 / WebFlux.

Take a look at this example REACTIVE Rest Controller (spring.main.web-application-type: "REACTIVE"):

@RestController
public class XService {

    class XDto{
        final int x;
        public XDto(int x) {this.x = x;}
    }

    Stream<XDto> produceX(){
        return IntStream.range(1,10).mapToObj(i -> {
            System.out.println("produce "+i);
            try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}
            return new XDto(i);
        });
    }

    // stream of Server-Sent Events (SSE)
    @GetMapping(value = "/api/x/sse", 
    produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<XDto> getXSse() {
        return Flux.fromStream(produceX());
    }

    // stream of JSON lines
    @GetMapping(value = "/api/x/json-stream", 
    produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<XDto> getAllJsonStream() {
        return Flux.fromStream(produceX());
    }

    // same as List<XDto> - blocking JSON list
    @GetMapping(value = "/api/x/json-list", 
    produces = MediaType.APPLICATION_JSON_VALUE)
    public Flux<XDto> getAll() {
        return Flux.fromStream(produceX());
    }
}

Spring Framework 5.0 - WebFlux:

Spring’s reactive stack web framework, new in 5.0, is fully reactive and non-blocking. It is suitable for event-loop style processing with a small number of threads.

Server-Sent Events (SSE):

Server-sent events is a standard describing how servers can initiate data transmission towards clients once an initial client connection has been established.

WebSockets vs. Server-Sent events/EventSource