class CoinmateStreamingMarketDataService implements StreamingMarketDataService {
  private final CoinmateStreamingService coinmateStreamingService;

  CoinmateStreamingMarketDataService(CoinmateStreamingService coinmateStreamingService) {
    this.coinmateStreamingService = coinmateStreamingService;
  }

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

  @Override
  public Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args) {
    String channelName = "statistics-" + CoinmateStreamingAdapter.getChannelPostfix(currencyPair);

    ObjectReader reader =
        StreamingObjectMapperHelper.getObjectMapper().readerFor(CoinmateTradeStatistics.class);
    return coinmateStreamingService
        .subscribeChannel(channelName)
        .map(
            s -> {
              CoinmateTradeStatistics tradeStatistics = reader.readValue(s.get("payload"));
              return CoinmateAdapters.adaptTradeStatistics(tradeStatistics, currencyPair);
            });
  }

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    String channelName = "trades-" + CoinmateStreamingAdapter.getChannelPostfix(currencyPair);

    ObjectReader reader =
        StreamingObjectMapperHelper.getObjectMapper()
            .readerFor(new TypeReference<List<CoinmateWebSocketTrade>>() {});

    return coinmateStreamingService
        .subscribeChannel(channelName)
        .map(s -> reader.<List<CoinmateWebSocketTrade>>readValue(s.get("payload")))
        .flatMapIterable(coinmateWebSocketTrades -> coinmateWebSocketTrades)
        .map(
            coinmateWebSocketTrade ->
                CoinmateStreamingAdapter.adaptTrade(coinmateWebSocketTrade, currencyPair));
  }
}

    String channelName = "order_book-" + CoinmateStreamingAdapter.getChannelPostfix(currencyPair);

    ObjectReader reader =
        StreamingObjectMapperHelper.getObjectMapper().readerFor(CoinmateOrderBookData.class);

    return coinmateStreamingService
        .subscribeChannel(channelName)
        .map(
            s -> {
              CoinmateOrderBookData orderBookData = reader.readValue(s.get("payload"));
              CoinmateOrderBook coinmateOrderBook =
                  new CoinmateOrderBook(false, null, orderBookData);

              return CoinmateAdapters.adaptOrderBook(coinmateOrderBook, currencyPair);
            });
  
<no_super_class>