Spring Boot Async Controller with ResponseBodyEmitter

Learn to write spring boot async rest controller using ResponseBodyEmitter. We can use this approach when we have a service, or multiple calls, and want to collect the results and send the response to the client.

ResponseBodyEmitter helps to collect and send the response to the client. It is a controller method return value type for asynchronous request processing where one or more objects are written to the response.

While DeferredResult is used to produce a single result, a ResponseBodyEmitter can be used to send multiple objects where each object is written with a compatible HttpMessageConverter.

1. How to use ResponseBodyEmitter

To use it, create controller method like this:

@RequestMapping(value="/resource-uri", method=RequestMethod.GET)
public ResponseBodyEmitter handle() 
{
     ResponseBodyEmitter emitter = new ResponseBodyEmitter();
     // Pass the emitter to another component...
     return emitter;
}

// in another thread
 emitter.send(foo1);

 // and again
 emitter.send(foo2);

 // and done
 emitter.complete();

2. Async controller example using ResponseBodyEmitter

In given controller method, we are accessing the data sets (use your own domain datatypes).

• There is data service which return data sets from DB or any other source.
• Each data set is then processed (e.g. retrieve related information from other source) which takes time. This is simulated using an artificial delay by calling thread.sleep() method.
• Each data set is then added to ResponseBodyEmitter object using emitter.send() method.
• Finally emitter.complete() is called to mark that request processing is complete so that the thread responsible for sending the response can complete the request and be freed up for the next response to handle.
• If any error is encountered while request processing, complete the process by emitter.completeWithError(). The exception will pass through the normal exception handling of Spring MVC and after that the response is completed.

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;

import com.howtodoinjava.springasyncexample.web.model.DataSet;
import com.howtodoinjava.springasyncexample.web.service.DataSetService;

@RestController
public class DataSetController {

      private final DataSetService dataSetService;

      public DataSetController(DataSetService dataSetService) {
            this.dataSetService = dataSetService;
      }

      @GetMapping("/fetch-data-sets")
      public ResponseBodyEmitter fetchData() 
      {
            ResponseBodyEmitter emitter = new ResponseBodyEmitter();

            ExecutorService executor = Executors.newSingleThreadExecutor();

            executor.execute(() -> {
                  List<DataSet> dataSets = dataSetService.findAll();
                  try 
                  {
                        for (DataSet dataSet : dataSets) 
                        {
                              randomDelay();
                              emitter.send(dataSet);
                        }
                        emitter.complete();
                  } 
                  catch (IOException e) 
                  {
                        emitter.completeWithError(e);
                  }
            });
            executor.shutdown();
            return emitter;
      }

      private void randomDelay() {
            try {
                  Thread.sleep(1000);
            } catch (InterruptedException e) {
                  Thread.currentThread().interrupt();
            }
      }
}

3. How to test async controller using ResponseBodyEmitter

3.1. Mock testing with JUnit

To test the above controller method, I am using mockito shipped with spring boot distribution.

import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;

import java.math.BigInteger;
import java.util.Arrays;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.MvcResult;
import org.springframework.test.web.servlet.result.MockMvcResultHandlers;

import com.howtodoinjava.springasyncexample.web.controller.DataSetController;
import com.howtodoinjava.springasyncexample.web.model.DataSet;
import com.howtodoinjava.springasyncexample.web.service.DataSetService;

@RunWith(SpringRunner.class)
@WebMvcTest(DataSetController.class)
public class DataSetControllerTest 
{
      @Autowired
      private MockMvc mockMvc;
      
      @MockBean
      private DataSetService dataSetService;

      @Test
      public void foo() throws Exception 
      {
            Mockito.when(dataSetService.findAll())
                        .thenReturn(Arrays.asList(new DataSet(BigInteger.valueOf(1), "data")));
            
            MvcResult mvcResult = mockMvc.perform(get("/fetch-data-sets"))
                                                            .andExpect(request().asyncStarted())
                                                            .andDo(MockMvcResultHandlers.log())
                                                            .andReturn();
            
            mockMvc.perform(asyncDispatch(mvcResult))
                        .andDo(MockMvcResultHandlers.log())
                        .andExpect(status().isOk())
                        .andExpect(content().json("{\"id\":1,\"name\":\"data\"}"));
      }
}

