Polish code by extracting func and extracting struct

Signed-off-by: Jiangnan Jia <[email protected]>

Author: jnan806

pkg/client/opensergo_client.go

@@ -15,40 +15,48 @@
 package client
 
 import (
-	"context"
-	"strconv"
-	"sync/atomic"
-	"time"
-
 	"github.com/golang/groupcache"
 	"github.com/opensergo/opensergo-go/pkg/api"
+	"github.com/opensergo/opensergo-go/pkg/client/stream"
 	"github.com/opensergo/opensergo-go/pkg/common/logging"
+	"github.com/opensergo/opensergo-go/pkg/global"
 	"github.com/opensergo/opensergo-go/pkg/model"
 	transportPb "github.com/opensergo/opensergo-go/pkg/proto/transport/v1"
 	"github.com/opensergo/opensergo-go/pkg/transport/subscribe"
-	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"strconv"
 )
 
 // OpenSergoClient is the client to communicate with opensergo-control-plane.
 type OpenSergoClient struct {
-	host                   string
-	port                   uint32
-	transportServiceClient transportPb.OpenSergoUniversalTransportServiceClient
+	host string
+	port uint32
 
-	subscribeConfigStreamPtr         atomic.Value // type of value is *client.subscribeConfigStream
-	subscribeConfigStreamObserverPtr atomic.Value // type of value is *client.SubscribeConfigStreamObserver
-	subscribeConfigStreamStatus      atomic.Value // type of value is client.OpensergoClientStreamStatus
+	opensergoOptions *api.OpensergoOptions
+	clientOptions    *api.ClientOptions
 
-	subscribeDataCache *subscribe.SubscribeDataCache
-	subscriberRegistry *subscribe.SubscriberRegistry
+	transportServiceClient    transportPb.OpenSergoUniversalTransportServiceClient
+	subscribeConfigOutsStream *stream.SubscribeConfigOutsStream
 
 	requestId groupcache.AtomicInt
 }
 
 // NewOpenSergoClient returns an instance of OpenSergoClient, and init some properties.
