@@ -16,6 +16,7 @@ The above copyright notice and this permission notice shall be included in all c
16
16
using System . Linq ;
17
17
using System . Text ;
18
18
using System . Threading . Tasks ;
19
+ using System . Threading ;
19
20
20
21
// different namespace for this since we don't want this to be easily accessible publically
21
22
namespace ExchangeSharp . OKGroup
@@ -643,28 +644,37 @@ private ExchangeOrderResult ParseOrder(JToken token)
643
644
644
645
private IWebSocket ConnectWebSocketOkex ( Func < IWebSocket , Task > connected , Func < IWebSocket , string , string [ ] , JToken , Task > callback , int symbolArrayIndex = 3 )
645
646
{
646
- return ConnectWebSocket ( string . Empty , async ( _socket , msg ) =>
647
+ Timer pingTimer = null ;
648
+ return ConnectWebSocket ( url : string . Empty , messageCallback : async ( _socket , msg ) =>
647
649
{
648
- // https://github.com/okcoin-okex/API-docs-OKEx.com/blob/master/README-en.md
649
- // All the messages returning from WebSocket API will be optimized by Deflate compression
650
- JToken token = JToken . Parse ( msg . ToStringFromUTF8Deflate ( ) ) ;
651
-
650
+ // https://github.com/okcoin-okex/API-docs-OKEx.com/blob/master/README-en.md
651
+ // All the messages returning from WebSocket API will be optimized by Deflate compression
652
+ var msgString = msg . ToStringFromUTF8Deflate ( ) ;
653
+ if ( msgString == "pong" )
654
+ { // received reply to our ping
655
+ return ;
656
+ }
657
+ JToken token = JToken . Parse ( msgString ) ;
652
658
var eventProperty = token [ "event" ] ? . ToStringInvariant ( ) ;
653
659
if ( eventProperty != null )
654
660
{
655
- if ( eventProperty == "error" )
656
- {
661
+ if ( eventProperty == "error" )
662
+ {
657
663
Logger . Info ( "Websocket unable to connect: " + token [ "message" ] ? . ToStringInvariant ( ) ) ;
658
664
return ;
659
- }
660
-
661
- if ( eventProperty == "subscribe" && token [ "channel" ] != null )
662
- {
663
- return ;
664
- }
665
+ }
666
+ else if ( eventProperty == "subscribe" && token [ "channel" ] != null )
667
+ { // subscription successful
668
+ if ( pingTimer == null )
669
+ {
670
+ pingTimer = new Timer ( callback : async s => await _socket . SendMessageAsync ( "ping" ) ,
671
+ state : null , dueTime : 0 , period : 15000 ) ; // send a ping every 15 seconds
672
+ }
673
+ return ;
674
+ }
675
+ else return ;
665
676
}
666
-
667
- if ( token [ "table" ] != null )
677
+ else if ( token [ "table" ] != null )
668
678
{
669
679
var data = token [ "data" ] ;
670
680
foreach ( var dataRow in data )
@@ -673,10 +683,13 @@ private IWebSocket ConnectWebSocketOkex(Func<IWebSocket, Task> connected, Func<I
673
683
await callback ( _socket , marketSymbol , null , dataRow ) ;
674
684
}
675
685
}
676
- } , async ( _socket ) =>
677
- {
678
- await connected ( _socket ) ;
679
- } ) ;
686
+ } , connectCallback : async ( _socket ) => await connected ( _socket )
687
+ , disconnectCallback : s =>
688
+ {
689
+ pingTimer . Dispose ( ) ;
690
+ pingTimer = null ;
691
+ return Task . CompletedTask ;
692
+ } ) ;
680
693
}
681
694
682
695
private IWebSocket ConnectPrivateWebSocketOkex ( Func < IWebSocket , Task > connected , Func < IWebSocket , string , string [ ] , JToken , Task > callback , int symbolArrayIndex = 3 )
0 commit comments