diff --git a/pkg/client/opensergo_client.go b/pkg/client/opensergo_client.go index f2807a7..f3615de 100644 --- a/pkg/client/opensergo_client.go +++ b/pkg/client/opensergo_client.go @@ -17,6 +17,7 @@ package client import ( "context" "strconv" + "sync" "sync/atomic" "time" @@ -70,6 +71,32 @@ func NewOpenSergoClient(host string, port uint32) (*OpenSergoClient, error) { return openSergoClient, nil } +var ( + clientMu sync.Mutex + clientPool = make(map[string]*OpenSergoClient) +) + +// GetOpenSergoClientFromPool returns an instance of OpenSergoClient from a pool, based on host:port, +// if it doesn't exist, it will be created and reused for the next call. +func GetOpenSergoClientFromPool(host string, port uint32) (*OpenSergoClient, error) { + address := host + ":" + strconv.FormatUint(uint64(port), 10) + + clientMu.Lock() + defer clientMu.Unlock() + + if client, ok := clientPool[address]; ok { + return client, nil + } + + client, err := NewOpenSergoClient(host, port) + if err != nil { + return nil, err + } + clientPool[address] = client + + return client, nil +} + func (c *OpenSergoClient) SubscribeDataCache() *subscribe.SubscribeDataCache { return c.subscribeDataCache } diff --git a/samples/main/main.go b/samples/main/main.go index 937e79e..6f2efc7 100644 --- a/samples/main/main.go +++ b/samples/main/main.go @@ -34,6 +34,12 @@ func main() { logging.Error(err, "Failed to StartAndSubscribeOpenSergoConfig: %s\n") } + err = StartAndSubscribeOpenSergoConfigFromPool() + if err != nil { + // Handle error here. + logging.Error(err, "Failed to StartAndSubscribeOpenSergoConfigFromPool: %s\n") + } + select {} } @@ -86,3 +92,53 @@ func StartAndSubscribeOpenSergoConfig() error { return err } + +func StartAndSubscribeOpenSergoConfigFromPool() error { + // Set OpenSergo console logger (optional) + consoleLogger := logging.NewConsoleLogger(logging.InfoLevel, logging.JsonFormat, true) + logging.AddLogger(consoleLogger) + // Set OpenSergo file logger (optional) + // fileLogger := logging.NewFileLogger("./opensergo-universal-transport-service.log", logging.InfoLevel, logging.JsonFormat, true) + //logging.AddLogger(fileLogger) + + // Get a OpenSergoClient by pool. + openSergoClient, err := client.GetOpenSergoClientFromPool("127.0.0.1", 10246) + if err != nil { + return err + } + + // Start OpenSergoClient + err = openSergoClient.Start() + if err != nil { + return err + } + + // Create a SubscribeKey for FaultToleranceRule. + faultToleranceSubscribeKey := model.NewSubscribeKey("default", "foo-app", configkind.ConfigKindRefFaultToleranceRule{}) + // Create a Subscriber. + sampleFaultToleranceRuleSubscriber := &samples.SampleFaultToleranceRuleSubscriber{} + // Subscribe data with the key and subscriber. + err = openSergoClient.SubscribeConfig(*faultToleranceSubscribeKey, api.WithSubscriber(sampleFaultToleranceRuleSubscriber)) + if err != nil { + return err + } + + // Create a SubscribeKey for RateLimitStrategy. + rateLimitSubscribeKey := model.NewSubscribeKey("default", "foo-app", configkind.ConfigKindRefRateLimitStrategy{}) + // Create another Subscriber. + sampleRateLimitStrategySubscriber := &samples.SampleRateLimitStrategySubscriber{} + // Subscribe data with the key and subscriber. + err = openSergoClient.SubscribeConfig(*rateLimitSubscribeKey, api.WithSubscriber(sampleRateLimitStrategySubscriber)) + + if err != nil { + return err + } + // Create a SubscribeKey for TrafficRouter + trafficRouterSubscribeKey := model.NewSubscribeKey("default", "service-provider", configkind.ConfigKindTrafficRouterStrategy{}) + // Create another Subscriber + sampleTrafficRouterSubscriber := &samples.SampleTrafficRouterSubscriber{} + // Subscribe data with the key and subscriber + err = openSergoClient.SubscribeConfig(*trafficRouterSubscribeKey, api.WithSubscriber(sampleTrafficRouterSubscriber)) + + return err +}