How to do sequence of operations and ensure one operation is complete before next one in Spring Reactor web app?

Question

I have Spring Boot 2 web app in which I need to identify site visitor by cookie and gather page view stats. So I need to intercept every web request. The code I had to write is more complex than call back hell (the very problem Spring reactor was supposed to solve).

Here is the code:

package mypack.conf;  import java.time.LocalDateTime; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.UUID;  import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Lazy; import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories; import org.springframework.http.HttpCookie; import org.springframework.http.ResponseCookie; import org.springframework.web.reactive.config.ResourceHandlerRegistry; import org.springframework.web.reactive.config.WebFluxConfigurer; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilter; import org.springframework.web.server.WebFilterChain;  import mypack.dao.PageViewRepository; import mypack.dao.UserRepository; import mypack.domain.PageView; import mypack.domain.User; import mypack.security.JwtProvider;  import reactor.core.publisher.Mono; @Configuration   @ComponentScan(basePackages = "mypack") @EnableReactiveMongoRepositories(basePackages = "mypack") public class WebConfig implements WebFluxConfigurer {      @Autowired     @Lazy     private UserRepository userRepository;      @Autowired     @Lazy     private PageViewRepository pageViewRepository;       @Autowired     @Lazy     JwtProvider jwtProvider;       @Bean     public WebFilter sampleWebFilter()  {         return new WebFilter() {              @Override             public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {                  String uri = exchange.getRequest().getURI().toString();                 String path = exchange.getRequest().getPath().pathWithinApplication().value();                   HttpCookie  cookie = null;                 String token = "";                 Map<String, List<HttpCookie>> cookies = exchange.getRequest().getCookies();                   try {                     if((exchange.getRequest().getCookies().containsKey("_token") )                             &&  (exchange.getRequest().getCookies().getFirst("_token"))!=null  ) {                          cookie = exchange.getRequest().getCookies().getFirst("_token");                         token = cookie.getValue();                           return userRepository.findByToken(token).map(user -> {                                  exchange.getAttributes().put("_token", user.getToken());                                   PageView pg = PageView.builder().createdDate(LocalDateTime.now()).URL(uri).build();                                 pageViewRepository.save(pg).subscribe(pg1 -> {user.getPageviews().add(pg1); });                                  userRepository.save(user).subscribe();                                     return user;                             })                               .flatMap(user-> chain.filter(exchange)); // ultimately this step executes regardless user exist or not                      // handle case when brand new user first time visits website                         } else {                         token = jwtProvider.genToken("guest", UUID.randomUUID().toString());                         User user = User.builder().createdDate(LocalDateTime.now()).token(token).emailId("guest").build();                         userRepository.save(user).subscribe();                         exchange.getResponse().getCookies().remove("_token");                          ResponseCookie rcookie  = ResponseCookie.from("_token", token).httpOnly(true).build();                         exchange.getResponse().addCookie(rcookie);                         exchange.getAttributes().put("_token", token);                      }                  } catch (Exception e) {                      e.printStackTrace();                 }                    return chain.filter(exchange);             } // end of  Mono<Void> filter method         }; // end of New WebFilter (anonymous class)     }  } 

Other relevant classes:

@Repository public interface PageViewRepository extends   ReactiveMongoRepository<PageView, String>{      Mono<PageView> findById(String id);  }   @Repository public interface UserRepository extends   ReactiveMongoRepository<User, String>{      Mono<User> findByToken(String token);  }      @Data @AllArgsConstructor @Builder @NoArgsConstructor public class User {      @Id     private String id;     private String token;       @Default     private LocalDateTime createdDate = LocalDateTime.now();      @DBRef     private List<PageView> pageviews;  }    Data @Document @Builder public class PageView {     @Id     private String id;      private String URL;      @Default     private LocalDateTime createdDate = LocalDateTime.now(); } 

Relevant part of gradle file:

buildscript {     ext {          springBootVersion = '2.0.1.RELEASE'     }     repositories {         mavenCentral()      }     dependencies {         classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")     } }  dependencies {      compile('org.springframework.boot:spring-boot-starter-data-mongodb-reactive')     compile('org.springframework.boot:spring-boot-starter-security')     compile('org.springframework.boot:spring-boot-starter-thymeleaf')      compile('org.springframework.boot:spring-boot-starter-webflux')      compile('org.springframework.security:spring-security-oauth2-client')     compile('org.springframework.security.oauth:spring-security-oauth2:2.3.4.RELEASE')     runtime('org.springframework.boot:spring-boot-devtools')     compileOnly('org.projectlombok:lombok')     compile "org.springframework.security:spring-security-jwt:1.0.9.RELEASE"     compile "io.jsonwebtoken:jjwt:0.9.0"      testCompile('org.springframework.boot:spring-boot-starter-test')      testCompile('io.projectreactor:reactor-test')      compile('com.fasterxml.jackson.core:jackson-databind') } 

The problem is in these lines:

PageView pg = PageView.builder().createdDate(LocalDateTime.now()).URL(uri).build(); pageViewRepository.save(pg).subscribe(pg1 -> {user.getPageviews().add(pg1); });

which hangs the browser (keeps waiting for response).

Basically what I want is this: Must not use block() which does not even work in webfilter code as block also hangs the browser. Save pageview in mongo db. After it is saved, pageview has valid mongodb id which is needed to be stored as reference in pageviews List of user entity. Therefore only after it is saved in db, the next step is update user's pageviews List. Next step is save the updated user without effecting downstream controller methods which may also update user and may need to save the user too. All this should work in the given WebFilter context.

How to solve this problem?

The solution provided must make sure that user is saved in webfilter before passing on to controller actions some of which also saves user with different values from query string params.

Answer 1

If I understand you correctly, you need to perform long operations with database asynchronously to prevent the filter (and the request itself) from blocking?

In this case, I would recommend the following solution that works for me:

@Bean public WebFilter filter() {     return (exchange, chain) -> {         ServerHttpRequest req = exchange.getRequest();         String uri = req.getURI().toString();         log.info("[i] Got request: {}", uri);          var headers = req.getHeaders();         List<String> tokenList = headers.get("token");          if (tokenList != null && tokenList.get(0) != null) {             String token = tokenList.get(0);             log.info("[i] Find a user by token {}", token);             return userRepo.findByToken(token)                     .map(user -> process(exchange, uri, token, user))                     .then(chain.filter(exchange));         } else {             String token = UUID.randomUUID().toString();             log.info("[i] Create a new user with token {}", token);             return userRepo.save(new User(token))                     .map(user -> process(exchange, uri, token, user))                     .then(chain.filter(exchange));         }     }; } 

Here I slightly change your logic and take the token value from the appropriate header (not from cookies) to simplify my implementation.

So if the token is present then we try to find its user. If the token isn't present then we create a new user. If the user is found or created successfully, then the process method is calling. After that, regardless of the result, we return chain.filter(exchange).

The method process puts a token value to the appropriate attribute of the request and call asynchronously the method updateUserStat of the userService:

private User process(ServerWebExchange exchange, String uri, String token, User user) {     exchange.getAttributes().put("_token", token);     userService.updateUserStat(uri, user); // async call     return user; } 

User service:

@Slf4j @Service public class UserService {      private final UserRepo userRepo;     private final PageViewRepo pageViewRepo;      public UserService(UserRepo userRepo, PageViewRepo pageViewRepo) {         this.userRepo = userRepo;         this.pageViewRepo = pageViewRepo;     }      @SneakyThrows     @Async     public void updateUserStat(String uri, User user) {         log.info("[i] Start updating...");         Thread.sleep(1000);         pageViewRepo.save(new PageView(uri))                 .flatMap(user::addPageView)                 .blockOptional()                 .ifPresent(u -> userRepo.save(u).block());         log.info("[i] User updated.");     } } 

I've added here a small delay for test purposes to make sure that requests work without any delay, regardless of the duration of this method.

A case when the user is found by the token:

2019-01-06 18:25:15.442  INFO 4992 --- [ctor-http-nio-3] : [i] Got request: http://localhost:8080/users?test=1000 2019-01-06 18:25:15.443  INFO 4992 --- [ctor-http-nio-3] : [i] Find a user by token 84b0f7ec-670c-4c04-8a7c-b692752d7cfa 2019-01-06 18:25:15.444 DEBUG 4992 --- [ctor-http-nio-3] : Created query Query: { "token" : "84b0f7ec-670c-4c04-8a7c-b692752d7cfa" }, Fields: { }, Sort: { } 2019-01-06 18:25:15.445 DEBUG 4992 --- [ctor-http-nio-3] : find using query: { "token" : "84b0f7ec-670c-4c04-8a7c-b692752d7cfa" } fields: Document{{}} for class: class User in collection: user 2019-01-06 18:25:15.457  INFO 4992 --- [ntLoopGroup-2-2] : [i] Get all users... 2019-01-06 18:25:15.457  INFO 4992 --- [         task-3] : [i] Start updating... 2019-01-06 18:25:15.458 DEBUG 4992 --- [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user 2019-01-06 18:25:16.459 DEBUG 4992 --- [         task-3] : Inserting Document containing fields: [URL, createdDate, _class] in collection: pageView 2019-01-06 18:25:16.476 DEBUG 4992 --- [         task-3] : Saving Document containing fields: [_id, token, pageViews, _class] 2019-01-06 18:25:16.479  INFO 4992 --- [         task-3] : [i] User updated. 

Here we can see that updating the user is performed in the independent task-3 thread after the user already has a result of 'get all users' request.

A case when the token is not present and the user is created:

2019-01-06 18:33:54.764  INFO 4992 --- [ctor-http-nio-3] : [i] Got request: http://localhost:8080/users?test=763 2019-01-06 18:33:54.764  INFO 4992 --- [ctor-http-nio-3] : [i] Create a new user with token d9bd40ea-b869-49c2-940e-83f1bf79e922 2019-01-06 18:33:54.765 DEBUG 4992 --- [ctor-http-nio-3] : Inserting Document containing fields: [token, _class] in collection: user 2019-01-06 18:33:54.776  INFO 4992 --- [ntLoopGroup-2-2] : [i] Get all users... 2019-01-06 18:33:54.777  INFO 4992 --- [         task-4] : [i] Start updating... 2019-01-06 18:33:54.777 DEBUG 4992 --- [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user 2019-01-06 18:33:55.778 DEBUG 4992 --- [         task-4] : Inserting Document containing fields: [URL, createdDate, _class] in collection: pageView 2019-01-06 18:33:55.792 DEBUG 4992 --- [         task-4] : Saving Document containing fields: [_id, token, pageViews, _class] 2019-01-06 18:33:55.795  INFO 4992 --- [         task-4] : [i] User updated. 

A case when the token is present but user is not found:

2019-01-06 18:35:40.970  INFO 4992 --- [ctor-http-nio-3] : [i] Got request: http://localhost:8080/users?test=150 2019-01-06 18:35:40.970  INFO 4992 --- [ctor-http-nio-3] : [i] Find a user by token 184b0f7ec-670c-4c04-8a7c-b692752d7cfa 2019-01-06 18:35:40.972 DEBUG 4992 --- [ctor-http-nio-3] : Created query Query: { "token" : "184b0f7ec-670c-4c04-8a7c-b692752d7cfa" }, Fields: { }, Sort: { } 2019-01-06 18:35:40.972 DEBUG 4992 --- [ctor-http-nio-3] : find using query: { "token" : "184b0f7ec-670c-4c04-8a7c-b692752d7cfa" } fields: Document{{}} for class: class User in collection: user 2019-01-06 18:35:40.977  INFO 4992 --- [ntLoopGroup-2-2] : [i] Get all users... 2019-01-06 18:35:40.978 DEBUG 4992 --- [ntLoopGroup-2-2] : find using query: { } fields: Document{{}} for class: class User in collection: user 

My demo project: sb-reactive-filter-demo