class BTCMarketsStreamingMarketDataService implements StreamingMarketDataService {

  private final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();

  private final BTCMarketsStreamingService service;

  public BTCMarketsStreamingMarketDataService(BTCMarketsStreamingService service) {
    this.service = service;
  }

  private OrderBook handleOrderbookMessage(BTCMarketsWebSocketOrderbookMessage message)
      throws InvalidFormatException {
    return BTCMarketsStreamingAdapters.adaptOrderbookMessageToOrderbook(message);
  }

  private Ticker handleTickerMessage(BTCMarketsWebSocketTickerMessage message)
      throws InvalidFormatException {
    return BTCMarketsStreamingAdapters.adaptTickerMessageToTicker(message);
  }

  @Override
  public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {
    final String marketId = BTCMarketsStreamingAdapters.adaptCurrencyPairToMarketId(currencyPair);
    return service
        .subscribeChannel(CHANNEL_ORDERBOOK, marketId)
        .map(node -> mapper.treeToValue(node, BTCMarketsWebSocketOrderbookMessage.class))
        .filter(orderEvent -> marketId.equals(orderEvent.marketId))
        .map(this::handleOrderbookMessage);
  }

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

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    final String marketId = BTCMarketsStreamingAdapters.adaptCurrencyPairToMarketId(currencyPair);
    return service
        .subscribeChannel(CHANNEL_TRADE, marketId)
        .map(node -> mapper.treeToValue(node, BTCMarketsWebSocketTradeMessage.class))
        .filter(event -> marketId.equals(event.getMarketId()))
        .map(this::handleTradeMessage);
  }

  private Trade handleTradeMessage(BTCMarketsWebSocketTradeMessage message)
      throws InvalidFormatException {
    return BTCMarketsStreamingAdapters.adaptTradeMessageToTrade(message);
  }
}

    final String marketId = BTCMarketsStreamingAdapters.adaptCurrencyPairToMarketId(currencyPair);
    return service
        .subscribeChannel(CHANNEL_TICKER, marketId)
        .map(node -> mapper.treeToValue(node, BTCMarketsWebSocketTickerMessage.class))
        .filter(tickerEvent -> marketId.equals(tickerEvent.getMarketId()))
        .map(this::handleTickerMessage);
  
<no_super_class>