实现tenant service

README.md

@@ -25,4 +25,8 @@
 
 ```bash
 ./run.sh
-```
+```
+
+
+## 开发计划：
+   1.实现规则引擎获取租户、规则链、规则节点相关的数据.

pkg/mqtt/connection.go

@@ -170,7 +170,7 @@ func (c *Connection) RcvMsgFromClient() {
 				return
 			}
 			device := &models.Device{}
-			err = server.RPCCallByName(nil, "registry", "Registry.FindDeviceById", DeviceID, device)
+			err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceById", DeviceID, device)
 			if err != nil {
 				server.Log.Warn("device not found %d", ret, DeviceID)
 				c.Close()

pkg/productconfig/productconfig.go

@@ -138,12 +138,7 @@ func (config *ProductConfig) StatusToMap(status []protocol.SubData) (map[string]
 			}
 		}
 		result[label] = values
-
-		if sub.Head.SubDeviceid == 0 {
-			result["device_id"] = sub.Head.ExternalDeviceId
-		} else {
-			result["device_id"] = sub.Head.SubDeviceid
-		}
+		result["device_id"] = sub.Head.SubDeviceid
 	}
 
 	return result, nil

pkg/productconfig/productconfig_test.go

@@ -50,7 +50,6 @@ func testStatus(c *ProductConfig, t *testing.T) {
 			SubDeviceid: uint16(1),
 			PropertyNum: uint16(1),
 			ParamsCount: uint16(2),
-			ExternalDeviceId: [8]byte{0x1,0x2},
 		},
 		Params: params,
 	}}

pkg/protocol/structure.go

@@ -39,7 +39,6 @@ type SubDataHead struct {
 	SubDeviceid      uint16
 	PropertyNum      uint16
 	ParamsCount      uint16
-	ExternalDeviceId [8]byte // 扩展设备Id为兼容网关类设备
 }
 
 type SubData struct {

pkg/rpcs/common.go

@@ -5,3 +5,9 @@ type ArgsDeviceId struct {
 }
 
 type ReplyEmptyResult struct{}
+
+
+const (
+	RegistryServerName = "registry"
+	AccessServerName = "access"
+)

pkg/rpcs/registry.go

@@ -1,5 +1,7 @@
 package rpcs
 
+
+
 // ArgsDeviceRegister device register args
 type ArgsDeviceRegister struct {
 	ProductKey    string

pkg/rule/ifttt.go

@@ -4,6 +4,7 @@ package rule
 import (
 	"sparrow/pkg/models"
 	"sparrow/pkg/productconfig"
+	"sparrow/pkg/rpcs"
 	"sparrow/pkg/server"
 )
 
@@ -19,7 +20,7 @@ func (ift *Ifttt) Check(deviceid string, eventid uint16) error {
 		RuleType: "ifttt",
 		DeviceID: deviceid,
 	}
-	err := server.RPCCallByName(nil, "registry", "Registry.QueryRules", query, actions)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.QueryRules", query, actions)
 	if err != nil {
 		server.Log.Warnf("load ifttt rules error : %v", err)
 		return err
@@ -27,14 +28,14 @@ func (ift *Ifttt) Check(deviceid string, eventid uint16) error {
 
 	if len(*actions) > 0 {
 		device := &models.Device{}
-		err := server.RPCCallByName(nil, "registry", "Registry.FindDeviceById", deviceid, device)
+		err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceById", deviceid, device)
 		if err != nil {
 			server.Log.Errorf("find device error : %v", err)
 			return err
 		}
 
 		product := &models.Product{}
-		err = server.RPCCallByName(nil, "registry", "Registry.FindProduct", device.ProductID, product)
+		err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindProduct", device.ProductID, product)
 		if err != nil {
 			server.Log.Errorf("find product error : %v", err)
 			return err

pkg/rule/rule_action.go

@@ -20,13 +20,13 @@ func performRuleAction(target string, action string) error {
 
 	identifier := parts[1]
 	device := &models.Device{}
-	err := server.RPCCallByName(nil, "registry", "Registry.FindDeviceByIdentifier", identifier, device)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", identifier, device)
 	if err != nil {
 		return err
 	}
 
 	product := &models.Product{}
-	err = server.RPCCallByName(nil, "registry", "Registry.FindProduct", device.ProductID, product)
+	err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindProduct", device.ProductID, product)
 	if err != nil {
 		return err
 	}

pkg/rule/timer.go

@@ -4,6 +4,7 @@ package rule
 import (
 	"fmt"
 	"sparrow/pkg/models"
+	"sparrow/pkg/rpcs"
 	"sparrow/pkg/server"
 	"time"
 
@@ -38,7 +39,7 @@ func (t *Timer) refresh() {
 	query := &models.Rule{
 		RuleType: "timer",
 	}
-	err := server.RPCCallByName(nil, "registry", "Registry.QueryRules", query, timers)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.QueryRules", query, timers)
 	if err != nil {
 		server.Log.Warnf("refresh timer rules error : %v", err)
 		return

run.sh

@@ -10,7 +10,7 @@ $GOPATH/bin/apiprovider -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -httphost
 $GOPATH/bin/devicemanager -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -rpchost internal:20033 &
 $GOPATH/bin/controller -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -rpchost internal:20032  &
 #$GOPATH/bin/mqttaccess -etcd http://localhost:2379 -loglevel debug  -rpchost localhost:20030 -tcphost internal:1883 -usetls -keyfile $GOPATH/src/github.com/PandoCloud/pando-cloud/pkg/server/testdata/key.pem -cafile $GOPATH/src/github.com/PandoCloud/pando-cloud/pkg/server/testdata/cert.pem &
-$GOPATH/bin/mqttaccess -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -rpchost internal:20030 -tcphost external:1883 -external 192.168.1.100  &
+$GOPATH/bin/mqttaccess -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -rpchost internal:20030 -tcphost internal:1883 &
 $GOPATH/bin/knowoapi -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -httphost internal:8889 -dbhost 39.98.250.155 -dbname gxt-iot-db -dbport 3306 -dbuser root -dbpass gEkYDPloQcp93t4WHr3X -aeskey ABCDEFGHIJKLMNOPABCDEFGHIJKLMNOP &
 $GOPATH/bin/fileaccess -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -rpchost internal:20035 -httphost internal:9000 &
 $GOPATH/bin/coapaccess -etcd http://127.0.0.1:2379 -loglevel $LEVEL  -udphost internal:56883 &

services/apiprovider/actions.go

@@ -69,7 +69,7 @@ func GetDeviceInfoByKey(params martini.Params, req *http.Request, r render.Rende
 	defer span.Finish()
 	ext.SpanKindRPCClient.Set(span)
 	span.SetTag("device_key", key)
-	err := server.RPCCallByName(ctx, "registry", "Registry.ValidateDevice", key, device)
+	err := server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.ValidateDevice", key, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrDeviceNotFound, err))
 		return
@@ -96,7 +96,7 @@ func GetDeviceInfoByIdentifier(urlparams martini.Params, r render.Render) {
 	defer span.Finish()
 	ext.SpanKindRPCClient.Set(span)
 	span.SetTag("identifier", identifier)
-	err := server.RPCCallByName(ctx, "registry", "Registry.FindDeviceByIdentifier", identifier, device)
+	err := server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", identifier, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrDeviceNotFound, err))
 		return
@@ -276,7 +276,7 @@ func AddRule(device *models.Device, req *http.Request, r render.Render) {
 	defer span.Finish()
 	ext.SpanKindRPCClient.Set(span)
 
-	err = server.RPCCallByName(ctx, "registry", "Registry.CreateRule", rule, reply)
+	err = server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.CreateRule", rule, reply)
 	if err != nil {
 		server.Log.Errorf("create device rule error: %v", err)
 		r.JSON(http.StatusOK, renderError(ErrSystemFault, err))
@@ -304,7 +304,7 @@ func AppAuth(req *http.Request, r render.Render) {
 
 	app := &models.Application{}
 
-	err = server.RPCCallByName(ctx, "registry", "Registry.FindApplicationByAppKey", ruleReq, app)
+	err = server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.FindApplicationByAppKey", ruleReq, app)
 	if err != nil {
 
 		r.JSON(http.StatusOK, renderError(ErrWrongSecret, errors.New("Invalid secret key")))

services/apiprovider/middleware.go

@@ -75,7 +75,7 @@ func ApplicationAuthOnDeviceIdentifer(context martini.Context, params martini.Pa
 	}
 
 	app := &models.Application{}
-	err := server.RPCCallByName(nil, "registry", "Registry.ValidateApplication", key, app)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.ValidateApplication", key, app)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrAccessDenied, err))
 		return
@@ -94,7 +94,7 @@ func CheckDeviceOnline(context martini.Context, params martini.Params, req *http
 	identifier := params["identifier"]
 
 	device := &models.Device{}
-	err := server.RPCCallByName(nil, "registry", "Registry.FindDeviceByIdentifier", identifier, device)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", identifier, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrDeviceNotFound, err))
 		return
@@ -119,7 +119,7 @@ func CheckDeviceIdentifier(context martini.Context, params martini.Params, req *
 	identifier := params["identifier"]
 
 	device := &models.Device{}
-	err := server.RPCCallByName(nil, "registry", "Registry.FindDeviceByIdentifier", identifier, device)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", identifier, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrDeviceNotFound, err))
 		return
@@ -132,7 +132,7 @@ func CheckDeviceIdentifier(context martini.Context, params martini.Params, req *
 func CheckProductConfig(context martini.Context, device *models.Device,
 	params martini.Params, req *http.Request, r render.Render) {
 	product := &models.Product{}
-	err := server.RPCCallByName(nil, "registry", "Registry.FindProduct", device.ProductID, product)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindProduct", device.ProductID, product)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrProductNotFound, err))
 		return

services/apiprovider/notifier.go

@@ -64,7 +64,7 @@ func (n *Notifier) processStatus() error {
 func (n *Notifier) updateApplications() error {
 	for {
 
-		err := server.RPCCallByName(nil, "registry", "Registry.GetApplications", 0, &n.apps)
+		err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.GetApplications", 0, &n.apps)
 		if err != nil {
 			server.Log.Errorf("get applications error : %v", err)
 		}
@@ -77,14 +77,14 @@ func (n *Notifier) reportEvent(event rpcs.ArgsOnEvent) error {
 	server.Log.Debugf("reporting event %v", event)
 
 	device := &models.Device{}
-	err := server.RPCCallByName(nil, "registry", "Registry.FindDeviceById", event.DeviceId, device)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceById", event.DeviceId, device)
 	if err != nil {
 		server.Log.Errorf("find device error : %v", err)
 		return err
 	}
 
 	product := &models.Product{}
-	err = server.RPCCallByName(nil, "registry", "Registry.FindProduct", device.ProductID, product)
+	err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindProduct", device.ProductID, product)
 	if err != nil {
 		server.Log.Errorf("find product error : %v", err)
 		return err

services/controller/controller.go

@@ -103,14 +103,14 @@ func (c *Controller) OnStatus(args rpcs.ArgsOnStatus, reply *rpcs.ReplyOnStatus)
 func (c *Controller) processStatusToQueue(args rpcs.ArgsOnStatus) (string, error) {
 	var result string
 	device := &models.Device{}
-	err := server.RPCCallByName(nil, "registry", "Registry.FindDeviceByRecordId", &rpcs.ArgsDeviceAuth{DeviceID: args.DeviceId}, device)
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindDeviceByRecordId", &rpcs.ArgsDeviceAuth{DeviceID: args.DeviceId}, device)
 	if err != nil {
 		server.Log.Errorf("find device error : %v", err)
 		return result, err
 	}
 
 	product := &models.Product{}
-	err = server.RPCCallByName(nil, "registry", "Registry.FindProduct", &device.ProductID, product)
+	err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.FindProduct", &device.ProductID, product)
 	if err != nil {
 		server.Log.Errorf("find product error : %v", err)
 		return result, err

services/controller/service.go

@@ -1,11 +1,13 @@
 package main
 
 import (
+	"context"
 	"github.com/gogf/gf/os/grpool"
 	"sparrow/pkg/models"
 	"sparrow/pkg/protocol"
 	"sparrow/pkg/queue"
 	"sparrow/pkg/rpcs"
+	"sparrow/pkg/ruleEngine"
 	"sparrow/pkg/server"
 )
 
@@ -35,38 +37,45 @@ func NewEventService() *EventService {
 }
 
 func (e *EventService) Save(data *models.Event) error {
-	return server.RPCCallByName(nil, "registry", "Registry.CreateEvent", data, &rpcs.ReplyEmptyResult{})
+	return server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.CreateEvent", data, &rpcs.ReplyEmptyResult{})
 }
 
 func (e *EventService) SaveAsync(data *models.Event) error {
 	return e.pool.Add(func() {
-		if err := server.RPCCallByName(nil, "registry", "Registry.CreateEvent", data, &rpcs.ReplyEmptyResult{}); err != nil {
+		if err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.CreateEvent", data, &rpcs.ReplyEmptyResult{}); err != nil {
 			return
 		}
 	})
 }
 
-//type TenantService struct {
-//
-//}
-//
-//func (t *TenantService) FindTenants() ([]*ruleEngine.Tenant, error) {
-//	var arg int
-//	var reply []*models.Vendor
-//	var result []*ruleEngine.Tenant
-//	err :=server.RPCCallByName(context.Background(), "registry", "registry.GetVendors", &arg, &reply)
-//	if err != nil {
-//		return nil, err
-//	}
-//	for _, vendor := range reply {
-//		result = append(result, &ruleEngine.Tenant{
-//			Id: vendor.RecordId,
-//		})
-//	}
-//	return result, nil
-//}
-//
-//func (t *TenantService) GetTenant(tId string) (*ruleEngine.Tenant, error) {
-//	var reply *models.Vendor
-//	err := server.
-//}
+// TODO:完成这里
+type TenantService struct {
+}
+
+func (t *TenantService) FindTenants() ([]*ruleEngine.Tenant, error) {
+	var arg int
+	var reply []*models.Vendor
+	var result []*ruleEngine.Tenant
+	err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.GetVendors", &arg, &reply)
+	if err != nil {
+		return nil, err
+	}
+	for _, vendor := range reply {
+		result = append(result, &ruleEngine.Tenant{
+			Id: vendor.RecordId,
+		})
+	}
+	return result, nil
+}
+
+func (t *TenantService) GetTenant(tId string) (*ruleEngine.Tenant, error) {
+	var reply *models.Vendor
+	err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.FindVendor", &tId, &reply)
+	if err != nil {
+		return nil, err
+	}
+	return &ruleEngine.Tenant{
+		Id:   reply.RecordId,
+		Name: reply.VendorName,
+	}, nil
+}

services/httpaccess/actions.go

@@ -63,7 +63,7 @@ func RegisterDevice(args DeviceRegisterArgs, r render.Render) {
 	}
 	device := &models.Device{}
 	ext.SpanKindRPCClient.Set(span)
-	err := server.RPCCallByName(ctx, "registry", "Registry.RegisterDevice", rpcargs, device)
+	err := server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.RegisterDevice", rpcargs, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrSystemFault, err))
 		return
@@ -91,7 +91,7 @@ func AuthDevice(args DeviceAuthArgs, r render.Render) {
 
 	ext.SpanKindRPCClient.Set(span)
 	arg := uint64(args.DeviceId)
-	err := server.RPCCallByName(ctx, "registry", "Registry.FindDeviceById", &arg, device)
+	err := server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.FindDeviceById", &arg, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrDeviceNotFound, err))
 		return

services/registry/main.go

@@ -2,12 +2,13 @@ package main
 
 import (
 	"sparrow/pkg/mysql"
+	"sparrow/pkg/rpcs"
 	"sparrow/pkg/server"
 )
 
 func main() {
 	// init server
-	err := server.Init("registry")
+	err := server.Init(rpcs.RegistryServerName)
 	if err != nil {
 		server.Log.Fatal(err)
 		return

tests/device/device.go

@@ -182,7 +182,6 @@ func (d *Device) reportStatus(client *MQTT.Client) {
 				SubDeviceid: uint16(225),
 				PropertyNum: uint16(1),
 				ParamsCount: uint16(len(params)),
-				ExternalDeviceId: [8]byte{0x01},
 			},
 			Params: params,
 		}

tests/device/main.go

@@ -6,7 +6,7 @@ import (
 )
 
 var (
-	testURL        = flag.String("url", "http://192.168.1.112:8088", "login url")
+	testURL        = flag.String("url", "http://192.168.1.100:8088", "login url")
 	testProductKey = flag.String("productkey", "2e397f5599a3f6f6a5a3c8fcd45437169501b3c6e239042ad5e9b65303561e41ab5519e9d205facbcbe75a2784354501", "product key")
 	testProtocol   = flag.String("protocol", "mqtt", "access protocol")
 )
@@ -19,7 +19,7 @@ func main() {
 		return
 	}
 
-	dev := NewDevice(*testURL, *testProductKey, "3344", "1.25", *testProtocol)
+	dev := NewDevice(*testURL, *testProductKey, "5566", "1.2.5", *testProtocol)
 
 	err := dev.DoRegister()
 	if err != nil {

tools/pdcfg/application.go

@@ -54,7 +54,7 @@ func addApplication() error {
 
 	reply := &models.Application{}
 
-	err = server.RPCCallByName(nil, "registry", "Registry.SaveApplication", &args, reply)
+	err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.SaveApplication", &args, reply)
 	if err != nil {
 		return err
 	}

tools/pdcfg/product.go

@@ -66,7 +66,7 @@ func addProduct() error {
 
 	reply := &models.Product{}
 
-	err = server.RPCCallByName(nil, "registry", "Registry.SaveProduct", &args, reply)
+	err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.SaveProduct", &args, reply)
 	if err != nil {
 		return err
 	}

tools/pdcfg/vendor.go

@@ -31,7 +31,7 @@ func addVendor() error {
 
 	reply := &models.Vendor{}
 
-	err = server.RPCCallByName(nil, "registry", "Registry.SaveVendor", &args, reply)
+	err = server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.SaveVendor", &args, reply)
 	if err != nil {
 		return err
 	}