Program output.

MockHttpServletResponse:
           Status = 200
    Error message = null
          Headers = {}
     Content type = null
             Body = {"id":1,"name":"data"}
    Forwarded URL = null
   Redirected URL = null
          Cookies = []

3.2. Browser testing

To test in browser, start the application using class SpringAsyncExampleApplication and hit the URL in browser: http://localhost:8080/fetch-data-sets

Check the response returned from server after some delay.

4. Async configuration options

To override the default async behavior such as thread pool and timeout, you can implement the WebMvcConfigurer interface and override it’s configureAsyncSupport() method.

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.servlet.config.annotation.AsyncSupportConfigurer;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

@SpringBootApplication
public class SpringAsyncExampleApplication implements WebMvcConfigurer {

      public static void main(String[] args) {
            SpringApplication.run(SpringAsyncExampleApplication.class, args);
      }

      @Override
      public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
            configurer.setTaskExecutor(mvcTaskExecutor());
            configurer.setDefaultTimeout(30_000);
      }

      @Bean
      public ThreadPoolTaskExecutor mvcTaskExecutor() {
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setThreadNamePrefix("mvc-task-");
            return taskExecutor;
      }
}

5. Sourcecode files

5.1. DataSetService.java

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import javax.annotation.PostConstruct;

import org.springframework.stereotype.Service;

import com.howtodoinjava.springasyncexample.web.model.DataSet;

@Service
public class DataSetService 
{
      private final List<DataSet> datasetList = new ArrayList<>();

      @PostConstruct
      public void setup() {
            createDataSets();
      }

      public List<DataSet> findAll() {
            return Collections.unmodifiableList(datasetList);
      }

      private Iterable<DataSet> createDataSets() 
      {
            String name = "dummy text_";
            
            for (int i = 0; i < 5; i++) {
                  this.datasetList.add( new DataSet(BigInteger.valueOf(i), name + i) );
            }
            return datasetList;
      }
}

5.2. DataSet.java

import java.math.BigInteger;

public class DataSet 
{
      private BigInteger id;
      private String name;
      
      public DataSet(BigInteger id, String name) {
            this.id = id;
            this.name = name;
      }

      //Getters and setters

      @Override
      public String toString() {
            return "DataSet [id=" + id + ", name=" + name + "]";
      }
}

5.3. application.properties

Enable debug logging here to understand the behavior of the application.

logging.level.org.springframework=DEBUG
logging.level.com.howtodoinjava=DEBUG

5.4. pom.xml

The used pom.xml is:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
      
      <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.2.RELEASE</version>
            <relativePath/> <!-- lookup parent from repository -->
      </parent>
      
      <groupId>com.howtodoinjava.demo</groupId>
      <artifactId>spring-async-demo</artifactId>
      <version>0.0.1-SNAPSHOT</version>
      <name>spring-async-demo</name>
      <description>Demo project for Spring Boot</description>

      <properties>
            <java.version>1.8</java.version>
      </properties>

      <dependencies>
      
            <dependency>
                  <groupId>org.springframework.boot</groupId>
                  <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            
            <dependency>
                  <groupId>org.springframework.boot</groupId>
                  <artifactId>spring-boot-starter-webflux</artifactId>
            </dependency>

            <dependency>
                  <groupId>org.springframework.boot</groupId>
                  <artifactId>spring-boot-starter-test</artifactId>
                  <scope>test</scope>
            </dependency>
            <dependency>
                  <groupId>io.projectreactor</groupId>
                  <artifactId>reactor-test</artifactId>
                  <scope>test</scope>
            </dependency>
      </dependencies>

      <build>
            <plugins>
                  <plugin>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-maven-plugin</artifactId>
                  </plugin>
            </plugins>
      </build>
      
      <repositories>
        <repository>
            <id>repository.spring.release</id>
            <name>Spring GA Repository</name>
            <url>http://repo.spring.io/release</url>
        </repository>
    </repositories>

</project>

Let me know if you face any error while executing this async rest controller example using ResponseBodyEmitter.

Happy Learning !!