ing

client/client-handler.go

@@ -4,6 +4,7 @@ import (
 	"sync/atomic"
 
 	crc "git.loafle.net/commons/rpc-go/client"
+	csc "git.loafle.net/commons/server-go/client"
 )
 
 type ClientHandler interface {
@@ -16,7 +17,7 @@ type ClientHandlers struct {
 	validated atomic.Value
 }
 
-func (ch *ClientHandlers) Init(clientCtx crc.ClientCtx) error {
+func (ch *ClientHandlers) Init(clientCtx csc.ClientCtx) error {
 	if err := ch.ClientHandlers.Init(clientCtx); nil != err {
 		return err
 	}
@@ -24,7 +25,7 @@ func (ch *ClientHandlers) Init(clientCtx crc.ClientCtx) error {
 	return nil
 }
 
-func (ch *ClientHandlers) OnStart(clientCtx crc.ClientCtx) error {
+func (ch *ClientHandlers) OnStart(clientCtx csc.ClientCtx) error {
 	if err := ch.ClientHandlers.OnStart(clientCtx); nil != err {
 		return err
 	}
@@ -32,12 +33,12 @@ func (ch *ClientHandlers) OnStart(clientCtx crc.ClientCtx) error {
 	return nil
 }
 
-func (ch *ClientHandlers) OnStop(clientCtx crc.ClientCtx) {
+func (ch *ClientHandlers) OnStop(clientCtx csc.ClientCtx) {
 
 	ch.ClientHandlers.OnStop(clientCtx)
 }
 
-func (ch *ClientHandlers) Destroy(clientCtx crc.ClientCtx) {
+func (ch *ClientHandlers) Destroy(clientCtx csc.ClientCtx) {
 
 	ch.ClientHandlers.Destroy(clientCtx)
 }

client/data/data.go

@@ -8,14 +8,13 @@ import (
 	"strings"
 
 	logging "git.loafle.net/commons/logging-go"
-	crc "git.loafle.net/commons/rpc-go/client"
 	csc "git.loafle.net/commons/server-go/client"
 	occp "git.loafle.net/overflow/commons-go/config/probe"
 	"git.loafle.net/overflow/probe/client"
 	"git.loafle.net/overflow/probe/config"
 )
 
-func New() (*crc.Client, error) {
+func New() (csc.Connector, error) {
 	_config := config.GetConfig()
 	if nil == _config {
 		return nil, fmt.Errorf("Config is not available")
@@ -47,5 +46,5 @@ func New() (*crc.Client, error) {
 		logging.Logger().Debugf("Client[%s] has been disconnected", connector.GetName())
 	}
 
-	return client.New("Data", connector, nil), nil
+	return connector, nil
 }

service/CollectorService.go

@@ -27,7 +27,6 @@ type CollectorService struct {
 	ocsp.CollectorService
 	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
 
-	DataClientService   *DataClientService   `annotation:"@Inject()"`
 	ContainerService    *ContainerService    `annotation:"@Inject()"`
 	SensorConfigService *SensorConfigService `annotation:"@Inject()"`
 

service/DataClientService.go

@@ -1,62 +0,0 @@
-package service
-
-import (
-	"fmt"
-	"reflect"
-
-	cda "git.loafle.net/commons/di-go/annotation"
-	cdr "git.loafle.net/commons/di-go/registry"
-	ocsp "git.loafle.net/overflow/commons-go/service/probe"
-	"git.loafle.net/overflow/probe/client/data"
-
-	// For annotation
-	_ "git.loafle.net/overflow/commons-go/core/annotation"
-)
-
-var DataClientServiceType = reflect.TypeOf((*DataClientService)(nil))
-
-func init() {
-	cdr.RegisterType(DataClientServiceType)
-}
-
-type DataClientService struct {
-	ocsp.DataClientService
-	RPCClientService
-
-	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
-}
-
-func (s *DataClientService) InitService() error {
-	if err := s.RPCClientService.InitService(); nil != err {
-		return fmt.Errorf("DataClientService: InitService failed %v", err)
-	}
-
-	return nil
-}
-
-func (s *DataClientService) StartService() error {
-	if err := s.RPCClientService.StartService(); nil != err {
-		return fmt.Errorf("DataClientService: StartService failed %v", err)
-	}
-
-	client, err := data.New()
-	if nil != err {
-		return fmt.Errorf("ProbeClientService: StartService failed %v", err)
-	}
-	s.client = client
-
-	if err := s.client.Start(); nil != err {
-		return fmt.Errorf("DataClientService: StartService failed %v", err)
-	}
-
-	return nil
-}
-
-func (s *DataClientService) StopService() {
-	s.RPCClientService.StopService()
-
-}
-
-func (s *DataClientService) DestroyService() {
-	s.RPCClientService.DestroyService()
-}

service/DataService.go

@@ -1,48 +0,0 @@
-package service
-
-import (
-	"reflect"
-
-	cda "git.loafle.net/commons/di-go/annotation"
-	cdr "git.loafle.net/commons/di-go/registry"
-	logging "git.loafle.net/commons/logging-go"
-
-	// For annotation
-	_ "git.loafle.net/overflow/commons-go/core/annotation"
-)
-
-var DataServiceType = reflect.TypeOf((*DataService)(nil))
-
-func init() {
-	cdr.RegisterType(DataServiceType)
-}
-
-type DataService struct {
-	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
-
-	DataClientService *DataClientService `annotation:"@Inject()"`
-}
-
-func (s *DataService) InitService() error {
-	return nil
-}
-
-func (s *DataService) StartService() error {
-
-	return nil
-}
-
-func (s *DataService) StopService() {
-
-}
-
-func (s *DataService) DestroyService() {
-
-}
-
-func (s *DataService) Metric(sensorConfigID string, metric map[string]string) error {
-	// s.DataClientService.Send("MS", sensorConfigID, metric)
-	logging.Logger().Debugf("Metric: %v", metric)
-
-	return nil
-}

service/MetricService.go

@@ -0,0 +1,83 @@
+package service
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+
+	cda "git.loafle.net/commons/di-go/annotation"
+	cdr "git.loafle.net/commons/di-go/registry"
+	logging "git.loafle.net/commons/logging-go"
+	csc "git.loafle.net/commons/server-go/client"
+	ocmd "git.loafle.net/overflow/commons-go/model/data"
+	ocsp "git.loafle.net/overflow/commons-go/service/probe"
+	"git.loafle.net/overflow/probe/client/data"
+
+	// For annotation
+	_ "git.loafle.net/overflow/commons-go/core/annotation"
+)
+
+var MetricServiceType = reflect.TypeOf((*MetricService)(nil))
+
+func init() {
+	cdr.RegisterType(MetricServiceType)
+}
+
+type MetricService struct {
+	ocsp.MetricService
+
+	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
+
+	connector csc.Connector
+	readChan  <-chan []byte
+	writeChan chan<- []byte
+}
+
+func (s *MetricService) InitService() error {
+
+	return nil
+}
+
+func (s *MetricService) StartService() error {
+	connector, err := data.New()
+	if nil != err {
+		return fmt.Errorf("MetricService: StartService failed %v", err)
+	}
+	err = connector.Validate()
+	if nil != err {
+		return fmt.Errorf("MetricService: StartService failed %v", err)
+	}
+
+	s.connector = connector
+
+	readChan, writeChan, err := s.connector.Connect()
+	if nil != err {
+		return fmt.Errorf("MetricService: StartService failed %v", err)
+	}
+	s.readChan = readChan
+	s.writeChan = writeChan
+
+	return nil
+}
+
+func (s *MetricService) StopService() {
+	if err := s.connector.Disconnect(); nil != err {
+		logging.Logger().Error(err)
+	}
+}
+
+func (s *MetricService) DestroyService() {
+
+}
+
+func (s *MetricService) Send(metric *ocmd.Metric) error {
+	buff, err := json.Marshal(metric)
+	if nil != err {
+		return err
+	}
+	// s.MetricService.Send("MS", sensorConfigID, metric)
+	// logging.Logger().Debugf("Metric: %v", metric)
+	s.writeChan <- buff
+
+	return nil
+}

service/ProbeClientService.go

@@ -1,14 +1,16 @@
 package service
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 
 	cda "git.loafle.net/commons/di-go/annotation"
 	cdr "git.loafle.net/commons/di-go/registry"
+	logging "git.loafle.net/commons/logging-go"
+	crc "git.loafle.net/commons/rpc-go/client"
 	occp "git.loafle.net/overflow/commons-go/config/probe"
 	"git.loafle.net/overflow/probe/client/probe"
-
 	// For annotation
 	_ "git.loafle.net/overflow/commons-go/core/annotation"
 )
@@ -20,28 +22,21 @@ func init() {
 }
 
 type ProbeClientService struct {
-	RPCClientService
-
 	cda.TypeAnnotation `annotation:"@overflow:RPCService()"`
 
 	DiscoveryService *DiscoveryService `annotation:"@Inject()"`
 
 	EncryptionKey string
+
+	client *crc.Client
 }
 
 func (s *ProbeClientService) InitService() error {
-	if err := s.RPCClientService.InitService(); nil != err {
-		return fmt.Errorf("ProbeClientService: InitService failed %v", err)
-	}
 
 	return nil
 }
 
 func (s *ProbeClientService) StartService() error {
-	if err := s.RPCClientService.StartService(); nil != err {
-		return fmt.Errorf("ProbeClientService: StartService failed %v", err)
-	}
-
 	client, err := probe.New(s.HandleResponse, s.DiscoveryService)
 	if nil != err {
 		return fmt.Errorf("ProbeClientService: StartService failed %v", err)
@@ -56,11 +51,14 @@ func (s *ProbeClientService) StartService() error {
 }
 
 func (s *ProbeClientService) StopService() {
-	s.RPCClientService.StopService()
+	if err := s.client.Stop(context.Background()); nil != err {
+		logging.Logger().Error(err)
+	}
+
 }
 
 func (s *ProbeClientService) DestroyService() {
-	s.RPCClientService.DestroyService()
+
 }
 
 func (s *ProbeClientService) HandleResponse(method string, param string) {
@@ -71,3 +69,17 @@ func (s *ProbeClientService) HandleResponse(method string, param string) {
 	}
 
 }
+
+func (s *ProbeClientService) Call(result interface{}, method string, params ...interface{}) error {
+	if nil == s.client {
+		return fmt.Errorf("rpc client is not valid")
+	}
+	return s.client.Call(result, method, params...)
+}
+
+func (s *ProbeClientService) Send(method string, params ...interface{}) error {
+	if nil == s.client {
+		return fmt.Errorf("rpc client is not valid")
+	}
+	return s.client.Send(method, params...)
+}

service/RPCClientService.go

@@ -1,45 +0,0 @@
-package service
-
-import (
-	"context"
-	"fmt"
-
-	"git.loafle.net/commons/logging-go"
-	crc "git.loafle.net/commons/rpc-go/client"
-)
-
-type RPCClientService struct {
-	client *crc.Client
-}
-
-func (s *RPCClientService) InitService() error {
-	return nil
-}
-
-func (s *RPCClientService) StartService() error {
-	return nil
-}
-
-func (s *RPCClientService) StopService() {
-	if err := s.client.Stop(context.Background()); nil != err {
-		logging.Logger().Error(err)
-	}
-}
-
-func (s *RPCClientService) DestroyService() {
-	s.client = nil
-}
-
-func (s *RPCClientService) Call(result interface{}, method string, params ...interface{}) error {
-	if nil == s.client {
-		return fmt.Errorf("rpc client is not valid")
-	}
-	return s.client.Call(result, method, params...)
-}
-
-func (s *RPCClientService) Send(method string, params ...interface{}) error {
-	if nil == s.client {
-		return fmt.Errorf("rpc client is not valid")
-	}
-	return s.client.Send(method, params...)
-}

service/service.go

@@ -5,7 +5,7 @@ import "reflect"
 var (
 	OrderedServices = []reflect.Type{
 		ProbeClientServiceType,
-		DataClientServiceType,
+		MetricServiceType,
 		ContainerServiceType,
 		SensorConfigServiceType,
 		CrawlerServiceType,