binance: Retry keep alive. (decred#2958)

* handle ws reconnect signal

* binance: Retry keep alive.

* testbinance: Add flappy websocket.

* binance: Check market depth subs.

* binance: Desync books on disconnect.

---------

Co-authored-by: Brian Stafford <buck54321@gmail.com>

client/cmd/testbinance/main.go

@@ -51,6 +51,7 @@ var (
 
 	walkingSpeedAdj float64
 	gapRange        float64
+	flappyWS        bool
 
 	xcInfo = &bntypes.ExchangeInfo{
 		Timezone:   "UTC",
@@ -155,6 +156,7 @@ func main() {
 	flag.Float64Var(&gapRange, "gaprange", 0.04, "a ratio of how much the gap can vary. default is 0.04 => 4%")
 	flag.BoolVar(&logDebug, "debug", false, "use debug logging")
 	flag.BoolVar(&logTrace, "trace", false, "use trace logging")
+	flag.BoolVar(&flappyWS, "flappyws", false, "periodically drop websocket clients and delete subscriptions")
 	flag.Parse()
 
 	switch {
@@ -453,6 +455,35 @@ func (f *fakeBinance) run(ctx context.Context) {
 		}
 	}()
 
+	if flappyWS {
+		go func() {
+			tick := func() <-chan time.Time {
+				const minDelay = time.Minute
+				const delayRange = time.Minute * 5
+				return time.After(minDelay + time.Duration(rand.Float64()*float64(delayRange)))
+			}
+			for {
+				select {
+				case <-tick():
+					f.marketsMtx.Lock()
+					for addr, sub := range f.marketSubscribers {
+						sub.Disconnect()
+						delete(f.marketSubscribers, addr)
+					}
+					f.marketsMtx.Unlock()
+					f.accountSubscribersMtx.Lock()
+					for apiKey, sub := range f.accountSubscribers {
+						sub.Disconnect()
+						delete(f.accountSubscribers, apiKey)
+					}
+					f.accountSubscribersMtx.Unlock()
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+	}
+
 	f.srv.Run(ctx)
 }
 
@@ -505,6 +536,11 @@ func (f *fakeBinance) handleAccountSubscription(w http.ResponseWriter, r *http.R
 	}()
 }
 
+type listSubsResp struct {
+	ID     uint64   `json:"id"`
+	Result []string `json:"result"`
+}
+
 func (f *fakeBinance) handleMarketStream(w http.ResponseWriter, r *http.Request) {
 	streamsStr := r.URL.Query().Get("streams")
 	if streamsStr == "" {
@@ -554,6 +590,25 @@ func (f *fakeBinance) handleMarketStream(w http.ResponseWriter, r *http.Request)
 		f.cleanMarkets()
 	}
 
+	listSubscriptions := func(id uint64) {
+		f.marketsMtx.Lock()
+		defer f.marketsMtx.Unlock()
+		var streams []string
+		for mktID := range cl.markets {
+			streams = append(streams, fmt.Sprintf("%s@depth", mktID))
+		}
+		resp := listSubsResp{
+			ID:     id,
+			Result: streams,
+		}
+		b, err := json.Marshal(resp)
+		if err != nil {
+			log.Errorf("LIST_SUBSCRIBE marshal error: %v", err)
+		}
+		cl.WSLink.SendRaw(b)
+		f.cleanMarkets()
+	}
+
 	conn, cm := f.newWSLink(w, r, func(b []byte) {
 		var req bntypes.StreamSubscription
 		if err := json.Unmarshal(b, &req); err != nil {
@@ -565,6 +620,8 @@ func (f *fakeBinance) handleMarketStream(w http.ResponseWriter, r *http.Request)
 			subscribe(req.Params)
 		case "UNSUBSCRIBE":
 			unsubscribe(req.Params)
+		case "LIST_SUBSCRIPTIONS":
+			listSubscriptions(req.ID)
 		}
 	})
 	if conn == nil {