refactored a lot of server, creating actual database package
KeeganForelight
2 years ago
parent
72fe309a25
commit
f209980e8b
@ -0,0 +1,64 @@
|
||||
// package Database wraps some influx db methods to provide functionality.
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
influx "github.com/influxdata/influxdb-client-go/v2"
|
||||
"github.com/spf13/viper"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrDBConnection = errors.New("connection to database failed")
|
||||
ErrNoURLFound = errors.New("database url not found")
|
||||
)
|
||||
|
||||
var db influx.Client
|
||||
|
||||
// Connect takes in a config and attempts to create a client for influxdb.
|
||||
// Will automatically write changes back to config for future attempts
|
||||
func Connect(config *viper.Viper) error {
|
||||
|
||||
url := config.GetString("db.url")
|
||||
token := config.GetString("db.token")
|
||||
|
||||
if url == "" {
|
||||
return ErrNoURLFound
|
||||
}
|
||||
|
||||
db = influx.NewClient(url, token)
|
||||
|
||||
if token == "" {
|
||||
// try setup
|
||||
fmt.Printf("attempting to setup database at %v\n", url)
|
||||
|
||||
user := config.GetString("db.username")
|
||||
password := config.GetString("db.password")
|
||||
org := config.GetString("db.org")
|
||||
bucket := config.GetString("db.bucket")
|
||||
|
||||
Setup(user, pass, org, bucket
|
||||
}
|
||||
|
||||
db = influx.NewClient(url, token)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func Setup(user, pass, org, bucket string, ret int) (string, error) {
|
||||
|
||||
resp, err := db.Setup(context.Background(), user, pass, org, bucket, ret)
|
||||
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func GetBucket(id int) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func GetToken(id int) (string, error) {
|
||||
// bucket, err := client.BucketsAPI().FindBucketByName(context.Background(), id)
|
||||
return "", nil
|
||||
}
|
||||
@ -1,42 +0,0 @@
|
||||
syntax = "proto3";
|
||||
package grpc;
|
||||
|
||||
option go_package = "internal/pkg/grpc";
|
||||
|
||||
service device {
|
||||
// groups basic device interactions
|
||||
// get/set name based on request
|
||||
rpc Name(NameRequest) returns (NameResponse)
|
||||
}
|
||||
|
||||
message NameRequest {
|
||||
// empty for future expansion
|
||||
string Name = 1;
|
||||
}
|
||||
|
||||
message NameResponse {
|
||||
string Name = 1;
|
||||
}
|
||||
|
||||
service sensor {
|
||||
// sensor specific functions
|
||||
rpc Reading(ReadingRequest) returns (ReadingResponse)
|
||||
rpc SampleRate(SampleRateRequest) returns (SampleRateResponse)
|
||||
}
|
||||
|
||||
message ReadingRequest {
|
||||
// empty
|
||||
}
|
||||
|
||||
message ReadingResponse {
|
||||
string Reading = 1; // formatted reading "9.7 pH"
|
||||
int64 Timestamp = 2; // when the reading was taken
|
||||
}
|
||||
|
||||
message SampleRateRequest {
|
||||
int32 SampleRate = 1; // 0 to return current sample rate, value in seconds
|
||||
}
|
||||
|
||||
message SampleRateResponse {
|
||||
int32 SampleRate = 1; // returns the set sample rate
|
||||
}
|
||||
@ -0,0 +1,260 @@
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.28.1
|
||||
// protoc v3.21.12
|
||||
// source: internal/pkg/grpc/handshake.proto
|
||||
|
||||
package grpc
|
||||
|
||||
import (
|
||||
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
|
||||
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
|
||||
reflect "reflect"
|
||||
sync "sync"
|
||||
)
|
||||
|
||||
const (
|
||||
// Verify that this generated code is sufficiently up-to-date.
|
||||
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
|
||||
// Verify that runtime/protoimpl is sufficiently up-to-date.
|
||||
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
|
||||
)
|
||||
|
||||
type ReactorClientRequest struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
|
||||
Port uint32 `protobuf:"varint,2,opt,name=port,proto3" json:"port,omitempty"` // client gRPC port
|
||||
}
|
||||
|
||||
func (x *ReactorClientRequest) Reset() {
|
||||
*x = ReactorClientRequest{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_internal_pkg_grpc_handshake_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *ReactorClientRequest) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*ReactorClientRequest) ProtoMessage() {}
|
||||
|
||||
func (x *ReactorClientRequest) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_internal_pkg_grpc_handshake_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use ReactorClientRequest.ProtoReflect.Descriptor instead.
|
||||
func (*ReactorClientRequest) Descriptor() ([]byte, []int) {
|
||||
return file_internal_pkg_grpc_handshake_proto_rawDescGZIP(), []int{0}
|
||||
}
|
||||
|
||||
func (x *ReactorClientRequest) GetId() uint32 {
|
||||
if x != nil {
|
||||
return x.Id
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ReactorClientRequest) GetPort() uint32 {
|
||||
if x != nil {
|
||||
return x.Port
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type ReactorClientResponse struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
|
||||
Url string `protobuf:"bytes,2,opt,name=url,proto3" json:"url,omitempty"`
|
||||
Org string `protobuf:"bytes,3,opt,name=org,proto3" json:"org,omitempty"`
|
||||
Token string `protobuf:"bytes,4,opt,name=token,proto3" json:"token,omitempty"`
|
||||
Bucket string `protobuf:"bytes,5,opt,name=bucket,proto3" json:"bucket,omitempty"`
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) Reset() {
|
||||
*x = ReactorClientResponse{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_internal_pkg_grpc_handshake_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*ReactorClientResponse) ProtoMessage() {}
|
||||
|
||||
func (x *ReactorClientResponse) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_internal_pkg_grpc_handshake_proto_msgTypes[1]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use ReactorClientResponse.ProtoReflect.Descriptor instead.
|
||||
func (*ReactorClientResponse) Descriptor() ([]byte, []int) {
|
||||
return file_internal_pkg_grpc_handshake_proto_rawDescGZIP(), []int{1}
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) GetId() uint32 {
|
||||
if x != nil {
|
||||
return x.Id
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) GetUrl() string {
|
||||
if x != nil {
|
||||
return x.Url
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) GetOrg() string {
|
||||
if x != nil {
|
||||
return x.Org
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) GetToken() string {
|
||||
if x != nil {
|
||||
return x.Token
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *ReactorClientResponse) GetBucket() string {
|
||||
if x != nil {
|
||||
return x.Bucket
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
var File_internal_pkg_grpc_handshake_proto protoreflect.FileDescriptor
|
||||
|
||||
var file_internal_pkg_grpc_handshake_proto_rawDesc = []byte{
|
||||
0x0a, 0x21, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x67,
|
||||
0x72, 0x70, 0x63, 0x2f, 0x68, 0x61, 0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x2e, 0x70, 0x72,
|
||||
0x6f, 0x74, 0x6f, 0x12, 0x04, 0x67, 0x72, 0x70, 0x63, 0x22, 0x3a, 0x0a, 0x14, 0x52, 0x65, 0x61,
|
||||
0x63, 0x74, 0x6f, 0x72, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
|
||||
0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69,
|
||||
0x64, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52,
|
||||
0x04, 0x70, 0x6f, 0x72, 0x74, 0x22, 0x79, 0x0a, 0x15, 0x52, 0x65, 0x61, 0x63, 0x74, 0x6f, 0x72,
|
||||
0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e,
|
||||
0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x12, 0x10,
|
||||
0x0a, 0x03, 0x75, 0x72, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c,
|
||||
0x12, 0x10, 0x0a, 0x03, 0x6f, 0x72, 0x67, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6f,
|
||||
0x72, 0x67, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28,
|
||||
0x09, 0x52, 0x05, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x75, 0x63, 0x6b,
|
||||
0x65, 0x74, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74,
|
||||
0x32, 0x5c, 0x0a, 0x09, 0x68, 0x61, 0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x12, 0x4f, 0x0a,
|
||||
0x14, 0x52, 0x65, 0x61, 0x63, 0x74, 0x6f, 0x72, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x48, 0x61,
|
||||
0x6e, 0x64, 0x6c, 0x65, 0x72, 0x12, 0x1a, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x52, 0x65, 0x61,
|
||||
0x63, 0x74, 0x6f, 0x72, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
|
||||
0x74, 0x1a, 0x1b, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x52, 0x65, 0x61, 0x63, 0x74, 0x6f, 0x72,
|
||||
0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x13,
|
||||
0x5a, 0x11, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x67,
|
||||
0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var (
|
||||
file_internal_pkg_grpc_handshake_proto_rawDescOnce sync.Once
|
||||
file_internal_pkg_grpc_handshake_proto_rawDescData = file_internal_pkg_grpc_handshake_proto_rawDesc
|
||||
)
|
||||
|
||||
func file_internal_pkg_grpc_handshake_proto_rawDescGZIP() []byte {
|
||||
file_internal_pkg_grpc_handshake_proto_rawDescOnce.Do(func() {
|
||||
file_internal_pkg_grpc_handshake_proto_rawDescData = protoimpl.X.CompressGZIP(file_internal_pkg_grpc_handshake_proto_rawDescData)
|
||||
})
|
||||
return file_internal_pkg_grpc_handshake_proto_rawDescData
|
||||
}
|
||||
|
||||
var file_internal_pkg_grpc_handshake_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
|
||||
var file_internal_pkg_grpc_handshake_proto_goTypes = []interface{}{
|
||||
(*ReactorClientRequest)(nil), // 0: grpc.ReactorClientRequest
|
||||
(*ReactorClientResponse)(nil), // 1: grpc.ReactorClientResponse
|
||||
}
|
||||
var file_internal_pkg_grpc_handshake_proto_depIdxs = []int32{
|
||||
0, // 0: grpc.handshake.ReactorClientHandler:input_type -> grpc.ReactorClientRequest
|
||||
1, // 1: grpc.handshake.ReactorClientHandler:output_type -> grpc.ReactorClientResponse
|
||||
1, // [1:2] is the sub-list for method output_type
|
||||
0, // [0:1] is the sub-list for method input_type
|
||||
0, // [0:0] is the sub-list for extension type_name
|
||||
0, // [0:0] is the sub-list for extension extendee
|
||||
0, // [0:0] is the sub-list for field type_name
|
||||
}
|
||||
|
||||
func init() { file_internal_pkg_grpc_handshake_proto_init() }
|
||||
func file_internal_pkg_grpc_handshake_proto_init() {
|
||||
if File_internal_pkg_grpc_handshake_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_internal_pkg_grpc_handshake_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*ReactorClientRequest); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_internal_pkg_grpc_handshake_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*ReactorClientResponse); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
|
||||
RawDescriptor: file_internal_pkg_grpc_handshake_proto_rawDesc,
|
||||
NumEnums: 0,
|
||||
NumMessages: 2,
|
||||
NumExtensions: 0,
|
||||
NumServices: 1,
|
||||
},
|
||||
GoTypes: file_internal_pkg_grpc_handshake_proto_goTypes,
|
||||
DependencyIndexes: file_internal_pkg_grpc_handshake_proto_depIdxs,
|
||||
MessageInfos: file_internal_pkg_grpc_handshake_proto_msgTypes,
|
||||
}.Build()
|
||||
File_internal_pkg_grpc_handshake_proto = out.File
|
||||
file_internal_pkg_grpc_handshake_proto_rawDesc = nil
|
||||
file_internal_pkg_grpc_handshake_proto_goTypes = nil
|
||||
file_internal_pkg_grpc_handshake_proto_depIdxs = nil
|
||||
}
|
||||
@ -0,0 +1,21 @@
|
||||
syntax = "proto3";
|
||||
package grpc;
|
||||
|
||||
option go_package = "internal/pkg/grpc";
|
||||
|
||||
service handshake {
|
||||
rpc ReactorClientHandler(ReactorClientRequest) returns (ReactorClientResponse);
|
||||
}
|
||||
|
||||
message ReactorClientRequest {
|
||||
uint32 id = 1;
|
||||
uint32 port = 2; // client gRPC port
|
||||
}
|
||||
|
||||
message ReactorClientResponse {
|
||||
uint32 id = 1;
|
||||
string url = 2;
|
||||
string org = 3;
|
||||
string token = 4;
|
||||
string bucket = 5;
|
||||
}
|
||||
@ -1,335 +0,0 @@
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.28.0
|
||||
// protoc v3.6.1
|
||||
// source: internal/pkg/grpc/server.proto
|
||||
|
||||
package grpc
|
||||
|
||||
import (
|
||||
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
|
||||
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
|
||||
reflect "reflect"
|
||||
sync "sync"
|
||||
)
|
||||
|
||||
const (
|
||||
// Verify that this generated code is sufficiently up-to-date.
|
||||
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
|
||||
// Verify that runtime/protoimpl is sufficiently up-to-date.
|
||||
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
|
||||
)
|
||||
|
||||
type ClientRequest struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
ClientId uint32 `protobuf:"varint,1,opt,name=clientId,proto3" json:"clientId,omitempty"`
|
||||
ClientType string `protobuf:"bytes,2,opt,name=clientType,proto3" json:"clientType,omitempty"`
|
||||
}
|
||||
|
||||
func (x *ClientRequest) Reset() {
|
||||
*x = ClientRequest{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_internal_pkg_grpc_server_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *ClientRequest) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*ClientRequest) ProtoMessage() {}
|
||||
|
||||
func (x *ClientRequest) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_internal_pkg_grpc_server_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use ClientRequest.ProtoReflect.Descriptor instead.
|
||||
func (*ClientRequest) Descriptor() ([]byte, []int) {
|
||||
return file_internal_pkg_grpc_server_proto_rawDescGZIP(), []int{0}
|
||||
}
|
||||
|
||||
func (x *ClientRequest) GetClientId() uint32 {
|
||||
if x != nil {
|
||||
return x.ClientId
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ClientRequest) GetClientType() string {
|
||||
if x != nil {
|
||||
return x.ClientType
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type ClientResponse struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
ClientId uint32 `protobuf:"varint,1,opt,name=clientId,proto3" json:"clientId,omitempty"`
|
||||
ServerPort uint32 `protobuf:"varint,2,opt,name=serverPort,proto3" json:"serverPort,omitempty"`
|
||||
Database *Database `protobuf:"bytes,3,opt,name=database,proto3" json:"database,omitempty"`
|
||||
}
|
||||
|
||||
func (x *ClientResponse) Reset() {
|
||||
*x = ClientResponse{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_internal_pkg_grpc_server_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *ClientResponse) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*ClientResponse) ProtoMessage() {}
|
||||
|
||||
func (x *ClientResponse) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_internal_pkg_grpc_server_proto_msgTypes[1]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use ClientResponse.ProtoReflect.Descriptor instead.
|
||||
func (*ClientResponse) Descriptor() ([]byte, []int) {
|
||||
return file_internal_pkg_grpc_server_proto_rawDescGZIP(), []int{1}
|
||||
}
|
||||
|
||||
func (x *ClientResponse) GetClientId() uint32 {
|
||||
if x != nil {
|
||||
return x.ClientId
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ClientResponse) GetServerPort() uint32 {
|
||||
if x != nil {
|
||||
return x.ServerPort
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ClientResponse) GetDatabase() *Database {
|
||||
if x != nil {
|
||||
return x.Database
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type Database struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
URL string `protobuf:"bytes,1,opt,name=URL,proto3" json:"URL,omitempty"`
|
||||
ORG string `protobuf:"bytes,2,opt,name=ORG,proto3" json:"ORG,omitempty"`
|
||||
Token string `protobuf:"bytes,3,opt,name=token,proto3" json:"token,omitempty"`
|
||||
Bucket string `protobuf:"bytes,4,opt,name=bucket,proto3" json:"bucket,omitempty"`
|
||||
}
|
||||
|
||||
func (x *Database) Reset() {
|
||||
*x = Database{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_internal_pkg_grpc_server_proto_msgTypes[2]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *Database) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*Database) ProtoMessage() {}
|
||||
|
||||
func (x *Database) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_internal_pkg_grpc_server_proto_msgTypes[2]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use Database.ProtoReflect.Descriptor instead.
|
||||
func (*Database) Descriptor() ([]byte, []int) {
|
||||
return file_internal_pkg_grpc_server_proto_rawDescGZIP(), []int{2}
|
||||
}
|
||||
|
||||
func (x *Database) GetURL() string {
|
||||
if x != nil {
|
||||
return x.URL
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *Database) GetORG() string {
|
||||
if x != nil {
|
||||
return x.ORG
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *Database) GetToken() string {
|
||||
if x != nil {
|
||||
return x.Token
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *Database) GetBucket() string {
|
||||
if x != nil {
|
||||
return x.Bucket
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
var File_internal_pkg_grpc_server_proto protoreflect.FileDescriptor
|
||||
|
||||
var file_internal_pkg_grpc_server_proto_rawDesc = []byte{
|
||||
0x0a, 0x1e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x67,
|
||||
0x72, 0x70, 0x63, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
|
||||
0x12, 0x04, 0x67, 0x72, 0x70, 0x63, 0x22, 0x4b, 0x0a, 0x0d, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74,
|
||||
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e,
|
||||
0x74, 0x49, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e,
|
||||
0x74, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70,
|
||||
0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54,
|
||||
0x79, 0x70, 0x65, 0x22, 0x78, 0x0a, 0x0e, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x73,
|
||||
0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49,
|
||||
0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49,
|
||||
0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x50, 0x6f, 0x72, 0x74, 0x18,
|
||||
0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x50, 0x6f, 0x72,
|
||||
0x74, 0x12, 0x2a, 0x0a, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x18, 0x03, 0x20,
|
||||
0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x62,
|
||||
0x61, 0x73, 0x65, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x22, 0x5c, 0x0a,
|
||||
0x08, 0x44, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x55, 0x52, 0x4c,
|
||||
0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x55, 0x52, 0x4c, 0x12, 0x10, 0x0a, 0x03, 0x4f,
|
||||
0x52, 0x47, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x4f, 0x52, 0x47, 0x12, 0x14, 0x0a,
|
||||
0x05, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f,
|
||||
0x6b, 0x65, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x18, 0x04, 0x20,
|
||||
0x01, 0x28, 0x09, 0x52, 0x06, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x32, 0x50, 0x0a, 0x09, 0x68,
|
||||
0x61, 0x6e, 0x64, 0x73, 0x68, 0x61, 0x6b, 0x65, 0x12, 0x43, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65,
|
||||
0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x79, 0x48, 0x61, 0x6e, 0x64, 0x6c,
|
||||
0x65, 0x72, 0x12, 0x13, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74,
|
||||
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x43,
|
||||
0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x13, 0x5a,
|
||||
0x11, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x67, 0x72,
|
||||
0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var (
|
||||
file_internal_pkg_grpc_server_proto_rawDescOnce sync.Once
|
||||
file_internal_pkg_grpc_server_proto_rawDescData = file_internal_pkg_grpc_server_proto_rawDesc
|
||||
)
|
||||
|
||||
func file_internal_pkg_grpc_server_proto_rawDescGZIP() []byte {
|
||||
file_internal_pkg_grpc_server_proto_rawDescOnce.Do(func() {
|
||||
file_internal_pkg_grpc_server_proto_rawDescData = protoimpl.X.CompressGZIP(file_internal_pkg_grpc_server_proto_rawDescData)
|
||||
})
|
||||
return file_internal_pkg_grpc_server_proto_rawDescData
|
||||
}
|
||||
|
||||
var file_internal_pkg_grpc_server_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
|
||||
var file_internal_pkg_grpc_server_proto_goTypes = []interface{}{
|
||||
(*ClientRequest)(nil), // 0: grpc.ClientRequest
|
||||
(*ClientResponse)(nil), // 1: grpc.ClientResponse
|
||||
(*Database)(nil), // 2: grpc.Database
|
||||
}
|
||||
var file_internal_pkg_grpc_server_proto_depIdxs = []int32{
|
||||
2, // 0: grpc.ClientResponse.database:type_name -> grpc.Database
|
||||
0, // 1: grpc.handshake.ClientDiscoveryHandler:input_type -> grpc.ClientRequest
|
||||
1, // 2: grpc.handshake.ClientDiscoveryHandler:output_type -> grpc.ClientResponse
|
||||
2, // [2:3] is the sub-list for method output_type
|
||||
1, // [1:2] is the sub-list for method input_type
|
||||
1, // [1:1] is the sub-list for extension type_name
|
||||
1, // [1:1] is the sub-list for extension extendee
|
||||
0, // [0:1] is the sub-list for field type_name
|
||||
}
|
||||
|
||||
func init() { file_internal_pkg_grpc_server_proto_init() }
|
||||
func file_internal_pkg_grpc_server_proto_init() {
|
||||
if File_internal_pkg_grpc_server_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_internal_pkg_grpc_server_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*ClientRequest); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_internal_pkg_grpc_server_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*ClientResponse); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_internal_pkg_grpc_server_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Database); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
|
||||
RawDescriptor: file_internal_pkg_grpc_server_proto_rawDesc,
|
||||
NumEnums: 0,
|
||||
NumMessages: 3,
|
||||
NumExtensions: 0,
|
||||
NumServices: 1,
|
||||
},
|
||||
GoTypes: file_internal_pkg_grpc_server_proto_goTypes,
|
||||
DependencyIndexes: file_internal_pkg_grpc_server_proto_depIdxs,
|
||||
MessageInfos: file_internal_pkg_grpc_server_proto_msgTypes,
|
||||
}.Build()
|
||||
File_internal_pkg_grpc_server_proto = out.File
|
||||
file_internal_pkg_grpc_server_proto_rawDesc = nil
|
||||
file_internal_pkg_grpc_server_proto_goTypes = nil
|
||||
file_internal_pkg_grpc_server_proto_depIdxs = nil
|
||||
}
|
||||
@ -1,28 +0,0 @@
|
||||
syntax = "proto3";
|
||||
package grpc;
|
||||
|
||||
option go_package = "internal/pkg/grpc";
|
||||
|
||||
service handshake {
|
||||
rpc ClientDiscoveryHandler(ClientRequest) returns (ClientResponse);
|
||||
}
|
||||
|
||||
message ClientRequest {
|
||||
uint32 clientId = 1;
|
||||
string clientType = 2;
|
||||
string ip = 3; // client ip
|
||||
uint32 port = 4; // client port for gRPC server
|
||||
}
|
||||
|
||||
message ClientResponse {
|
||||
uint32 clientId = 1;
|
||||
uint32 serverPort = 2;
|
||||
Database database = 3;
|
||||
}
|
||||
|
||||
message Database {
|
||||
string URL = 1;
|
||||
string ORG = 2;
|
||||
string token = 3;
|
||||
string bucket = 4;
|
||||
}
|
||||
@ -1,81 +0,0 @@
|
||||
package influxdb
|
||||
|
||||
import (
|
||||
_ "fmt"
|
||||
|
||||
_ "github.com/influxdata/influxdb-client-go/v2"
|
||||
"github.com/spf13/viper"
|
||||
)
|
||||
|
||||
type DBInfo struct {
|
||||
URL string `mapstructure:"url"`
|
||||
Org string `mapstructure:"org,omitempty`
|
||||
Bucket string `mapstructure:"bucket,omitempty"`
|
||||
Token string `mapstructure:"token,omitempty"`
|
||||
// Client *influxdb2.Client
|
||||
}
|
||||
|
||||
type DBAdmin struct {
|
||||
// struct for admin methods
|
||||
*DBInfo
|
||||
Config *viper.Viper
|
||||
}
|
||||
|
||||
type DBClient struct {
|
||||
// struct for client methods
|
||||
*DBInfo
|
||||
Config *viper.Viper
|
||||
}
|
||||
|
||||
func NewDBInfo(config *viper.Viper) (*DBInfo, error) {
|
||||
db := &DBInfo{}
|
||||
// grabbing config vals
|
||||
err := config.UnmarshalKey("db", db)
|
||||
return db, err
|
||||
}
|
||||
|
||||
func NewDBClient(config *viper.Viper) (*DBClient, error) {
|
||||
|
||||
client := &DBClient{Config: config}
|
||||
// grabbing config vals
|
||||
var err error
|
||||
client.DBInfo, err = NewDBInfo(config)
|
||||
return client, err
|
||||
}
|
||||
|
||||
func NewDBAdmin(config *viper.Viper) (*DBAdmin, error) {
|
||||
admin := &DBAdmin{Config: config}
|
||||
var err error
|
||||
// creating client
|
||||
admin.DBInfo, err = NewDBInfo(config)
|
||||
return admin, err
|
||||
}
|
||||
|
||||
// base level funcs
|
||||
func (d *DBInfo) Start() error {
|
||||
// connect to DB based w/ info
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *DBAdmin) GetReactorClient(id int) (url, bucket, org, token string, err error) {
|
||||
// given an id returns
|
||||
// (url, org, bucket, token, error) for said id
|
||||
/*
|
||||
client := influxdb2.NewClient(d.URL, d.Token)
|
||||
defer client.Close()
|
||||
bucket, err := client.BucketsAPI().FindBucketByName(context.Background(), id)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if d.ReactorExists(id) {
|
||||
// get corresponding reactor token and bucket
|
||||
}
|
||||
*/
|
||||
url = d.URL
|
||||
org = d.Org
|
||||
token = ""
|
||||
bucket = ""
|
||||
//err = errors.New("Unimpl")
|
||||
err = nil
|
||||
return
|
||||
}
|
||||
@ -0,0 +1,13 @@
|
||||
package server
|
||||
|
||||
type dbinfo struct {
|
||||
url string
|
||||
org string
|
||||
token string
|
||||
bucket string
|
||||
}
|
||||
|
||||
func (c *Coordinator) getReactorDatabaseCred(id int) (*dbinfo, error) {
|
||||
|
||||
return &dbinfo{}, nil
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
"FRMS/internal/pkg/logging"
|
||||
"context"
|
||||
)
|
||||
|
||||
// ClientDiscoveryHandler implements the grpc method which can be called
|
||||
// by incoming clients to first make connection to the central
|
||||
// coordinator and receive database credentials.
|
||||
func (c *Coordinator) ReactorClientHandler(
|
||||
ctx context.Context,
|
||||
req *pb.ReactorClientRequest,
|
||||
) (*pb.ReactorClientResponse, error) {
|
||||
|
||||
id := int(req.GetId())
|
||||
|
||||
logging.Debug(
|
||||
logging.DClient,
|
||||
"LIS 00 reactor %v has connected\n",
|
||||
id,
|
||||
)
|
||||
|
||||
db, err := c.getReactorDatabaseCred(id)
|
||||
if err != nil {
|
||||
return &pb.ReactorClientResponse{}, err
|
||||
}
|
||||
|
||||
return &pb.ReactorClientResponse{
|
||||
Id: id,
|
||||
Url: db.url,
|
||||
Org: db.org,
|
||||
Token: db.token,
|
||||
Bucket: db.bucket,
|
||||
}, err
|
||||
}
|
||||
|
||||
// ReactorStatusHandler is a gRPC handler used to handle incoming
|
||||
// reactor requests containing information about said reactor.
|
||||
// It will get the associate reactor manager and pass the
|
||||
// request device information before returning an acknowledgement.
|
||||
func (c *Coordinator) ReactorStatusHandler(ctx context.Context, req *pb.ReactorStatusPing) (*pb.ReactorStatusResponse, error) {
|
||||
|
||||
// rm, err := c.LoadReactorManager(int(req.GetId()))
|
||||
|
||||
return &pb.ReactorStatusResponse{}, nil
|
||||
}
|
||||
@ -1,120 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
"FRMS/internal/pkg/logging"
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
// Listener is a struct that listens for incoming clients on a given port
|
||||
// and passes them the central coordinator.
|
||||
// Implements the gRPC handshake server for clients.
|
||||
type Listener struct {
|
||||
Port int `mapstructure:"lis"`
|
||||
ClientConnections chan *ClientPacket
|
||||
Err chan error
|
||||
pb.UnimplementedHandshakeServer
|
||||
}
|
||||
|
||||
// ClientPacket is a uniform type to pass on a channel to the server.
|
||||
type ClientPacket struct {
|
||||
*Client
|
||||
Response chan *ClientResponse
|
||||
}
|
||||
|
||||
// Client is a struct containing information about the client on
|
||||
// the incoming connection.
|
||||
type Client struct {
|
||||
//Ip string
|
||||
//Port int
|
||||
Id int
|
||||
Model string
|
||||
Type string
|
||||
}
|
||||
|
||||
// ClientResponse is the database credentials returned from the central
|
||||
// coordinator for the given client.
|
||||
type ClientResponse struct {
|
||||
Port int
|
||||
URL string
|
||||
Org string
|
||||
Token string
|
||||
Bucket string
|
||||
}
|
||||
|
||||
// NewListener createsa new listener with the given client and error channels
|
||||
func NewListener(
|
||||
cch chan *ClientPacket,
|
||||
ech chan error,
|
||||
) (*Listener, error) {
|
||||
|
||||
return &Listener{
|
||||
Err: ech,
|
||||
ClientConnections: cch,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Start activates the listener and kicks off the gRPC binding process
|
||||
func (l *Listener) Start() error {
|
||||
logging.Debug(logging.DStart, "LIS 01 Started client listener")
|
||||
return l.Register()
|
||||
}
|
||||
|
||||
// Register creates a net listener on the port and binds a grpc server to it
|
||||
// before registering a handshake server.
|
||||
func (l *Listener) Register() error {
|
||||
|
||||
lis, err := net.Listen("tcp", fmt.Sprintf(":%v", l.Port))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
grpcServer := grpc.NewServer()
|
||||
pb.RegisterHandshakeServer(grpcServer, l)
|
||||
|
||||
go grpcServer.Serve(lis)
|
||||
|
||||
logging.Debug(logging.DStart, "LIS 01 Registered on port %v", l.Port)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ClientDiscoveryHandler implements the grpc method which can be called
|
||||
// by incoming clients to first make connection to the central
|
||||
// coordinator and receive database credentials.
|
||||
func (l *Listener) ClientDiscoveryHandler(ctx context.Context, ping *pb.ClientRequest) (*pb.ClientResponse, error) {
|
||||
|
||||
c := &Client{
|
||||
Id: int(ping.GetClientId()),
|
||||
Type: ping.GetClientType(),
|
||||
}
|
||||
|
||||
logging.Debug(logging.DClient, "LIS 01 %v %v has connected\n", c.Type, c.Id)
|
||||
|
||||
ch := make(chan *ClientResponse)
|
||||
p := &ClientPacket{
|
||||
Client: c,
|
||||
Response: ch,
|
||||
}
|
||||
|
||||
l.ClientConnections <- p
|
||||
|
||||
resp := <-ch
|
||||
|
||||
db := &pb.Database{
|
||||
URL: resp.URL,
|
||||
ORG: resp.Org,
|
||||
Token: resp.Token,
|
||||
Bucket: resp.Bucket,
|
||||
}
|
||||
|
||||
return &pb.ClientResponse{
|
||||
ClientId: uint32(c.Id),
|
||||
ServerPort: uint32(resp.Port),
|
||||
Database: db,
|
||||
}, nil
|
||||
}
|
||||
@ -1,17 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// TestNewListener tries to create a new listener
|
||||
func TestNewListener(t *testing.T) {
|
||||
assert := assert.New(t)
|
||||
|
||||
cch := make(chan *ClientPacket)
|
||||
ech := make(chan error)
|
||||
_, err := NewListener(cch, ech)
|
||||
assert.Equal(err, nil, "creating listener failed")
|
||||
}
|
||||
@ -0,0 +1,121 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"FRMS/internal/pkg/logging"
|
||||
"FRMS/internal/pkg/manager"
|
||||
"errors"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNoReactorManager = errors.New("no reactor manager found")
|
||||
)
|
||||
|
||||
// ReactorManager can be started/stopped as clients connect/disconnect.
|
||||
type ReactorManager struct {
|
||||
Manager // base manager interface
|
||||
}
|
||||
|
||||
// Manager is an interface requiring a structure that can be started
|
||||
// and stopped as well as provide timeouts in milliseconds.
|
||||
type Manager interface {
|
||||
Start() error // status checks
|
||||
Stop() error
|
||||
Timeout() (time.Duration, error) // TO Generator
|
||||
}
|
||||
|
||||
// NewManager returns a manager fulfilling the Manager interface as well as
|
||||
// any potential errors.
|
||||
func NewManager(max int) (Manager, error) {
|
||||
return manager.New(max)
|
||||
}
|
||||
|
||||
// GetReactorManager returns a reactor manager for passed id.
|
||||
// Throws error if manager not found for id.
|
||||
func (c *Coordinator) LoadReactorManager(id int) (*ReactorManager, error) {
|
||||
|
||||
c.managerMu.RLock()
|
||||
defer c.managerMu.RUnlock()
|
||||
|
||||
rm, exists := c.directory[id]
|
||||
|
||||
if !exists {
|
||||
logging.Debug(
|
||||
logging.DClient,
|
||||
"RCO 00 creating manager for %v",
|
||||
id,
|
||||
)
|
||||
|
||||
m, err := NewManager(0)
|
||||
|
||||
rm = &ReactorManager{
|
||||
Manager: m,
|
||||
}
|
||||
|
||||
if err = rm.Start(); err != nil {
|
||||
return rm, err
|
||||
}
|
||||
|
||||
c.directory[id] = rm
|
||||
}
|
||||
|
||||
return rm, nil
|
||||
}
|
||||
|
||||
// // NewReactorManager takes in a client, config and channel to pass errors on.
|
||||
// // Returns a new reactor manager as well as any errors that occured during
|
||||
// // creation.
|
||||
// // Uses MaxConnectionAttempts which defaults to 10 to prevent
|
||||
// // unnessecary network load and/or timeout lengths.
|
||||
// func NewReactorManager(
|
||||
// ) (*ReactorManager, error) {
|
||||
|
||||
// m, err := NewManager(MaxConnectionAttempts)
|
||||
|
||||
// if err != nil {
|
||||
// return &ReactorManager{}, err
|
||||
// }
|
||||
|
||||
// return r, err
|
||||
// }
|
||||
|
||||
// Start logs the start and calls start on the embedded manager.
|
||||
func (r *ReactorManager) Start() error {
|
||||
// logging.Debug(logging.DStart, "RMA starting", r.Id)
|
||||
return r.Manager.Start()
|
||||
}
|
||||
|
||||
// Stop logs the stop and calls stop on the embedded manager.
|
||||
func (r *ReactorManager) Stop() error {
|
||||
// logging.Debug(logging.DExit, "RMA %v stopping", r.Id)
|
||||
return r.Manager.Stop()
|
||||
}
|
||||
|
||||
// UpdateClient is used to change the underlying manager client if there
|
||||
// changes to its data.
|
||||
//
|
||||
// BUG(Keegan): Client is not protected by a lock and may lead to races
|
||||
// func (r *ReactorManager) UpdateClient(cl *Client) error {
|
||||
// logging.Debug(logging.DClient, "RMA %v updating client", r.Id)
|
||||
// r.Client = cl
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// // ReactorDeviceHandler processes incoming device information and
|
||||
// // updates the manager accordingly.
|
||||
// func (r *ReactorManager) ReactorDeviceHandler(devs []*pb.Device) error {
|
||||
|
||||
// logging.Debug(logging.DClient, "CCO recieved ping from %v", r.Id)
|
||||
|
||||
// for _, dev := range devs {
|
||||
// logging.Debug(
|
||||
// logging.DClient,
|
||||
// "CCO %v device %v is %v",
|
||||
// r.Id,
|
||||
// dev.GetAddr(),
|
||||
// dev.GetStatus().String(),
|
||||
// )
|
||||
// }
|
||||
|
||||
// return nil
|
||||
// }
|
||||
@ -1,108 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
"FRMS/internal/pkg/logging"
|
||||
"FRMS/internal/pkg/manager"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/viper"
|
||||
)
|
||||
|
||||
// MaxConnectionAttempts is the max number of tries to allow
|
||||
// when connecting to a reactor.
|
||||
const MaxConnectionAttempts = 10
|
||||
|
||||
// Manager is an interface requiring a structure that can be started
|
||||
// and stopped as well as provide timeouts in milliseconds.
|
||||
type Manager interface {
|
||||
Start() error // status checks
|
||||
Stop() error
|
||||
Timeout() (time.Duration, error) // TO Generator
|
||||
}
|
||||
|
||||
// NewManager returns a manager fulfilling the Manager interface as well as
|
||||
// any potential errors.
|
||||
func NewManager(max int) (Manager, error) {
|
||||
return manager.New(max)
|
||||
}
|
||||
|
||||
// ReactorManager contains a base manager, client, global
|
||||
// config, and error channel.
|
||||
// The ReactorManager can be started/stopped as clients connect/disconnect.
|
||||
// Also serves as handler for gRPC requests from reactors.
|
||||
// Can be extended to write changes to config.
|
||||
type ReactorManager struct {
|
||||
Manager // base manager interface
|
||||
*Client
|
||||
Config *viper.Viper // global config to maintain
|
||||
Err chan error
|
||||
}
|
||||
|
||||
// NewReactorManager takes in a client, config and channel to pass errors on.
|
||||
// Returns a new reactor manager as well as any errors that occured during
|
||||
// creation.
|
||||
// Uses MaxConnectionAttempts which defaults to 10 to prevent
|
||||
// unnessecary network load and/or timeout lengths.
|
||||
func NewReactorManager(
|
||||
cl *Client,
|
||||
config *viper.Viper,
|
||||
errCh chan error,
|
||||
) (*ReactorManager, error) {
|
||||
|
||||
m, err := NewManager(MaxConnectionAttempts)
|
||||
|
||||
if err != nil {
|
||||
return &ReactorManager{}, err
|
||||
}
|
||||
|
||||
r := &ReactorManager{
|
||||
Manager: m,
|
||||
Client: cl,
|
||||
Config: config,
|
||||
Err: errCh,
|
||||
}
|
||||
|
||||
return r, err
|
||||
}
|
||||
|
||||
// Start logs the start and calls start on the embedded manager.
|
||||
func (r *ReactorManager) Start() error {
|
||||
logging.Debug(logging.DStart, "RMA %v starting", r.Id)
|
||||
return r.Manager.Start()
|
||||
}
|
||||
|
||||
// Stop logs the stop and calls stop on the embedded manager.
|
||||
func (r *ReactorManager) Stop() error {
|
||||
logging.Debug(logging.DExit, "RMA %v stopping", r.Id)
|
||||
return r.Manager.Stop()
|
||||
}
|
||||
|
||||
// UpdateClient is used to change the underlying manager client if there
|
||||
// changes to its data.
|
||||
//
|
||||
// BUG(Keegan): Client is not protected by a lock and may lead to races
|
||||
func (r *ReactorManager) UpdateClient(cl *Client) error {
|
||||
logging.Debug(logging.DClient, "RMA %v updating client", r.Id)
|
||||
r.Client = cl
|
||||
return nil
|
||||
}
|
||||
|
||||
// ReactorDeviceHandler processes incoming device information and
|
||||
// updates the manager accordingly.
|
||||
func (r *ReactorManager) ReactorDeviceHandler(devs []*pb.Device) error {
|
||||
|
||||
logging.Debug(logging.DClient, "RMA %v recieved ping", r.Id)
|
||||
|
||||
for _, dev := range devs {
|
||||
logging.Debug(
|
||||
logging.DClient,
|
||||
"RMA %v device %v is %v",
|
||||
r.Id,
|
||||
dev.GetAddr(),
|
||||
dev.GetStatus().String(),
|
||||
)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -1,85 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
"math/rand"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/viper"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// dummyClient creates a dummy client for testing.
|
||||
func dummyClient() *Client {
|
||||
return &Client{
|
||||
Id: rand.Int(),
|
||||
Model: "dummy",
|
||||
Type: "dummy",
|
||||
}
|
||||
}
|
||||
|
||||
func dummyDevices() []*pb.Device {
|
||||
numDevs := 10
|
||||
|
||||
devs := make([]*pb.Device, numDevs)
|
||||
|
||||
for i := 0; i < numDevs; i++ {
|
||||
dev := &pb.Device{
|
||||
Addr: int32(rand.Intn(255)),
|
||||
Status: pb.Status(rand.Intn(2)),
|
||||
}
|
||||
devs = append(devs, dev)
|
||||
}
|
||||
|
||||
return devs
|
||||
}
|
||||
|
||||
// dummyReactorManager creates a dummy reactor manager for testing.
|
||||
func dummyReactorManager() (*ReactorManager, error) {
|
||||
|
||||
ch := make(chan error)
|
||||
cl := dummyClient()
|
||||
return NewReactorManager(cl, viper.New(), ch)
|
||||
}
|
||||
|
||||
// TestNewReactorManager tries to create a new reactor manager.
|
||||
func TestNewReactorManager(t *testing.T) {
|
||||
assert := assert.New(t)
|
||||
_, err := dummyReactorManager()
|
||||
assert.Equal(err, nil, "failed to create reactor manager")
|
||||
}
|
||||
|
||||
// TestReactorManager tries to start/stop reactor manager
|
||||
func TestReactorManager(t *testing.T) {
|
||||
assert := assert.New(t)
|
||||
rm, err := dummyReactorManager()
|
||||
assert.Equal(err, nil, "failed to create reactor manager")
|
||||
|
||||
cycles := 10
|
||||
for i := 0; i < cycles; i++ {
|
||||
assert.NoError(rm.Start(), "failed to start")
|
||||
assert.NoError(rm.Stop(), "failed to start")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateClient tries to update a reactor managers embedded client.
|
||||
func TestUpdateClient(t *testing.T) {
|
||||
|
||||
assert := assert.New(t)
|
||||
rm, err := dummyReactorManager()
|
||||
assert.Equal(err, nil, "failed to create reactor manager")
|
||||
|
||||
cl := dummyClient()
|
||||
|
||||
assert.NoError(rm.UpdateClient(cl), "failed to update client")
|
||||
}
|
||||
|
||||
// TestReactorDeviceHandler ensures that a list of devices can be processed.
|
||||
func TestReactorDeviceHandler(t *testing.T) {
|
||||
assert := assert.New(t)
|
||||
rm, err := dummyReactorManager()
|
||||
assert.Equal(err, nil, "failed to create reactor manager")
|
||||
|
||||
devs := dummyDevices()
|
||||
assert.NoError(rm.ReactorDeviceHandler(devs), "failed to handle devices")
|
||||
}
|
||||
@ -0,0 +1,19 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
"FRMS/internal/pkg/logging"
|
||||
)
|
||||
|
||||
func (c *Coordinator) Register() error {
|
||||
// register services
|
||||
pb.RegisterHandshakeServer(c.grpcServer, c)
|
||||
|
||||
go c.grpcServer.Serve(c.listener)
|
||||
// testing
|
||||
pb.RegisterMonitoringServer(c.grpcServer, c)
|
||||
|
||||
logging.Debug(logging.DStart, "CCO 00 registered grpc")
|
||||
|
||||
return nil
|
||||
}
|
||||
Loading…
Reference in New Issue