title: Spring WebFlux with MongoDB - throttling SSE clients
body: <p>I am working on a simple chat service ran by Spring Boot 2.1.1 with WebFlux, Reactor 3.2.3, Mongo 3.8.2 and Netty 4.1.31.</p>
  
  <p>Each chat room has 2 collections - messages archive and a capped collection with current events (eg. new message event, user typing indicators etc.). The capped collection has 100 elements and I am using tail() method of ReactiveMongoTemplate to retrieve latest events.</p>
  
  <p>The service exposes 2 kinds of endpoints for retrieving the recent events: SSE and for polling. I have done some stress testing with 2000 concurrent users which apart from listening to the chat, were spamming tons of events.</p>
  
  <p><strong>The observations are:</strong></p>
  
  <ul>
  <li>polling every 2 seconds brings a bit of stress to the service (~40% CPU usage during the test) and almost no stress to the MongoDB (~4%)</li>
  <li>listening via SSE maxes out the MongoDB (~90%), also stresses the service (which tries to use the rest of available resources), but Mongo is particularly struggling and overall the service becomes almost unresponsive.</li>
  </ul>
  
  <p>The observation seems obvious, because when I have connected via SSE during the test, it has updated me almost instantly when new event arrived - basically SSE was hundreds of times more responsive than polling every 2 seconds.</p>
  
  <p><strong>The question is:</strong></p>
  
  <p>Given that the client is ultimately the subscriber (or at least I think it is given by limited knowledge), can I somehow throttle the rate of publishing messages by ReactiveMongoTemplate? Or somehow decrease the demand for new events without having to do that client-side?</p>
  
  <p>I have been trying my luck with Flux buffering and cache'ing, however it caused even more stress...</p>
  
  <p><strong>Code:</strong></p>
  
  <pre><code>// ChatRepository.java
  
  private static final Query chatEventsQuery = new Query();
  
  public Flux&lt;ChatEvent&gt; getChatEventsStream(String chatId) {
      return reactiveMongoTemplate.tail(
              chatEventsQuery,
              ChatEvent.class,
              chatId
      );
  }
  </code></pre>
  
  <p>,</p>
  
  <pre><code>// ChatHandler.java
  
  public Mono&lt;ServerResponse&gt; getChatStream(ServerRequest request) {
  
      String chatId = request.pathVariable(CHAT_ID_PATH_VARIABLE);
      String username = getUsername(request);
  
      Flux&lt;ServerSentEvent&gt; chatEventsStream = chatRepository
              .getChatEventsStream(chatId)
              .map(addUserSpecificPropsToChatEvent(username))
              .map(event -&gt; ServerSentEvent.&lt;ChatEvent&gt;builder()
                      .event(event.getType().getEventName())
                      .data(event)
                      .build());
  
      log.debug("\nExposing chat stream\nchat: {}\nuser: {}", chatId, username);
  
      return ServerResponse.ok().body(
              chatEventsStream,
              ServerSentEvent.class
      );
  }
  </code></pre>
  
  <p>,</p>
  
  <pre><code>// ChatRouter.java
  
  RouterFunction&lt;ServerResponse&gt; routes(ChatHandler handler) {
      return route(GET("/api/chat/{chatId}/stream"), handler::getChatStream);
  }
  </code></pre>