code: @SuppressWarnings("unchecked")
    private void handlePublish(PublishRequest request) {
      state.getLogger().trace("{} - Received {}", state.getSessionId(), request);
  
      // If the request is for another session ID, this may be a session that was previously opened
      // for this client.
      if (request.session() != state.getSessionId()) {
        state.getLogger().trace("{} - Inconsistent session ID: {}", state.getSessionId(), request.session());
        return;
      }
  
      // If the request event index has already been processed, return.
      if (request.eventIndex() <= state.getEventIndex()) {
        return;
      }
  
      // If the request's previous event index doesn't equal the previous received event index,
      // respond with an undefined error and the last index received. This will cause the cluster
      // to resend events starting at eventIndex + 1.
      if (request.previousIndex() != state.getEventIndex()) {
        state.getLogger().trace("{} - Inconsistent event index: {}", state.getSessionId(), request.previousIndex());
        connection.send(ResetRequest.builder()
          .withSession(state.getSessionId())
          .withIndex(state.getEventIndex())
          .build());
        return;
      }
  
      // Store the event index. This will be used to verify that events are received in sequential order.
      state.setEventIndex(request.eventIndex());
  
      sequencer.sequenceEvent(request, () -> {
        for (Event<?> event : request.events()) {
          Set<Consumer> listeners = eventListeners.get(event.name());
          if (listeners != null) {
            for (Consumer listener : listeners) {
              listener.accept(event.message());
            }
          }
        }
      });
    }
docs: Handles a publish request.
  
  @param request The publish request to handle.
  @return A completable future to be completed with the publish response.