class GeminiStreamingService extends JsonNettyStreamingService {
  private static final Logger LOG = LoggerFactory.getLogger(GeminiStreamingService.class);
  private static final String SHARE_CHANNEL_NAME = "ALL";
  private static final String SUBSCRIBE = "subscribe";
  private static final String UNSUBSCRIBE = "unsubscribe";

  private final Map<String, Observable<JsonNode>> subscriptions = new ConcurrentHashMap<>();
  private ProductSubscription product = null;

  public GeminiStreamingService(String baseUri) {
    super(baseUri, Integer.MAX_VALUE);
  }

  public ProductSubscription getProduct() {
    return this.product;
  }

  public Observable<GeminiWebSocketTransaction> getRawWebSocketTransactions(
      CurrencyPair currencyPair, boolean filterChannelName) {
    String channelName = currencyPair.base.toString() + currencyPair.counter.toString();
    final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
    return subscribeChannel(channelName)
        .map(s -> mapper.treeToValue(s, GeminiWebSocketTransaction.class))
        .filter(t -> channelName.equals(t.getSymbol()))
        .filter(t -> !StringUtil.isNullOrEmpty(t.getType()));
  }

  public void subscribeMultipleCurrencyPairs(ProductSubscription... products) {
    this.product = products[0];
  }

  @Override
  public Observable<JsonNode> subscribeChannel(String channelName, Object... args) {<FILL_FUNCTION_BODY>}

  @Override
  public boolean processArrayMessageSeparately() {
    return false;
  }

  @Override
  protected String getChannelNameFromMessage(JsonNode message) throws IOException {
    return SHARE_CHANNEL_NAME;
  }

  @Override
  public String getSubscribeMessage(String channelName, Object... args) throws IOException {
    return objectMapper.writeValueAsString(
        new GeminiWebSocketSubscriptionMessage(SUBSCRIBE, product));
  }

  @Override
  public String getUnsubscribeMessage(String channelName, Object... args) throws IOException {
    return objectMapper.writeValueAsString(
        new GeminiWebSocketSubscriptionMessage(UNSUBSCRIBE, product));
  }
}

    channelName = SHARE_CHANNEL_NAME;

    if (!channels.containsKey(channelName) && !subscriptions.containsKey(channelName)) {
      subscriptions.put(channelName, super.subscribeChannel(channelName, args));
    }

    return subscriptions.get(channelName);
  
<methods>public void <init>(java.lang.String) ,public void <init>(java.lang.String, int) ,public void <init>(java.lang.String, int, java.time.Duration, java.time.Duration, int) ,public void messageHandler(java.lang.String) ,public boolean processArrayMessageSeparately() <variables>private static final Logger LOG,protected final ObjectMapper objectMapper