grpc generic stub refresh

cmd/exchange/main.go

@@ -107,8 +107,10 @@ func main() {
 	// consul 服务注册
 	register := exchangeConf.Register
 	register.Name = pb.ExchangeService_ServiceDesc.ServiceName
-	if err := dis.Registry(grpcServer, register); err != nil {
+	if deregister, err := dis.Registry(grpcServer, register); err != nil {
 		panic(err)
+	} else {
+		defer deregister()
 	}
 
 	// run grpc server

cmd/market/main.go

@@ -68,8 +68,10 @@ func main() {
 	register := marketConf.Register
 	register.Name = pb.MarketService_ServiceDesc.ServiceName
 	dis := discovery.NewConsulDiscovery(client)
-	if err := dis.Registry(grpcServer, register); err != nil {
+	if deregister, err := dis.Registry(grpcServer, register); err != nil {
 		panic(err)
+	} else {
+		defer deregister()
 	}
 
 	// run grpc server

cmd/sig-admin/main.go

@@ -38,6 +38,9 @@ func main() {
 	if err := gpcGenericClientFactory.Init(); err != nil {
 		panic(err)
 	}
+	if err := dis.WatchServices(gpcGenericClientFactory.RefreshService); err != nil {
+		panic(err)
+	}
 
 	sigServer := sig.NewSigServer(gpcGenericClientFactory)
 	if err := sigServer.Init(); err != nil {

cmd/trading/main.go

@@ -99,8 +99,10 @@ func main() {
 	// consul 服务注册
 	register := tradingConf.Register
 	register.Name = pb.TradingService_ServiceDesc.ServiceName
-	if err := dis.Registry(grpcServer, register); err != nil {
+	if deregister, err := dis.Registry(grpcServer, register); err != nil {
 		panic(err)
+	} else {
+		defer deregister()
 	}
 
 	// run grpc server

config/exchange.toml

@@ -18,8 +18,8 @@ receiveBuffer = 4096
 marketSubscribeLimit = 16
 consumeBatch = 1024
 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume
-httpProxy = "http://192.168.1.5:7890"
-# httpProxy = "http://10.255.183.209:7890"
+# httpProxy = "http://192.168.1.5:7890"
+httpProxy = "http://10.255.183.209:7890"
 
 # 模拟盘API交易地址如下：
 # REST：https://www.okx.com

pkg/grpc/discovery/consul/resolver.go

@@ -132,6 +132,7 @@ func (r *consulResolver) watch() {
 				addr := resolver.Address{
 					Addr:       fmt.Sprintf("%s:%d", s.Service.Address, s.Service.Port),
 					Attributes: attributes.New("service_id", s.Service.ID),
+					ServerName: s.Service.Service,
 				}
 				addrs = append(addrs, addr)
 			}

pkg/grpc/discovery/consul/watch.go

@@ -0,0 +1,112 @@
+package consul
+
+import (
+	"math"
+	"sig-pub/pkg/utils/collect"
+	"sig-pub/pkg/utils/retry"
+	"sig-pub/pkg/zlog"
+	"time"
+
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/api/watch"
+)
+
+// 定义watcher
+type Watcher struct {
+	client   *api.Client
+	wp       *watch.Plan                           // 总的Services变化对应的Plan
+	watchers *collect.SyncMap[string, *watch.Plan] // 对已经进行监控的service作个记录
+	handler  func(serviceName string, passing bool)
+}
+
+func NewWatcher(client *api.Client, handler func(serviceName string, passing bool)) *Watcher {
+	return &Watcher{
+		client:   client,
+		watchers: collect.NewSyncMap[string, *watch.Plan](),
+		handler:  handler,
+	}
+}
+
+// WatchServices
+// watch services doc: https://www.consul.io/docs/dynamic-app-config/watches#services
+func (w *Watcher) WatchServices() (err error) {
+	w.wp, err = watch.Parse(map[string]any{
+		"type": "services",
+	})
+	if err != nil {
+		return
+	}
+	w.wp.Handler = func(i uint64, data any) {
+		switch d := data.(type) {
+		// services
+		case map[string][]string:
+			services := make([]string, 0, len(d))
+			for i := range d {
+				if i == "consul" {
+					continue
+				}
+				services = append(services, i)
+				if _, loaded := w.watchers.LoadOrStore(i, nil); !loaded {
+					w.registerServiceWatcher(i)
+				}
+			}
+			zlog.Infof("consul services update: %v", services)
+
+			// remove unknown services from watchers
+			// var dels []string
+			// w.watchers.Range(func(k string, wp *watch.Plan) bool {
+			// 	wp.Stop()
+			// 	dels = append(dels, k)
+			// 	return true
+			// })
+			// for _, svc := range dels {
+			// 	w.watchers.Delete(svc)
+			// }
+		default:
+			zlog.Debugf("unknown consul watch type: %#v", &d)
+		}
+	}
+
+	go func() {
+		if err := w.wp.RunWithClientAndHclog(w.client, nil); err != nil {
+			zlog.Error("watch services error: ", err)
+			retry.DoWithFixDelay(math.MaxUint32, 3*time.Second, func(retryTimes uint32) (_ struct{}, err error) {
+				err = w.WatchServices()
+				return
+			})
+		}
+	}()
+	return
+}
+
+// 将consul新增的service加入，并监控
+// doc: https://www.consul.io/docs/dynamic-app-config/watches#service
+func (w *Watcher) registerServiceWatcher(serviceName string) error {
+	wp, err := watch.Parse(map[string]any{
+		"type":    "service",
+		"service": serviceName,
+	})
+	if err != nil {
+		return err
+	}
+
+	// 定义service变化后所执行的程序(函数)handler
+	wp.Handler = func(idx uint64, data any) {
+		switch d := data.(type) {
+		case []*api.ServiceEntry:
+			for _, i := range d {
+				if w.handler != nil {
+					w.handler(i.Service.Service, i.Checks.AggregatedStatus() == "passing") // status: passing / critical
+				}
+				// fmt.Printf("service %s 已变化", i.Service.Service)
+				// // 打印service的状态
+				// fmt.Println("service status: ", i.Checks.AggregatedStatus())
+			}
+		}
+	}
+	// 启动监控
+	go wp.RunWithClientAndHclog(w.client, nil)
+
+	w.watchers.Store(serviceName, wp)
+	return nil
+}

pkg/grpc/discovery/consul_naming.go

@@ -24,6 +24,7 @@ func ConsulDialUrl(svrName string) string {
 
 type ConsulDiscovery struct {
 	client  *api.Client
+	watcher *consul.Watcher
 }
 
 func NewConsulDiscovery(client *api.Client) *ConsulDiscovery {
@@ -32,7 +33,7 @@ func NewConsulDiscovery(client *api.Client) *ConsulDiscovery {
 	}
 }
 
-func (r *ConsulDiscovery) Registry(grpcServer grpc.ServiceRegistrar, register Server) (err error) {
+func (r *ConsulDiscovery) Registry(grpcServer grpc.ServiceRegistrar, register Server) (deregister func(), err error) {
 	if strings.Trim(register.Name, " ") == "" {
 		err = errors.New("registry name is empty")
 		return
@@ -66,6 +67,10 @@ func (r *ConsulDiscovery) Registry(grpcServer grpc.ServiceRegistrar, register Se
 		zlog.Errorf("registry service %s error: %v", register.Name, err)
 		return
 	}
+	deregister = func() {
+		err := r.client.Agent().ServiceDeregister(reg.ID)
+		zlog.Infof("service id %s deregister to consul, err=%v", reg.ID, err)
+	}
 	zlog.Infof("service %s registered to consul", register.Name)
 	return
 }
@@ -74,3 +79,8 @@ func (r *ConsulDiscovery) Resolver() (builder resolver.Builder) {
 	builder = consul.NewConsulBuilder(r.client)
 	return
 }
+
+func (r *ConsulDiscovery) WatchServices(handler func(service string, passing bool)) (err error) {
+	r.watcher = consul.NewWatcher(r.client, handler)
+	return r.watcher.WatchServices()
+}

pkg/grpc/generic/generic_client.go

@@ -32,7 +32,7 @@ func NewGpcGenericClient(serviceName string, conn *grpc.ClientConn) *GrpcGeneric
 	}
 }
 
-func (c *GrpcGenericClient) Init(ctx context.Context) (err error) {
+func (c *GrpcGenericClient) InitStub(ctx context.Context) (err error) {
 	client := grpcreflect.NewClientV1(ctx, refv1.NewServerReflectionClient(c.conn))
 	marketServiceSymbol, err := client.FileContainingSymbol(protoreflect.FullName(c.serviceName))
 	if err != nil {

pkg/grpc/generic/generic_client_factory.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"fmt"
 	"sig-pub/pkg/utils/collect"
+	"sig-pub/pkg/utils/retry"
+	"sig-pub/pkg/zlog"
+	"time"
 
 	"google.golang.org/grpc"
 )
@@ -36,7 +39,7 @@ func (f *GrpcGenericClientFactory) NewClient(ctx context.Context, serviceName st
 		return
 	}
 	client = NewGpcGenericClient(serviceName, conn)
-	err = client.Init(ctx)
+	err = client.InitStub(ctx)
 	return
 }
 
@@ -47,3 +50,23 @@ func (f *GrpcGenericClientFactory) GetClient(ctx context.Context, serviceName st
 	})
 	return
 }
+
+// RefreshService 服务重新注册时可能有更改, 刷新旧的 grpc generic stub
+func (f *GrpcGenericClientFactory) RefreshService(serviceName string, passing bool) {
+	zlog.Debugf("service status updated: %s, health=%v", serviceName, passing)
+	if passing {
+		service, ok := f.clientCache.Load(serviceName)
+		if ok {
+			go retry.DoWithFixDelay(10, 2*time.Second, func(_ uint32) (_ struct{}, err error) {
+				ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+				defer cancel()
+				if err = service.InitStub(ctx); err != nil {
+					zlog.Errorf("refresh service stub error: %s, %v", serviceName, err)
+				} else {
+					zlog.Debugf("refresh service stub success: %s", serviceName)
+				}
+				return
+			})
+		}
+	}
+}

pkg/trade/risk_strategy.go

@@ -5,7 +5,12 @@ import (
 )
 
 type IRickStrategy interface {
-	OnSingle(signalSide types.Side)
+	OnSignal(signalSide types.Side) (ok bool, cause Cause)
+}
+
+type RiskStrategyParam struct {
+	SkipOnSideOpposite bool // 当前持有反方向单时
+	SkipOnSideSame     bool // 当前持有相同方向单时
 }
 
 // RiskStrategy 风险管理策略

pkg/trade/trade_strategy.go

@@ -3,6 +3,7 @@ package trade
 // ITradeStrategy 下单策略
 // 根据购买信号和账户信息生成下单参数
 type ITradeStrategy interface {
+	OnSingal()
 }
 
 type TradeStrategyParam struct {

pkg/utils/collect/sync_map.go

@@ -42,5 +42,11 @@ func (m *SyncMap[K, V]) CompareAndSwap(k K, old V, new V) (swapped bool) {
 }
 
 func (m *SyncMap[K, V]) LoadOrStore(k K, v V) (actual any, loaded bool) {
-	return m.m.LoadOrStore(k, v)
+	var cur any
+	cur, loaded = m.m.LoadOrStore(k, v)
+	if !loaded {
+		return
+	}
+	actual = cur.(V)
+	return
 }