SimplifiedSwitchmanCore.java

inputStream.groupBy(Event::getItemId)
    .subscribe(substream -> 
        substream.onBackpressureLatest()
               .subscribe(new Subscriber<Event>() {
                   public void onNext(Event event) {
                      if (shouldSkipEvent(event)) {
                          outputStream.onNext(Response.createSkippedResponse());
                          request(1);
                      } else {
                          saveLastLaunchedEvent(event);
                          launchNetworkCall(event).doOnSuccess(outputStream::onNext)
                                        .subscribe(ignored -> request(1));
                      }
                  }
              }
          }));