-func NewOpenSergoClient(host string, port uint32) (*OpenSergoClient, error) {
+func NewOpenSergoClient(host string, port uint32, opts ...api.ClientOption) (*OpenSergoClient, error) {
+	// globalOpts
+	globalOptions := global.GetGlobalOptions()
+	// OpensergoOptions from GetGlobalOptions
+	opensergoOptions := globalOptions.OpensergoOptions()
+	// ClientOptions from GetGlobalOptions
+	clientOptions := globalOptions.ClientOptions()
+	// override default ClientOptions by params
+	if len(opts) > 0 {
+		for _, opt := range opts {
+			opt(clientOptions)
+		}
+	}
+
 	address := host + ":" + strconv.FormatUint(uint64(port), 10)
 	clientConn, err := grpc.Dial(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
@@ -57,165 +65,58 @@ func NewOpenSergoClient(host string, port uint32) (*OpenSergoClient, error) {
 	transportServiceClient := transportPb.NewOpenSergoUniversalTransportServiceClient(clientConn)
 
 	openSergoClient := &OpenSergoClient{
-		host:                   host,
-		port:                   port,
-		transportServiceClient: transportServiceClient,
-		subscribeDataCache:     &subscribe.SubscribeDataCache{},
-		subscriberRegistry:     &subscribe.SubscriberRegistry{},
-		requestId:              groupcache.AtomicInt(0),
+		host:                      host,
+		port:                      port,
+		opensergoOptions:          opensergoOptions,
+		clientOptions:             clientOptions,
+		transportServiceClient:    transportServiceClient,
+		subscribeConfigOutsStream: stream.NewSubscribeConfigOutsStream(opensergoOptions, clientOptions, transportServiceClient),
+		requestId:                 groupcache.AtomicInt(0),
 	}
-	openSergoClient.subscribeConfigStreamPtr.Store(&subscribeConfigStream{})
-	openSergoClient.subscribeConfigStreamObserverPtr.Store(&subscribeConfigStreamObserver{})
-	openSergoClient.subscribeConfigStreamStatus.Store(initial)
 	return openSergoClient, nil
 }
 
 func (c *OpenSergoClient) SubscribeDataCache() *subscribe.SubscribeDataCache {
-	return c.subscribeDataCache
+	return c.subscribeConfigOutsStream.SubscribeDataCache
 }
 
 func (c *OpenSergoClient) SubscriberRegistry() *subscribe.SubscriberRegistry {
-	return c.subscriberRegistry
-}
-
-// KeepAlive
-//
-// keepalive OpenSergoClient
-func (c *OpenSergoClient) keepAlive() {
-	// TODO change to event-driven-model instead of for-loop
-	for {
-		status := c.CurrentStreamStatus()
-		if status == interrupted {
-			logging.Info("Try to restart OpenSergoClient...")
-			// We do not handle error here because error has print in Start()
-			_ = c.Start()
-		}
-		time.Sleep(time.Duration(10) * time.Second)
-	}
-}
-
-func (c *OpenSergoClient) CurrentStreamStatus() OpensergoClientStreamStatus {
-	return c.subscribeConfigStreamStatus.Load().(OpensergoClientStreamStatus)
+	return c.subscribeConfigOutsStream.SubscriberRegistry
 }
 
 // Start the OpenSergoClient.
 func (c *OpenSergoClient) Start() error {
 	logging.Info("OpenSergoClient is starting...")
 
-	// keepalive OpensergoClient
-	if c.CurrentStreamStatus() == initial {
-		logging.Info("Start a keepalive() daemon goroutine to keep OpenSergoClient alive")
-		go c.keepAlive()
+	if err := c.subscribeConfigOutsStream.Start(); err != nil {
+		logging.Error(err, "OpenSergoClient cannot start", "host", c.host, "port", c.port)
+		return err
 	}
 
-	c.subscribeConfigStreamStatus.Store(starting)
+	logging.Info("OpenSergoClient has been started", "host", c.host, "port", c.port)
+	return nil
+}
 
-	stream, err := c.transportServiceClient.SubscribeConfig(context.Background())
-	if err != nil {
-		logging.Error(err, "[OpenSergo SDK] SubscribeConfigStream can not connect.")
-		c.subscribeConfigStreamStatus.Store(interrupted)
-		return err
-	}
-	c.subscribeConfigStreamPtr.Store(&subscribeConfigStream{stream: stream})
+// Close the OpenSergoClient.
+func (c *OpenSergoClient) Close() error {
+	logging.Info("OpenSergoClient is closing...")
 
-	if c.subscribeConfigStreamObserverPtr.Load().(*subscribeConfigStreamObserver).observer == nil {
-		c.subscribeConfigStreamObserverPtr.Store(&subscribeConfigStreamObserver{observer: NewSubscribeConfigStreamObserver(c)})
-		c.subscribeConfigStreamObserverPtr.Load().(*subscribeConfigStreamObserver).observer.Start()
-	}
+	// TODO do close
 
-	logging.Info("[OpenSergo SDK] begin to subscribe config-data...")
-	// TODO: handle error inside the ForEach here.
-	c.subscriberRegistry.ForEachSubscribeKey(func(key model.SubscribeKey) bool {
-		err := c.SubscribeConfig(key)
-		if err != nil {
-			logging.Error(err, "Failed to SubscribeConfig for key", "namespace", key.Namespace(), "app", key.App(), "kind", key.Kind())
-		}
-		return true
-	})
+	c.subscribeConfigOutsStream.Shutdown()
 
-	logging.Info("OpenSergoClient has been started", "host", c.host, "port", c.port)
-	c.subscribeConfigStreamStatus.Store(started)
+	logging.Info("OpenSergoClient has been closed", "host", c.host, "port", c.port)
 	return nil
 }
 
 // SubscribeConfig sends a subscribe request to opensergo-control-plane,
 // and return the result of subscribe config-data from opensergo-control-plane.
 func (c *OpenSergoClient) SubscribeConfig(subscribeKey model.SubscribeKey, opts ...api.SubscribeOption) error {
-	configStream := c.subscribeConfigStreamPtr.Load().(*subscribeConfigStream)
-	if configStream == nil || configStream.stream == nil {
-		logging.Warn("gRPC stream in OpenSergoClient is nil! Cannot subscribe config-data. waiting for keepalive goroutine to restart...", "namespace", subscribeKey.Namespace(), "app", subscribeKey.App(), "kinds", subscribeKey.Kind().GetName())
-		c.subscribeConfigStreamStatus.Store(interrupted)
-		return errors.New("configStream not ready (nil)")
-	}
-
-	options := &api.SubscribeOptions{}
-	if len(opts) > 0 {
-		for _, opt := range opts {
-			opt(options)
-		}
-	}
-
-	// Register subscribers.
-	if len(options.Subscribers) > 0 {
-		for _, subscriber := range options.Subscribers {
-			c.subscriberRegistry.RegisterSubscriber(subscribeKey, subscriber)
-		}
-	}
-
-	subscribeRequestTarget := transportPb.SubscribeRequestTarget{
-		Namespace: subscribeKey.Namespace(),
-		App:       subscribeKey.App(),
-		Kinds:     []string{subscribeKey.Kind().GetName()},
-	}
-
-	subscribeRequest := &transportPb.SubscribeRequest{
-		Target: &subscribeRequestTarget,
-		OpType: transportPb.SubscribeOpType_SUBSCRIBE,
-	}
-
-	err := c.subscribeConfigStreamPtr.Load().(*subscribeConfigStream).stream.Send(subscribeRequest)
-	if err != nil {
-		logging.Error(err, "Failed to send or handle SubscribeConfig request", "namespace", subscribeKey.Namespace(), "app", subscribeKey.App(), "kinds", subscribeKey.Kind().GetName())
-		return err
-	}
-
-	logging.Info("Subscribe success", "namespace", subscribeKey.Namespace(), "app", subscribeKey.App(), "kinds", subscribeKey.Kind().GetName())
-
-	return nil
+	return c.subscribeConfigOutsStream.SubscribeConfig(subscribeKey, opts...)
 }
 
-// UnsubscribeConfig sends an un-subscribe request to opensergo-control-plane
+// UnSubscribeConfig sends an un-subscribe request to opensergo-control-plane
 // and remove all the subscribers by subscribeKey.
-func (c *OpenSergoClient) UnsubscribeConfig(subscribeKey model.SubscribeKey) error {
-
-	if c.subscribeConfigStreamPtr.Load().(*subscribeConfigStream).stream == nil {
-		logging.Warn("gRPC stream in OpenSergoClient is nil! Cannot unsubscribe config-data. waiting for keepalive goroutine to restart...", "namespace", subscribeKey.Namespace(), "app", subscribeKey.App(), "kinds", subscribeKey.Kind().GetName())
-		c.subscribeConfigStreamStatus.Store(interrupted)
-		return errors.New("configStream not ready (nil)")
-	}
-
-	subscribeRequestTarget := transportPb.SubscribeRequestTarget{
-		Namespace: subscribeKey.Namespace(),
-		App:       subscribeKey.App(),
-	}
-	subscribeRequestTarget.Kinds = append(subscribeRequestTarget.Kinds, subscribeKey.Kind().GetName())
-
-	subscribeRequest := &transportPb.SubscribeRequest{
-		Target: &subscribeRequestTarget,
-		OpType: transportPb.SubscribeOpType_UNSUBSCRIBE,
-	}
-
-	// Send SubscribeRequest (unsubscribe command)
-	err := c.subscribeConfigStreamPtr.Load().(*subscribeConfigStream).stream.Send(subscribeRequest)
-	if err != nil {
-		logging.Error(err, "Failed to send or handle UnsubscribeConfig request", "namespace", subscribeKey.Namespace(), "app", subscribeKey.App(), "kinds", subscribeKey.Kind().GetName())
-		return err
-	}
-
-	logging.Info("Unsubscribe success", "namespace", subscribeKey.Namespace(), "app", subscribeKey.App(), "kinds", subscribeKey.Kind().GetName())
-
-	// Remove subscribers of the subscribe target.
-	c.subscriberRegistry.RemoveSubscribers(subscribeKey)
-
-	return nil
+func (c *OpenSergoClient) UnSubscribeConfig(subscribeKey model.SubscribeKey) error {
+	return c.subscribeConfigOutsStream.UnSubscribeConfig(subscribeKey)
 }

pkg/client/opensergo_client_model.go renamed to pkg/client/stream/stream_model.go

@@ -12,23 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package client
+package stream
 
-import transportPb "github.com/opensergo/opensergo-go/pkg/proto/transport/v1"
+type keepalive struct {
+	flag bool
+}
 
-type OpensergoClientStreamStatus uint8
+// PbStreamStatus is the stream status of OpenSergo Universal Transport Service
+type PbStreamStatus uint8
 
 const (
-	initial OpensergoClientStreamStatus = iota
-	starting
-	started
-	interrupted
+	PbStreamInitial PbStreamStatus = iota
+	PbStreamStarting
+	PbStreamStarted
+	PbStreamInterrupted
+	PbStreamShutdown
 )
 
-type subscribeConfigStream struct {
-	stream transportPb.OpenSergoUniversalTransportService_SubscribeConfigClient
-}
+// PbStreamObserverStatus is the stream status of OpenSergo Universal Transport Service
+type PbStreamObserverStatus uint8
+
+const (
+	PbStreamObserverInitial PbStreamObserverStatus = iota
+	PbStreamObserverRunning
+	PbStreamObserverStopped
+)
 
-type subscribeConfigStreamObserver struct {
-	observer *SubscribeConfigStreamObserver
+type outsStream interface {
+	StreamStatus() PbStreamStatus
 }