class GeminiStreamingMarketDataService implements StreamingMarketDataService {
  private final GeminiStreamingService service;
  private final Map<CurrencyPair, GeminiOrderbook> orderbooks = new HashMap<>();

  private final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();

  public GeminiStreamingMarketDataService(GeminiStreamingService service) {
    this.service = service;
  }

  private boolean filterEventsByReason(JsonNode message, String type, String reason) {<FILL_FUNCTION_BODY>}

  @Override
  public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {

    int maxDepth = (int) MoreObjects.firstNonNull(args.length > 0 ? args[0] : null, 1);

    Observable<GeminiOrderbook> subscribedOrderbookSnapshot =
        service
            .subscribeChannel(currencyPair, maxDepth, maxDepth)
            .filter(
                s ->
                    filterEventsByReason(s, "change", "initial")
                        || filterEventsByReason(s, "change", "place")
                        || filterEventsByReason(s, "change", "cancel")
                        || filterEventsByReason(s, "change", "trade"))
            .filter(
                s -> // filter out updates that arrive before initial book
                orderbooks.get(currencyPair) != null
                        || filterEventsByReason(s, "change", "initial"))
            .map(
                (JsonNode s) -> {
                  if (filterEventsByReason(s, "change", "initial")) {
                    GeminiWebSocketTransaction transaction =
                        mapper.treeToValue(s, GeminiWebSocketTransaction.class);
                    GeminiOrderbook orderbook = transaction.toGeminiOrderbook(currencyPair);
                    orderbooks.put(currencyPair, orderbook);
                    return orderbook;
                  }

                  if (filterEventsByReason(s, "change", "place")
                      || filterEventsByReason(s, "change", "cancel")
                      || filterEventsByReason(s, "change", "trade")) {

                    GeminiWebSocketTransaction transaction =
                        mapper.treeToValue(s, GeminiWebSocketTransaction.class);
                    GeminiLimitOrder[] levels = transaction.toGeminiLimitOrdersUpdate();
                    GeminiOrderbook orderbook = orderbooks.get(currencyPair);
                    orderbook.updateLevels(levels);
                    return orderbook;
                  }

                  throw new NotYetImplementedForExchangeException(
                      " Unknown message type, even after filtering: " + s.toString());
                });

    return subscribedOrderbookSnapshot.map(
        geminiOrderbook -> GeminiAdaptersX.toOrderbook(geminiOrderbook, maxDepth, new Date()));
  }

  @Override
  public Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args) {
    return PublishSubject.create(
        emitter ->
            getOrderBook(currencyPair, args)
                .subscribe(
                    orderBook -> {
                      LimitOrder firstBid = orderBook.getBids().iterator().next();
                      LimitOrder firstAsk = orderBook.getAsks().iterator().next();
                      emitter.onNext(
                          new Ticker.Builder()
                              .instrument(currencyPair)
                              .bid(firstBid.getLimitPrice())
                              .bidSize(firstBid.getOriginalAmount())
                              .ask(firstAsk.getLimitPrice())
                              .askSize(firstAsk.getOriginalAmount())
                              .timestamp(
                                  firstBid.getTimestamp().after(firstAsk.getTimestamp())
                                      ? firstBid.getTimestamp()
                                      : firstAsk.getTimestamp())
                              .build());
                    },
                    emitter::onError));
  }

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    Observable<GeminiTrade[]> subscribedTrades =
        service
            .subscribeChannel(currencyPair, args)
            .filter(s -> filterEventsByReason(s, "trade", null))
            .map(
                (JsonNode s) -> {
                  GeminiWebSocketTransaction transaction =
                      mapper.treeToValue(s, GeminiWebSocketTransaction.class);
                  return transaction.toGeminiTrades();
                });

    return subscribedTrades.flatMapIterable(s -> adaptTrades(s, currencyPair).getTrades());
  }
}

    boolean hasEvents = false;
    if (message.has("events")) {
      for (JsonNode jsonEvent : message.get("events")) {
        boolean reasonResult =
            reason == null
                || (jsonEvent.has("reason") && jsonEvent.get("reason").asText().equals(reason));
        if (jsonEvent.get("type").asText().equals(type) && reasonResult) {
          hasEvents = true;
          break;
        }
      }
    }

    return hasEvents;
  
<no_super_class>