Issue with [xchange-stream-gateio] Module

[Lxk-Kallen]

The WebSocket connection and message communication are working fine, but an issue arises when handling the data. The subscription ID and the parameters being passed in the handleChannelMessage method don't match.

1. Log Information: The subscription ID in the log shows:

   subscriptionUniqueId=spot.order_book-ETH/USDT_ETH/USDT-5-PT1S

   However, in the handleChannelMessage method, the parameter being passed is:

   spot.order_book-ETH/USDT

   The parameters do not align, which may cause issues in processing the data.

   Log Snippet:

   12:21:31.246 [main] INFO info.bitrich.xchangestream.gateio.GateioStreamingService - Subscribing to subscriptionUniqueId=spot.order_book-ETH/USDT_ETH/USDT-5-PT1S, args=[ETH/USDT, 5, PT1S]
   12:21:31.270 [main] DEBUG info.bitrich.xchangestream.gateio.GateioStreamingService - Sending message: {"time":1730780491,"id":2680389527639564518,"channel":"spot.order_book","event":"subscribe","payload":["ETH_USDT","5","1000ms"]}
   12:21:31.333 [nioEventLoopGroup-2-1] DEBUG info.bitrich.xchangestream.gateio.GateioStreamingService - Received message: {"time":1730780491,"time_ms":1730780491457,"id":2680389527639564518,"conn_id":"c1e37870ccc34d2c","trace_id":"67eca671a09429bb79decb93271e33c3","channel":"spot.order_book","event":"subscribe","payload":["ETH_USDT","5","1000ms"],"result":{"status":"success"},"requestId":"67eca671a09429bb79decb93271e33c3"}
   12:21:31.828 [nioEventLoopGroup-2-1] DEBUG info.bitrich.xchangestream.gateio.GateioStreamingService - Received message: {"time":1730780491,"time_ms":1730780491885,"channel":"spot.order_book","event":"update","result":{"t":1730780491870,"lastUpdateId":13401343538,"s":"ETH_USDT","bids":[["2429.7","30.3249"],["2429.67","2.6356"],["2429.66","2.5064"],["2429.63","2.0925"],["2429.6","0.06"]],"asks":[["2429.71","25.5988"],["2429.74","1.6891"],["2429.76","5.4382"],["2429.8","0.1317"],["2429.83","0.19"]]}}
   12:21:31.858 [nioEventLoopGroup-2-1] DEBUG info.bitrich.xchangestream.gateio.GateioStreamingService - Channel has been closed spot.order_book-ETH/USDT.

2. Code: The following is the code where the subscription and message handling are taking place:

   public static void main(String[] args) {
      GateioStreamingExchange exchange = 
              (GateioStreamingExchange) StreamingExchangeFactory.INSTANCE.createExchange(GateioStreamingExchange.class);
      exchange.connect().blockingAwait();
      Duration defaultUpdateSpeed = Duration.ofSeconds(1);
      Observable<OrderBook> observable = exchange
              .getStreamingMarketDataService()
              .getOrderBook(CurrencyPair.ETH_USDT, 5, defaultUpdateSpeed);
   
      TestObserver<OrderBook> testObserver = observable.test();
      OrderBook orderBook = testObserver.awaitCount(1).values().get(0);
      System.out.println(orderBook);
   
      testObserver.dispose();
   }

3. Relevant Code Snippet: The handleChannelMessage method is where the message is processed, but it doesn’t seem to handle the subscription parameters correctly:

   protected void handleChannelMessage(String channel, T message) {
      if (channel == null) {
          LOG.debug("Channel provided is null");
          return;
      }
      NettyStreamingService<T>.Subscription subscription = channels.get(channel);
      if (subscription == null) {
          LOG.debug("Channel has been closed {}.", channel);
          return;
      }
      ObservableEmitter<T> emitter = subscription.emitter;
      if (emitter == null) {
          LOG.debug("No subscriber for channel {}.", channel);
          return;
      }
      emitter.onNext(message);
   }

Expected Behavior: The subscription ID and the parameters passed to handleChannelMessage should match to ensure that the message is processed correctly.

[Lxk-Kallen]

@rizer1980 @timmolter

Thank you very much for your help! 🙏

[bigscoop]

hi @Lxk-Kallen

thanks for logs and detailed description of an error. I'll have a look today during the day