OkCoinStreamingService extends JsonNettyStreamingService { @Override protected String getChannelNameFromMessage(JsonNode message) throws IOException { return message.get("channel").asText(); } OkCoinStreamingService(String apiUrl); @Override Completable connect(); @Override String getSubscribeMessage(String channelName, Object... args); @Override String getUnsubscribeMessage(String channelName);  }
@Test public void testGetChannelFromMessage() throws Exception { String expected = new String( Files.readAllBytes( Paths.get(ClassLoader.getSystemResource("order-book.json").toURI()))); JsonNode data = new ObjectMapper().readTree(expected); String channel = streamingService.getChannelNameFromMessage(data); assertThat(channel).isEqualTo("ok_sub_spot_btc_usd_depth"); }