Package info.bitrich.xchangestream.binance

Class BinanceStreamingService

java.lang.Object

info.bitrich.xchangestream.service.ConnectableService

info.bitrich.xchangestream.service.netty.NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

info.bitrich.xchangestream.service.netty.JsonNettyStreamingService

info.bitrich.xchangestream.binance.BinanceStreamingService

public class BinanceStreamingService
extends JsonNettyStreamingService

Nested Class Summary

Nested classes/interfaces inherited from class info.bitrich.xchangestream.service.netty.NettyStreamingService

NettyStreamingService.NettyWebSocketClientHandler, NettyStreamingService.Subscription

Field Summary

Fields inherited from class info.bitrich.xchangestream.service.netty.JsonNettyStreamingService

objectMapper

Fields inherited from class info.bitrich.xchangestream.service.netty.NettyStreamingService

channels, DEFAULT_CONNECTION_TIMEOUT, DEFAULT_IDLE_TIMEOUT, DEFAULT_RETRY_DURATION, uri

Fields inherited from class info.bitrich.xchangestream.service.ConnectableService

BEFORE_CONNECTION_HANDLER

Constructor Summary

Constructors

Constructor	Description
BinanceStreamingService(String baseUri, ProductSubscription productSubscription, KlineSubscription klineSubscription)
BinanceStreamingService(String baseUri, ProductSubscription productSubscription, KlineSubscription klineSubscription, int maxFramePayloadLength, Duration connectionTimeout, Duration retryDuration, int idleTimeoutSeconds)

Method Summary

Modifier and Type	Method	Description
void	disableLiveSubscription()
void	enableLiveSubscription()
protected String	getChannelNameFromMessage(com.fasterxml.jackson.databind.JsonNode message)
KlineSubscription	getKlineSubscription()
ProductSubscription	getProductSubscription()	The available subscriptions for this streaming service.
String	getSubscribeMessage(String channelName, Object... args)	Get the live subscription message
String	getUnsubscribeMessage(String channelName, Object... args)	Get the live unsubscription message
protected io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler	getWebSocketClientExtensionHandler()
protected WebSocketClientHandler	getWebSocketClientHandler(io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker handshake, WebSocketClientHandler.WebSocketMessageHandler handler)
protected void	handleMessage(com.fasterxml.jackson.databind.JsonNode message)
boolean	isLiveSubscriptionEnabled()
void	resubscribeChannels()	We override this method because we must not use Live Subscription in case of reconnection.
void	sendMessage(String message)
void	setChannelInactiveHandler(WebSocketClientHandler.WebSocketMessageHandler channelInactiveHandler)
void	unsubscribeChannel(String channelId)	Live Unsubscription from stream.

Methods inherited from class info.bitrich.xchangestream.service.netty.JsonNettyStreamingService

messageHandler, processArrayMessageSeparately, sendObjectMessage

Methods inherited from class info.bitrich.xchangestream.service.netty.NettyStreamingService

disconnect, getChannel, getCustomHeaders, getSubscriptionUniqueId, handleChannelError, handleChannelMessage, handleError, handleIdle, isSocketOpen, openConnection, setAcceptAllCertificates, setAutoReconnect, setEnableLoggingHandler, setLoggingHandlerLevel, setSocksProxyHost, setSocksProxyPort, subscribeChannel, subscribeConnectionState, subscribeConnectionSuccess, subscribeDisconnect, subscribeIdle, subscribeReconnectFailure, useCompressedMessages

Methods inherited from class info.bitrich.xchangestream.service.ConnectableService

connect, setBeforeConnectionHandler

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

BinanceStreamingService

public BinanceStreamingService(String baseUri,
 ProductSubscription productSubscription,
 KlineSubscription klineSubscription)

BinanceStreamingService

public BinanceStreamingService(String baseUri,
 ProductSubscription productSubscription,
 KlineSubscription klineSubscription,
 int maxFramePayloadLength,
 Duration connectionTimeout,
 Duration retryDuration,
 int idleTimeoutSeconds)

Method Details

getChannelNameFromMessage

protected String getChannelNameFromMessage(com.fasterxml.jackson.databind.JsonNode message)

Specified by:

getChannelNameFromMessage in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

handleMessage

protected void handleMessage(com.fasterxml.jackson.databind.JsonNode message)

Overrides:

handleMessage in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

resubscribeChannels

public void resubscribeChannels()

We override this method because we must not use Live Subscription in case of reconnection. The reason is that Binance has a Websocket limits to 5 incoming messages per second. If we pass this limit the socket is closed automatically by Binance. See ... for more details. All the channels will be resubscribed at connection time.

Overrides:

resubscribeChannels in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

getSubscribeMessage

public String getSubscribeMessage(String channelName,
 Object... args)
                           throws IOException

Get the live subscription message

Specified by:

getSubscribeMessage in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

Throws:

IOException

getUnsubscribeMessage

public String getUnsubscribeMessage(String channelName,
 Object... args)
                             throws IOException

Get the live unsubscription message

Specified by:

getUnsubscribeMessage in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

Throws:

IOException

sendMessage

public void sendMessage(String message)

Overrides:

sendMessage in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

getWebSocketClientExtensionHandler

protected io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler getWebSocketClientExtensionHandler()

Overrides:

getWebSocketClientExtensionHandler in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

getProductSubscription

public ProductSubscription getProductSubscription()

The available subscriptions for this streaming service.

Returns:

The subscriptions for the currently open connection.

getKlineSubscription

public KlineSubscription getKlineSubscription()

enableLiveSubscription

public void enableLiveSubscription()

disableLiveSubscription

public void disableLiveSubscription()

isLiveSubscriptionEnabled

public boolean isLiveSubscriptionEnabled()

unsubscribeChannel

public void unsubscribeChannel(String channelId)

Live Unsubscription from stream. This send a message through the websocket to Binance with method UNSUBSCRIBE. (see ... for more details) This is the only way to really stop receiving data from the stream (Disposable.dispose() dispose the resource but don't stop the data to be received from Binance).

Parameters:

channelId - e.g. btcusdt@depth, btcusdt@trade

getWebSocketClientHandler

protected WebSocketClientHandler getWebSocketClientHandler(io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker handshake,
 WebSocketClientHandler.WebSocketMessageHandler handler)

Overrides:

getWebSocketClientHandler in class NettyStreamingService<com.fasterxml.jackson.databind.JsonNode>

setChannelInactiveHandler

public void setChannelInactiveHandler(WebSocketClientHandler.WebSocketMessageHandler channelInactiveHandler)