$tree..├── client│   ├── client.go│   ├── request.go│   └── response.go├── go.mod├── go.sum└── main.go1 directory, 6 files

packagemainimport("context""practice/bybit/websocket/bybit")funcmain(){clt,err:=bybit.NewClient()iferr!=nil{panic(err)}ctx:=context.Background()tickerSubsciption:=bybit.Request{Op:"subscribe",Args:[]interface{}{bybit.TickersTONUSDTTopic,},}subscriptions:=[]bybit.Request{tickerSubsciption,}clt.Run(ctx,subscriptions)}

packageclientimport("context""encoding/json""fmt""net/url""os""os/signal""time""github.com/gorilla/websocket")// Bybit organize the topics to which// we can subscribe into Public and Private topics.// Basically topics to which you can subscribe without authentication// and the ones you can only subscribe with Authentication.typeChannelTypestring// There are several types of trades on bybit// we are going to use only spot.typeTradesTypestringconst(PublicChannelChannelType="public"PrivateChannelChannelType="private"SpotTradesType="spot"// topic for getting latest prices of// TONUSDT pair in spot trading.TickersTONUSDTTopic="tickers.TONUSDT"APIVersion="v5"ByBitWebsocketDomain="stream.bybit.com"// In order to comply with bybit documentation is necessary to// send every 20 seconds a ping command to avoid disconnection.// A basic heartbeat mechanism.PingTimeout=20// Even providing the heartbeat you will suffer disconnections// simply because that's how programming works, you should be prepared// for every kind of errors.MaxRetrialConnections=10)// Client represents connection with ByBit Websocket API.typeClientstruct{conn*websocket.Conn}funcNewClient()(*Client,error){conn,err:=connect()iferr!=nil{returnnil,err}return&Client{conn:conn},nil}funcpath(channelTypeChannelType,operationTradesType)string{returnfmt.Sprintf("/%s/%s/%s",APIVersion,channelType,operation)}// connect to bybit websocket stream.funcconnect()(*websocket.Conn,error){spotPath:=path(PublicChannel,Spot)u:=url.URL{Scheme:"wss",Host:ByBitWebsocketDomain,Path:spotPath}fmt.Printf("connecting to %s\n",u.String())conn,_,err:=websocket.DefaultDialer.Dial(u.String(),nil)returnconn,err}func(c*Client)sendPing()error{pingReq:=Request{ReqID:"100001",Op:"ping",}returnc.conn.WriteJSON(&pingReq)}// replace connection with a new onefunc(c*Client)reconnect()error{conn,err:=connect()iferr!=nil{returnerr}c.conn=connreturnnil}// Run connect to bybit websocket, general idea of what it does.// 1. Subscribe to tickers// 2. Read message from websocket.// 3. Send every 20 seconds a ping(heartbeat), to avoid disconnections.// 4. In case of abnormal close of connection, performs a reconnection.// 5. In case the reconnection exceed the max allowed, shut the program.// 6. Also listen to Ctr+C commands to shutdown gratefully.func(c*Client)Run(ctxcontext.Context,subscriptions[]Request,)error{// interupt signal necessary for catching Ctr+C// and shut down gratefullyinterrupt:=make(chanos.Signal,1)signal.Notify(interrupt,os.Interrupt)// ticker for sending ping command every 20 seconds// for the heartbeat mechanism.pingTicker:=time.NewTicker(PingTimeout*time.Second)// we are going to report our errors in this channelerrChn:=make(chanerror)done:=make(chanstruct{})varconnectionsint=0// in case we are going to have a disconnection we are going// back to this tag to reconnect.CONNECTION:ifconnections>0{err:=c.reconnect()iferr!=nil{returnerr}}deferfunc(){pingTicker.Stop()close(errChn)c.conn.Close()}()// let's increase our connection counter// to keep a record of our max retriesconnections++// this gourutine will process every message incomming// related to the subscriptions passed.gofunc(){deferclose(done)err:=c.processRead(ctx,done,subscriptions)iferr!=nil{errChn<-errreturn}}()// wait for possible errors and interrupt signals.// send ping command every PingTimeout(20) seconds.for{select{caseerr:=<-errChn:// not all the errors are retriableifwaitTime,ok:=retriableError(err,connections);ok{ifconnections>MaxRetrialConnections{returnerr}fmt.Println("RECONNECTING.....")fmt.Printf("Sleeping %v milliseconds\n",waitTime)time.Sleep(waitTime)// go to tag defined abovegotoCONNECTION}returnerr// in case of Ctr+C let's handle the connectioncase<-interrupt:returnc.handleInterruptSignal(done)// when ticker is triggered send a ping commandcase<-pingTicker.C:err:=c.sendPing()iferr!=nil{returnerr}}}}func(c*Client)processMsg(ctxcontext.Context,data[]byte)error{varmsgPublicResponseerr:=json.Unmarshal(data,&msg)iferr!=nil{returnerr}fmt.Printf("MSG: %s\n",string(data))returnnil}func(c*Client)processRead(ctxcontext.Context,donechanstruct{},subscriptions[]Request,)error{// first ping to send.c.sendPing()for_,subscription:=rangesubscriptions{err:=c.conn.WriteJSON(subscription)iferr!=nil{returnfmt.Errorf("sending subscription %w",err)}}for{select{case<-done:returnnildefault:_,message,err:=c.conn.ReadMessage()iferr!=nil{returnfmt.Errorf("reading %w",err)}err=c.processMsg(ctx,message)iferr!=nil{fmt.Println("ERR PROCESSING READ: ",err.Error())}}}}func(c*Client)handleInterruptSignal(donechanstruct{},)error{fmt.Println("Closing connection...it might take a few seconds until the next tickers comes")done<-struct{}{}// Cleanly close the connection by sending a close message and then// waiting (with timeout) for the server to close the connection.err:=c.conn.WriteMessage(websocket.CloseMessage,websocket.FormatCloseMessage(websocket.CloseNormalClosure,""))iferr!=nil{returnerr}select{case<-time.After(time.Second):}returnnil}funcretriableError(errerror,connectionsint)(time.Duration,bool){waitTime:=time.Duration(connections)*500*time.Millisecond// 1006(CloseAbnormalClosure) checking if was an abnormal closure.isAbnormalClosure:=websocket.IsUnexpectedCloseError(err,websocket.CloseAbnormalClosure)// 1001(CloseGoingAway) indicates that an endpoint is "going away", such as a server// going down or a browser having navigated away from a page.isGoingAway:=websocket.IsUnexpectedCloseError(err,websocket.CloseGoingAway)isRetriable:=isAbnormalClosure||isGoingAway// 1013(CloseTryAgainLater) not defined in rfc6455 but can be faced as well.ifwebsocket.IsUnexpectedCloseError(err,websocket.CloseTryAgainLater){isRetriable=truewaitTime=time.Duration(connections)*time.Second}returnwaitTime,isRetriable}

