property-device/pb/setup.go

// Copyright 2019 getensh.com. All rights reserved.
// Use of this source code is governed by getensh.com.

package pb

import (
	fmt "fmt"
	"property-device/parser"
	"time"

	"go.etcd.io/etcd/client/v3/naming/resolver"
	grpc "google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

// 客户端集合
var System SystemClient
var Garden GardenClient
var Mqtt MqttClient
var Common CommonClient

func setupMqttClient(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string
	if parser.Conf.K8s {
		serviceName = parser.Conf.Rpc.Mqtt.ServiceName
	} else {
		serviceName = parser.Conf.Rpc.Mqtt.ServiceIp
	}

	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial(fmt.Sprintf("%s:%d", serviceName,
		parser.Conf.Rpc.Mqtt.ServicePort),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		Mqtt = NewMqttClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial mqtt conn err", err)
	}

	return
}

func setupCommonClient(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string
	if parser.Conf.K8s {
		serviceName = parser.Conf.Rpc.Common.ServiceName
	} else {
		serviceName = parser.Conf.Rpc.Common.ServiceIp
	}

	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial(fmt.Sprintf("%s:%d", serviceName,
		parser.Conf.Rpc.Common.ServicePort),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		Common = NewCommonClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial Common conn err", err)
	}

	return
}

func setupSystemClient(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string
	if parser.Conf.K8s {
		serviceName = parser.Conf.Rpc.System.ServiceName
	} else {
		serviceName = parser.Conf.Rpc.System.ServiceIp
	}

	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial(fmt.Sprintf("%s:%d", serviceName,
		parser.Conf.Rpc.System.ServicePort),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		System = NewSystemClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial system conn err", err)
	}

	return
}

func setupGardenClient(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string
	if parser.Conf.K8s {
		serviceName = parser.Conf.Rpc.Garden.ServiceName
	} else {
		serviceName = parser.Conf.Rpc.Garden.ServiceIp
	}

	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial(fmt.Sprintf("%s:%d", serviceName,
		parser.Conf.Rpc.Garden.ServicePort),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		Garden = NewGardenClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial garden conn err", err)
	}

	return
}

func setupSystemClient2(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string

	serviceName = parser.Conf.Rpc.System.ServiceName

	builder, err := resolver.NewBuilder(parser.GetEtcdClient())
	if err != nil {
		panic(err)
	}
	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial("etcd:///"+parser.Conf.Rpc.Prefix+"/"+serviceName,
		grpc.WithResolvers(builder),
		grpc.WithBalancerName("round_robin"),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		System = NewSystemClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial system conn err", err)
	}

	return
}

func setupCommonClient2(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string

	serviceName = parser.Conf.Rpc.Common.ServiceName

	builder, err := resolver.NewBuilder(parser.GetEtcdClient())
	if err != nil {
		panic(err)
	}
	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial("etcd:///"+parser.Conf.Rpc.Prefix+"/"+serviceName,
		grpc.WithResolvers(builder),
		grpc.WithBalancerName("round_robin"),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		Common = NewCommonClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial Common conn err", err)
	}

	return
}

func setupMqttClient2(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string

	serviceName = parser.Conf.Rpc.Mqtt.ServiceName

	builder, err := resolver.NewBuilder(parser.GetEtcdClient())
	if err != nil {
		panic(err)
	}
	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial("etcd:///"+parser.Conf.Rpc.Prefix+"/"+serviceName,
		grpc.WithResolvers(builder),
		grpc.WithBalancerName("round_robin"),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		Mqtt = NewMqttClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial mqtt conn err", err)
	}

	return
}

func setupGardenClient2(kacp keepalive.ClientParameters, conns []*grpc.ClientConn) {
	// 根据是否为k8s来组装targets
	var serviceName string

	serviceName = parser.Conf.Rpc.Garden.ServiceName

	builder, err := resolver.NewBuilder(parser.GetEtcdClient())
	if err != nil {
		panic(err)
	}
	// 发起一个连接并记录连接conn，后期释放
	if conn, err := grpc.Dial("etcd:///"+parser.Conf.Rpc.Prefix+"/"+serviceName,
		grpc.WithResolvers(builder),
		grpc.WithBalancerName("round_robin"),
		grpc.WithInsecure(), grpc.WithKeepaliveParams(kacp)); err == nil {
		Garden = NewGardenClient(conn)
		conns = append(conns, conn)
	} else {
		fmt.Println("[rpc] dial Garden conn err", err)
	}

	return
}

// SetupClients 创建客户端
func SetupClients() (conns []*grpc.ClientConn) {
	// 客户端配置参数
	var kacp = keepalive.ClientParameters{
		// send pings every n seconds if there is no activity
		Time: time.Duration(parser.Conf.Rpc.Keepalive.ClientTime) * time.Second,
		// wait n second for ping ack before considering the connection dead
		Timeout: time.Duration(parser.Conf.Rpc.Keepalive.ClientTimeout) * time.Second,
		// send pings even without active streams
		PermitWithoutStream: true,
	}
	if len(parser.Conf.EtcdAddrs) == 0 {
		setupSystemClient(kacp, conns)
		setupGardenClient(kacp, conns)
		setupMqttClient(kacp, conns)
		setupCommonClient(kacp, conns)
		return
	}
	setupSystemClient2(kacp, conns)
	setupGardenClient2(kacp, conns)
	setupMqttClient2(kacp, conns)
	setupCommonClient2(kacp, conns)
	return
}