发布到allCandles

core/candle.go

@@ -156,6 +156,7 @@ func IsModOf(curInt int64, duration time.Duration) bool {
 }
 
 func (core *Core) SaveCandle(instId string, period string, rsp *CandleData, dura time.Duration, withWs bool) {
+	leng := len(rsp.Data)
 	for _, v := range rsp.Data {
 		candle := Candle{
 			InstId: instId,
@@ -164,6 +165,13 @@ func (core *Core) SaveCandle(instId string, period string, rsp *CandleData, dura
 			From:   "rest",
 		}
 		candle.SetToKey(core)
+		// 发布到allCandles|publish, 给外部订阅者用于setToKey
+		arys := []string{ALLCANDLES_PUBLISH}
+		if withWs {
+			arys = append(arys, ALLCANDLES_INNER_PUBLISH)
+		}
+		core.AddToGeneralCandleChnl(&candle, arys)
+		time.Sleep(dura / time.Duration(leng))
 	}
 }
 

core/core.go

@@ -513,3 +513,20 @@ func (cr *Core) RestPostWrapper(url string, param map[string]interface{}) (rest.
 	cr.RedisCli.Publish(ORDER_RESP_PUBLISH+suffix, string(bj))
 	return res.V5Response, nil
 }
+
+func (cr *Core) AddToGeneralCandleChnl(candle *Candle, channels []string) {
+	redisCli := cr.RedisCli
+	ab, _ := json.Marshal(candle)
+	for _, v := range channels {
+		suffix := ""
+		env := os.Getenv("GO_ENV")
+		if env == "demoEnv" {
+			suffix = "-demoEnv"
+		}
+		vd := v + suffix
+		_, err := redisCli.Publish(vd, string(ab)).Result()
+		if err != nil {
+			fmt.Println("err of ma7|ma30 add to redis2:", err, candle.From)
+		}
+	}
+}