Feature request: Implement ability to check SocketIO status from StreamingMarketDataService

[aido]

Currently StreamingMarketDataService does not give the ability to check a socketIO connection isConnected(). If a connection drops the current implementaton does not allow a test and reconnect.

[timmolter]

That's true. I'll take a look at it when I get back from vacation in a week...

[aido]

Thanks Tim.

Enjoy your vacation!

[timmolter]

Hi Aido, I'm taking a look at this now. Do you know of a good way to test for a disconnect from MtGox Streaming? Also, how often does it occur by itself on average?

[aido]

Hi Tim,

Thanks for looking at this.

Connections to the MtGox StreamingMarketDataService consistently drop every hour or so.

I am not at my computer right now but when the connection drops I get a "java.lang.IllegalStateException: Timer already cancelled" exception. I thought that if I catch this exception I could then somehow utilise the isConnected method in the socketIO class and then reconnect (http://xeiam.com/xchange/javadoc/com/xeiam/xchange/streaming/socketio/SocketIO.html#isConnected().
).

I will confirm and update this comment when I get to my development machine shortly.

[timmolter]

Hi Aido,

I just built a new SNAPSHOT release (1.2.1-SNAPSHOT). What I did was add a method isConnected() to StreamingExchangeService which you can access as shown in *.streaming.TickerDemo.java.

Thanks for the info about the exception. I'll let the TickerDemo run until the connection drops and see the stacktrace for myself. If you see it, and you could post the stacktrace here, that would help as well.

In the next coming days I'd like to beef up the socketIO streaming code so that coders like you won't have to worry about catching exceptions and reconnecting. I'm going to try to come up with an elegant solution that perhaps even automatically reconnects.

Cheers, Tim

[aido]

Hi Tim,

Excellent, thanks a lot.

I have not been able to reproduce the "java.lang.IllegalStateException: Timer already cancelled" exception. It may have been coincidental.

However, I have implemented the isConnected() method. I'm not sure if it is working properly.

When I manually bring down my internet connection get the following repeated several times in the error log:

2012-12-05 00:34:10,453 : ERROR com.xeiam.xchange.service.RunnableSocketIOEventProducer [backgroundTimer] - onError: Timeout Error. No heartbeat from server within life time of the socket. closing.

I also get the following exception:

Exception in thread "backgroundTimer" Exception in thread "backgroundTimer" java.lang.NullPointerException
       at com.xeiam.xchange.service.RunnableSocketIOEventProducer.onError   (RunnableSocketIOEventProducer.java:190)
        at com.xeiam.xchange.streaming.socketio.IOConnection.error(IOConnection.java:437)
        at com.xeiam.xchange.streaming.socketio.IOConnection$HearbeatTimeoutTask.run(IOConnection.java:159)
        at java.util.TimerThread.mainLoop(Timer.java:555)
        at java.util.TimerThread.run(Timer.java:505)

The code I am using to keep the connection alive is first wait for initial connection then if connection fails try again every minute. From looking at the error messages in the error log, I don't think it is getting as far as the retry code:

exchange = com.xeiam.xchange.mtgox.v1.MtGoxExchange.newInstance();
marketData = exchange.getStreamingMarketDataService();
tickerQueue = marketData.requestTicker(Currencies.BTC, currency.getCurrencyCode());
while(!marketData.isConnected()){
    try {
        log.info("Waiting to connect to "+currency.getCode()+" ticker");
        Thread.currentThread().sleep(Constants.ONESECOND * 10);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}
.
.
.
.
.
while(!marketData.isConnected()){
    try {
        marketData.cancelTicker();
        exchange = com.xeiam.xchange.mtgox.v1.MtGoxExchange.newInstance();
        marketData = exchange.getStreamingMarketDataService();
        tickerQueue = marketData.requestTicker(Currencies.BTC, currency.getCurrencyCode());
        log.error("ERROR: Lost "+currency.getCode()+" ticker connection. Will try to connect again in one minute");
        Thread.currentThread().sleep(Constants.ONEMINUTE);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

Rather than looking to work around these errors I can wait for your 'beefier' code in the coming days.

Thanks in advance,
Aido

[gary-rowe]

I'll take a look at this one tonight.

[timmolter]

Hi Gary,

I was able to reproduce the problem by simply letting it run. It took over 3 hours, but here is the log:

07:34:15.033 [default] [main] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - requesting Ticker...
07:34:18.788 [default] [pool-4-thread-1] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - {"message":"connected"}
07:34:18.788 [default] [pool-4-thread-1] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - socketIO.isConnected(): true
07:34:19.463 [default] [pool-4-thread-1] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - {"op":"subscribe","channel":"d5f06780-30a8-4a48-a2f8-7ed181b4a13f"}
07:34:19.463 [default] [pool-4-thread-1] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - socketIO.isConnected(): true
10:51:39.012 [default] [backgroundTimer] ERROR
c.x.x.s.RunnableSocketIOEventProducer - onError: Timeout Error. No heartbeat from server within life time of the socket. closing.
10:51:39.053 [default] [pool-4-thread-1] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - {"message":"SocketIOException"}
10:51:39.053 [default] [pool-4-thread-1] INFO
c.x.x.m.v.s.m.s.s.MtGoxStreamingMarketDataService - socketIO.isConnected(): false

There are multiple things to consider, and the current implementation of the streaming socket IO code is pretty rough still. Starting from the client, if you look at *streaming.TickerDemo.java, you'll see that by calling requestTicker, you get a BlockingQueue. This is nice, but there is really no way to get any other information back such as errors that have occurred.

In MtGoxStreamingMarketDataService.java, requestTicker(), we get all the ExchangeEvents, which include the Ticker data and also any error data etc. In RunnableSocketIOEventProducer.java, I recently added code to pass a JSON snippet into the queue in all the callback methods.

All in all, the whole design needs some thought to bring it up to a production quality level.

[gary-rowe]

Performed initial commit against the XChange-6 branch to cover this work. Can you take a look at it, Tim, and let me know if it works for you?

[timmolter]

Hi Gary,

Thanks for this! I merged it into develop. I still need to create some mechanism to detect the lost connection through the messages queue and re-establish connection. Any ideas?

[gary-rowe]

The SocketIO underlying code should automatically attempt to re-establish a connection in the event of a failure, if that fails then the DISCONNECT event type is added to the exchange event queue. Consuming applications should then be able use that to bin their market data streaming service and start again through the ticker/connect approach.

[timmolter]

Grat, thanks. I'll try implementing that and hope I can test it with the disconnect events that started this feature request.