overflow_cloud/probe_gateway_metric
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-05-11 17:24:12 +09:00
parent 82625bdaae
commit 71acf24d9c
3 changed files with 21 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
glide.yaml
View File

@ -22,3 +22,4 @@ import:
- server - server
- package: github.com/valyala/fasthttp - package: github.com/valyala/fasthttp
version: ^20160617.0.0 version: ^20160617.0.0
- package: git.loafle.net/overflow/external-go

10
server/server-handler.go
View File

@ -2,7 +2,7 @@ package server
import ( import (
cs "git.loafle.net/commons/server-go" cs "git.loafle.net/commons/server-go"
oge "git.loafle.net/overflow/gateway/external" oe "git.loafle.net/overflow/external-go"
ogrs "git.loafle.net/overflow/gateway_rpc/server" ogrs "git.loafle.net/overflow/gateway_rpc/server"
"git.loafle.net/overflow/probe_gateway_metric/config" "git.loafle.net/overflow/probe_gateway_metric/config"
) )
@ -21,7 +21,7 @@ func (sh *ServerHandlers) Init(serverCtx cs.ServerCtx) error {
if err := sh.ServerHandlers.Init(serverCtx); nil != err { if err := sh.ServerHandlers.Init(serverCtx); nil != err {
return err return err
} }
oge.InitPackage(sh.Config.External) oe.InitPackage(sh.Config.External)
return nil return nil
} }
@ -31,19 +31,19 @@ func (sh *ServerHandlers) OnStart(serverCtx cs.ServerCtx) error {
return err return err
} }
oge.StartPackage(sh.Config.External) oe.StartPackage(sh.Config.External)
return nil return nil
} }
func (sh *ServerHandlers) OnStop(serverCtx cs.ServerCtx) { func (sh *ServerHandlers) OnStop(serverCtx cs.ServerCtx) {
oge.StopPackage(sh.Config.External) oe.StopPackage(sh.Config.External)
sh.ServerHandlers.OnStop(serverCtx) sh.ServerHandlers.OnStop(serverCtx)
} }
func (sh *ServerHandlers) Destroy(serverCtx cs.ServerCtx) { func (sh *ServerHandlers) Destroy(serverCtx cs.ServerCtx) {
oge.DestroyPackage(sh.Config.External) oe.DestroyPackage(sh.Config.External)
sh.ServerHandlers.Destroy(serverCtx) sh.ServerHandlers.Destroy(serverCtx)
} }

22
servlet/data-servlet.go
View File

@ -2,6 +2,7 @@ package servlet
import ( import (
"context" "context"
"encoding/json"
"fmt" "fmt"
"github.com/valyala/fasthttp" "github.com/valyala/fasthttp"
@ -11,9 +12,11 @@ import (
"git.loafle.net/commons/server-go/socket" "git.loafle.net/commons/server-go/socket"
cssw "git.loafle.net/commons/server-go/socket/web" cssw "git.loafle.net/commons/server-go/socket/web"
occp "git.loafle.net/overflow/commons-go/config/probe" occp "git.loafle.net/overflow/commons-go/config/probe"
ocmd "git.loafle.net/overflow/commons-go/model/data"
oe "git.loafle.net/overflow/external-go"
oeg "git.loafle.net/overflow/external-go/grpc"
oek "git.loafle.net/overflow/external-go/kafka"
og "git.loafle.net/overflow/gateway" og "git.loafle.net/overflow/gateway"
"git.loafle.net/overflow/gateway/external/grpc"
"git.loafle.net/overflow/gateway/external/kafka"
) )
type DataServlet interface { type DataServlet interface {
@ -73,13 +76,13 @@ func (s *DataServlets) Handshake(servletCtx server.ServletCtx, ctx *fasthttp.Req
probeKey := string(bProbeKey) probeKey := string(bProbeKey)
grpcCTX := context.Background() grpcCTX := context.Background()
_, err := grpc.Exec(grpcCTX, "ProbeService.readByProbeKey", probeKey) _, err := oeg.Exec(grpcCTX, "ProbeService.readByProbeKey", probeKey)
if nil != err { if nil != err {
return nil, fmt.Errorf("grpc call Error: %s", err.Error()) return nil, fmt.Errorf("grpc call Error: %s", err.Error())
} }
servletCtx.SetAttribute(og.SessionIDKey, probeKey) servletCtx.SetAttribute(og.SessionIDKey, probeKey)
servletCtx.SetAttribute(og.SessionClientTypeKey, og.PROBE) servletCtx.SetAttribute(og.SessionClientTypeKey, oe.PROBE)
servletCtx.SetAttribute(og.SessionTargetIDKey, probeKey) servletCtx.SetAttribute(og.SessionTargetIDKey, probeKey)
return nil, nil return nil, nil
@ -111,7 +114,14 @@ func (s *DataServlets) Handle(servletCtx server.ServletCtx,
return return
} }
err = kafka.Write("metric", msg, msg) m := &ocmd.Metric{}
err = json.Unmarshal(msg, m)
if nil != err {
logging.Logger().Error(err)
break
}
err = oek.Write("metric", []byte(m.SensorConfigID), msg)
if nil != err { if nil != err {
logging.Logger().Error(err) logging.Logger().Error(err)
break break
@ -120,7 +130,5 @@ func (s *DataServlets) Handle(servletCtx server.ServletCtx,
case <-stopChan: case <-stopChan:
return return
} }
} }
} }