class CoinjarStreamingMarketDataService implements StreamingMarketDataService {

  private static final Logger logger =
      LoggerFactory.getLogger(CoinjarStreamingMarketDataService.class);

  private final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();

  private final CoinjarStreamingService service;

  public CoinjarStreamingMarketDataService(CoinjarStreamingService service) {
    this.service = service;
  }

  private static void updateOrderbook(Map<BigDecimal, LimitOrder> book, List<LimitOrder> orders) {
    orders.forEach(
        order -> {
          if (order.getOriginalAmount().compareTo(BigDecimal.ZERO) > 0) {
            book.put(order.getLimitPrice(), order);
          } else {
            book.remove(order.getLimitPrice());
          }
        });
  }

  private static OrderBook handleOrderbookEvent(
      CoinjarWebSocketBookEvent event,
      Map<BigDecimal, LimitOrder> bids,
      Map<BigDecimal, LimitOrder> asks) {<FILL_FUNCTION_BODY>}

  @Override
  public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {
    final SortedMap<BigDecimal, LimitOrder> bids =
        Maps.newTreeMap((o1, o2) -> Math.negateExact(o1.compareTo(o2)));
    final SortedMap<BigDecimal, LimitOrder> asks = Maps.newTreeMap(BigDecimal::compareTo);
    String channelName = CoinjarStreamingAdapters.adaptCurrencyPairToBookTopic(currencyPair);
    return service
        .subscribeChannel(channelName)
        .doOnError(
            throwable -> {
              logger.warn(
                  "encoutered error while subscribing to channel " + channelName, throwable);
            })
        .map(
            node -> {
              CoinjarWebSocketBookEvent orderEvent =
                  mapper.treeToValue(node, CoinjarWebSocketBookEvent.class);
              return handleOrderbookEvent(orderEvent, bids, asks);
            })
        .filter(orderbook -> !orderbook.getBids().isEmpty() && !orderbook.getAsks().isEmpty());
  }

  @Override
  public Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args) {
    throw new NotYetImplementedForExchangeException();
  }

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    throw new NotYetImplementedForExchangeException();
  }
}

    final CurrencyPair pairFromEvent =
        CoinjarStreamingAdapters.adaptTopicToCurrencyPair(event.topic);
    switch (event.event) {
      case CoinjarWebSocketBookEvent.UPDATE:
      case CoinjarWebSocketBookEvent.INIT:
        updateOrderbook(
            bids,
            CoinjarStreamingAdapters.toLimitOrders(
                event.payload.bids, pairFromEvent, Order.OrderType.BID));
        updateOrderbook(
            asks,
            CoinjarStreamingAdapters.toLimitOrders(
                event.payload.asks, pairFromEvent, Order.OrderType.ASK));
        break;
    }
    return new OrderBook(
        null, Lists.newArrayList(asks.values()), Lists.newArrayList(bids.values()));
  
<no_super_class>