@@ -2,90 +2,45 @@ package main
22
33import (
44 "context"
5+ "fmt"
6+ "golang.org/x/sync/errgroup"
57 "log"
6- "net"
78 "os"
89 "os/signal"
910 "syscall"
10- "time"
1111
1212 "github.com/spf13/cobra"
1313 "github.com/spf13/viper"
1414
15- "google.golang.org/grpc"
16- "google.golang.org/grpc/reflection"
17-
1815 "github.com/datum-cloud/galactic-agent/api/local"
16+ "github.com/datum-cloud/galactic-agent/api/remote"
1917)
2018
21- const DEFAULT_SOCKET_PATH = "/var/run/galactic/agent.sock"
22-
23- type LocalServer struct {
24- local.UnimplementedLocalServer
25- }
26-
27- func (s * LocalServer ) Register (ctx context.Context , req * local.RegisterRequest ) (* local.RegisterReply , error ) {
28- log .Printf ("REGISTER: vpc='%v', vpcattachment='%v', networks='%v'\n " , req .GetVpc (), req .GetVpcattachment (), req .GetNetworks ())
29- return & local.RegisterReply {Confirmed : true }, nil
30- }
31-
32- func (s * LocalServer ) Deregister (ctx context.Context , req * local.DeregisterRequest ) (* local.DeregisterReply , error ) {
33- log .Printf ("DEREGISTER: vpc='%v', vpcattachment='%v', networks='%v'\n " , req .GetVpc (), req .GetVpcattachment (), req .GetNetworks ())
34- return & local.DeregisterReply {Confirmed : true }, nil
35- }
36-
37- func Serve (socketPath string ) error {
38- listener , err := net .Listen ("unix" , socketPath )
39- if err != nil {
40- return err
41- }
42- defer listener .Close () //nolint:errcheck
43-
44- s := grpc .NewServer ()
45- local .RegisterLocalServer (s , & LocalServer {})
46-
47- reflection .Register (s )
48-
49- stop := make (chan os.Signal , 1 )
50- signal .Notify (stop , syscall .SIGINT , syscall .SIGTERM )
51-
52- go func () {
53- log .Printf ("gRPC listening on unix://%s" , socketPath )
54- if err := s .Serve (listener ); err != nil {
55- log .Printf ("serve exited: %v" , err )
56- }
57- }()
58-
59- <- stop
60- log .Println ("shutting down..." )
61- done := make (chan struct {})
62- go func () {
63- s .GracefulStop ()
64- close (done )
65- }()
66- select {
67- case <- done :
68- case <- time .After (5 * time .Second ):
69- s .Stop ()
70- }
71- return nil
72- }
73-
7419var configFile string
7520
7621func initConfig () {
22+ viper .SetDefault ("socket_path" , "/var/run/galactic/agent.sock" )
23+ viper .SetDefault ("mqtt_host" , "mqtt" )
24+ viper .SetDefault ("mqtt_port" , 1883 )
25+ viper .SetDefault ("mqtt_qos" , 0 )
26+ viper .SetDefault ("mqtt_topic_receive" , "galactic/default/receive" )
27+ viper .SetDefault ("mqtt_topic_send" , "galactic/default/send" )
7728 if configFile != "" {
7829 viper .SetConfigFile (configFile )
7930 }
8031 viper .AutomaticEnv ()
81- viper .SetDefault ("socket_path" , "/var/run/galactic/agent.sock" )
8232 if err := viper .ReadInConfig (); err == nil {
8333 log .Printf ("Using config file: %s\n " , viper .ConfigFileUsed ())
8434 } else {
8535 log .Printf ("No config file found - using defaults." )
8636 }
8737}
8838
39+ var (
40+ l local.Local
41+ r remote.Remote
42+ )
43+
8944func main () {
9045 cmd := & cobra.Command {
9146 Use : "galactic-agent" ,
@@ -94,9 +49,45 @@ func main() {
9449 initConfig ()
9550 },
9651 Run : func (cmd * cobra.Command , args []string ) {
97- if err := Serve (viper .GetString ("socket_path" )); err != nil {
98- log .Fatalf ("Serve failed: %v" , err )
52+ ctx , stop := signal .NotifyContext (context .Background (), syscall .SIGINT , syscall .SIGTERM )
53+ defer stop () //nolint:errcheck
54+
55+ l = local.Local {
56+ SocketPath : viper .GetString ("socket_path" ),
57+ RegisterHandler : func (vpc , vpcAttachment string , networks []string ) error {
58+ log .Printf ("REGISTER: vpc='%v', vpcattachment='%v', networks='%v'\n " , vpc , vpcAttachment , networks )
59+ r .Send (fmt .Sprintf ("REGISTER: vpc='%v', vpcattachment='%v', networks='%v'" , vpc , vpcAttachment , networks ))
60+ return nil
61+ },
62+ DeregisterHandler : func (vpc , vpcAttachment string , networks []string ) error {
63+ log .Printf ("DEREGISTER: vpc='%v', vpcattachment='%v', networks='%v'\n " , vpc , vpcAttachment , networks )
64+ r .Send (fmt .Sprintf ("DEREGISTER: vpc='%v', vpcattachment='%v', networks='%v'" , vpc , vpcAttachment , networks ))
65+ return nil
66+ },
67+ }
68+
69+ r = remote.Remote {
70+ Host : viper .GetString ("mqtt_host" ),
71+ Port : viper .GetInt ("mqtt_port" ),
72+ QoS : byte (viper .GetInt ("mqtt_qos" )),
73+ TopicRX : viper .GetString ("mqtt_topic_receive" ),
74+ TopicTX : viper .GetString ("mqtt_topic_send" ),
75+ ReceiveHandler : func (payload interface {}) {
76+ log .Printf ("MQTT received: %s" , payload )
77+ },
78+ }
79+
80+ g , ctx := errgroup .WithContext (ctx )
81+ g .Go (func () error {
82+ return l .Serve (ctx )
83+ })
84+ g .Go (func () error {
85+ return r .Run (ctx )
86+ })
87+ if err := g .Wait (); err != nil {
88+ log .Printf ("Error: %v" , err )
9989 }
90+ log .Printf ("Shutdown" )
10091 },
10192 }
10293 cmd .PersistentFlags ().StringVar (& configFile , "config" , "" , "config file" )
0 commit comments