connecting to wss://stream.bybit.com/v5/public/spotMSG:{"success":true,"ret_msg":"pong","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","req_id":"100001","op":"ping"}MSG:{"success":true,"ret_msg":"subscribe","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","op":"subscribe"}MSG:{"topic":"tickers.TONUSDT","ts":1684322899614,"type":"snapshot","cs":9133729124,"data":{"symbol":"TONUSDT","lastPrice":"1.9916","highPrice24h":"2.0063","lowPrice24h":"1.9694","prevPrice24h":"1.9729","volume24h":"115096.79","turnover24h":"228228.375398","price24hPcnt":"0.0095","usdIndexPrice":""}}MSG:{"topic":"tickers.TONUSDT","ts":1684322975467,"type":"snapshot","cs":9133777693,"data":{"symbol":"TONUSDT","lastPrice":"1.9916","highPrice24h":"2.0063","lowPrice24h":"1.9694","prevPrice24h":"1.9729","volume24h":"115096.79","turnover24h":"228228.375398","price24hPcnt":"0.0095","usdIndexPrice":""}}MSG:{"success":true,"ret_msg":"pong","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","req_id":"100001","op":"ping"}MSG:{"success":true,"ret_msg":"pong","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","req_id":"100001","op":"ping"}MSG:{"success":true,"ret_msg":"pong","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","req_id":"100001","op":"ping"}MSG:{"topic":"tickers.TONUSDT","ts":1684323035468,"type":"snapshot","cs":9133813935,"data":{"symbol":"TONUSDT","lastPrice":"1.9916","highPrice24h":"2.0063","lowPrice24h":"1.9694","prevPrice24h":"1.9736","volume24h":"115096.79","turnover24h":"228228.375398","price24hPcnt":"0.0091","usdIndexPrice":""}}MSG:{"success":true,"ret_msg":"pong","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","req_id":"100001","op":"ping"}MSG:{"success":true,"ret_msg":"pong","conn_id":"d0822166-72b0-44f7-b6d9-101e5c0c98ad","req_id":"100001","op":"ping"}