class KrakenFuturesStreamingMarketDataService implements StreamingMarketDataService {

  private final ObjectMapper objectMapper = StreamingObjectMapperHelper.getObjectMapper();
  private final KrakenFuturesStreamingService service;
  private final Map<Instrument, OrderBook> orderBookMap = new HashMap<>();

  public KrakenFuturesStreamingMarketDataService(KrakenFuturesStreamingService service) {
    this.service = service;
  }

  @Override
  public Observable<OrderBook> getOrderBook(Instrument instrument, Object... args) {
    String channelName =
        service.ORDERBOOK + KrakenFuturesAdapters.adaptKrakenFuturesSymbol(instrument);
    return service
        .subscribeChannel(channelName)
        .filter(message -> message.has("feed"))
        .map(
            message -> {
              try {
                if (message.get("feed").asText().contains("book_snapshot")) {
                  orderBookMap.put(
                      instrument,
                      KrakenFuturesStreamingAdapters.adaptKrakenFuturesSnapshot(
                          objectMapper.treeToValue(
                              message, KrakenFuturesStreamingOrderBookSnapshotResponse.class)));
                } else if (message.get("feed").asText().equals(service.ORDERBOOK)) {
                  KrakenFuturesStreamingOrderBookDeltaResponse delta =
                      objectMapper.treeToValue(
                          message, KrakenFuturesStreamingOrderBookDeltaResponse.class);
                  orderBookMap
                      .get(instrument)
                      .update(
                          new LimitOrder.Builder(
                                  (delta
                                          .getSide()
                                          .equals(
                                              KrakenFuturesStreamingOrderBookDeltaResponse
                                                  .KrakenFuturesStreamingSide.sell))
                                      ? Order.OrderType.ASK
                                      : Order.OrderType.BID,
                                  instrument)
                              .limitPrice(delta.getPrice())
                              .originalAmount(delta.getQty())
                              .timestamp(delta.getTimestamp())
                              .build());
                }
                if (orderBookMap
                        .get(instrument)
                        .getBids()
                        .get(0)
                        .getLimitPrice()
                        .compareTo(orderBookMap.get(instrument).getAsks().get(0).getLimitPrice())
                    > 0) {
                  throw new IOException("OrderBook crossed!!!");
                }
                return orderBookMap.get(instrument);
              } catch (Exception e) {
                throw new IOException(e);
              }
            });
  }

  @Override
  public Observable<Ticker> getTicker(Instrument instrument, Object... args) {
    String channelName =
        service.TICKER + KrakenFuturesAdapters.adaptKrakenFuturesSymbol(instrument);

    return service
        .subscribeChannel(channelName)
        .filter(message -> message.has("feed") && message.has("product_id"))
        .filter(
            message ->
                message
                    .get("product_id")
                    .asText()
                    .equals(KrakenFuturesAdapters.adaptKrakenFuturesSymbol(instrument)))
        .map(
            message ->
                KrakenFuturesStreamingAdapters.adaptTicker(
                    objectMapper.treeToValue(message, KrakenFuturesStreamingTickerResponse.class)));
  }

  @Override
  public Observable<Trade> getTrades(Instrument instrument, Object... args) {
    String channelName =
        service.TRADES + KrakenFuturesAdapters.adaptKrakenFuturesSymbol(instrument);

    return service
        .subscribeChannel(channelName)
        .filter(message -> message.has("feed") && message.has("product_id"))
        .filter(message -> message.get("feed").asText().equals("trade"))
        .map(
            message ->
                KrakenFuturesStreamingAdapters.adaptTrade(
                    objectMapper.treeToValue(message, KrakenFuturesStreamingTradeResponse.class)));
  }

  @Override
  public Observable<FundingRate> getFundingRate(Instrument instrument, Object... args) {<FILL_FUNCTION_BODY>}
}

    String channelName =
        service.TICKER + KrakenFuturesAdapters.adaptKrakenFuturesSymbol(instrument);

    return service
        .subscribeChannel(channelName)
        .filter(message -> message.has("feed") && message.has("product_id"))
        .filter(
            message ->
                message
                    .get("product_id")
                    .asText()
                    .equals(KrakenFuturesAdapters.adaptKrakenFuturesSymbol(instrument)))
        .map(
            message ->
                KrakenFuturesStreamingAdapters.adaptFundingRate(
                    objectMapper.treeToValue(message, KrakenFuturesStreamingTickerResponse.class)));
  
<no_super_class>