cmd: Add Xbus support

cmd/program.go

 package cmd
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -15,6 +16,8 @@
 	"orus.io/orus-io/go-orusapi/database"
 )
 
+var ErrInvalidConfiguration = errors.New("invalid configuration")
+
 type ConfigFile struct {
 	ConfigFile string `long:"config" short:"c" env:"CONFIG" no-ini:"t" description:"A configuration file"`
 }
@@ -23,6 +26,11 @@
 	Environment string `long:"environment" env:"ENVIRONMENT" ini-name:"environment" default:"default" description:"A environment name, used in sentry and prometheus"`
 }
 
+type XbusActorFactory struct {
+	Name    string
+	Factory any
+}
+
 type subcommand struct {
 	name string
 	init func(*Program) any
@@ -38,6 +46,7 @@
 	InfoOptions      InfoOptions
 	LoggingOptions   *orusapi.LoggingOptions
 	DatabaseOptions  database.Options
+	XbusOptions      XbusOptions
 
 	DB *sqlx.DB
 
@@ -47,6 +56,8 @@
 	hasDB           bool
 	dbMigrateSource source.Driver
 
+	setupXbusActors  func(*Program) []XbusActorFactory
+	xbusActorNames   []string
 	setupHandler     func(*Program) http.Handler
 	setupSubcommands []subcommand
 }
@@ -59,6 +70,21 @@
 	}
 }
 
+func WithXbusActors(factories func(*Program) []XbusActorFactory) Option {
+	return func(program *Program) {
+		program.setupXbusActors = factories
+
+		g, err := program.Parser.AddGroup("Xbus", "Xbus options", &program.XbusOptions)
+		if err != nil {
+			panic(err)
+		}
+		g.Namespace = "xbus"
+		g.EnvNamespace = "XBUS"
+
+		SetupXbusCmd(program)
+	}
+}
+
 func WithDatabase(migrateSource source.Driver) Option {
 	return func(program *Program) {
 		program.hasDB = true

cmd/xbus.go

+package cmd
+
+import (
+	"fmt"
+
+	"orus.io/xbus/xbus-stdlib/service"
+	"xbus.io/go-xbus/v4"
+	"xbus.io/go-xbus/v4/api"
+)
+
+type XbusOptions struct {
+	Config string `long:"config-file" ini-name:"config-file" description:"xbus client configuration file"`
+}
+
+func SetupXbusCmd(program *Program) *XbusCmd {
+	cmd := XbusCmd{}
+	c, err := program.Parser.AddCommand("xbus", "Standalone xbus client", "", &cmd)
+	if err != nil {
+		panic(err)
+	}
+
+	SetupXbusServeCmd(program, c)
+
+	return &cmd
+}
+
+type XbusCmd struct{}
+
+func (p *Program) xbusClientOptions() (xbus.COptions, error) {
+	if p.XbusOptions.Config == "" {
+		return xbus.COptions{}, fmt.Errorf("%w: missing xbus.config-file", ErrInvalidConfiguration)
+	}
+	coptions, err := xbus.COptionsFromFile(p.XbusOptions.Config, api.Account_ACTOR, p.Logger)
+	// TODO allow the program user to make verifications on the configuration
+	return coptions, err
+}
+
+func (p *Program) xbusRegisterActors() error {
+	for _, factory := range p.setupXbusActors(p) {
+		switch f := factory.Factory.(type) {
+		case service.ConsumerFactory:
+			if err := service.RegisterConsumer(factory.Name, f); err != nil {
+				return err
+			}
+		case service.ConsumerFunc:
+			if err := service.RegisterConsumerFunc(factory.Name, f); err != nil {
+				return err
+			}
+		case service.WorkerFactory:
+			if err := service.RegisterWorker(factory.Name, f); err != nil {
+				return err
+			}
+		case service.WorkerFunc:
+			if err := service.RegisterWorkerFunc(factory.Name, f); err != nil {
+				return err
+			}
+		case xbus.NewActorServiceFunc:
+			xbus.RegisterActorService(factory.Name, f)
+		default:
+			return fmt.Errorf("%s has an factory type: %t", factory.Name, factory.Factory)
+		}
+		p.xbusActorNames = append(p.xbusActorNames, factory.Name)
+	}
+
+	return nil
+}
+
+func (p *Program) xbusUnregisterActors() {
+	for _, n := range p.xbusActorNames {
+		xbus.UnregisterActorService(n)
+	}
+	p.xbusActorNames = nil
+}

cmd/xbus_serve.go

+package cmd
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/jessevdk/go-flags"
+	"xbus.io/go-xbus/v4"
+)
+
+func SetupXbusServeCmd(program *Program, cmd *flags.Command) *XbusServeCmd {
+	serveCmd := XbusServeCmd{
+		program: program,
+	}
+
+	_, err := cmd.AddCommand("serve", "Start a standalone xbus client", "", &serveCmd)
+	if err != nil {
+		panic(err)
+	}
+
+	return &serveCmd
+}
+
+type XbusServeCmd struct {
+	program *Program
+
+	NoReconnect bool `long:"no-reconnect" description:"When the connection to xbus is lost, stops the program instead of attempting a reconnection"`
+}
+
+func (cmd *XbusServeCmd) Execute([]string) error {
+	coptions, err := cmd.program.xbusClientOptions()
+	if err != nil {
+		return err
+	}
+	coptions.ForceStart = !cmd.NoReconnect
+	coptions.NoReconnect = cmd.NoReconnect
+
+	if err := cmd.program.xbusRegisterActors(); err != nil {
+		return err
+	}
+	defer cmd.program.xbusUnregisterActors()
+
+	client := xbus.NewClient(coptions, cmd.program.Logger, cmd.program.LoggingOptions.Output())
+
+	if err := cmd.program.EnsureDB(false); err != nil {
+		return err
+	}
+
+	if err := client.Startup(true); err != nil {
+		if client.GetConn() != nil {
+			client.Close()
+		}
+
+		return err
+	}
+	cmd.program.Logger.Info().Msg("Xbus client ready")
+
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	select {
+	case <-interrupt:
+		client.Shutdown()
+	case <-client.Done():
+		return nil
+	}
+
+	return nil
+}

go.mod

@@ -21,6 +21,8 @@
 	github.com/stretchr/testify v1.8.4
 	golang.org/x/net v0.15.0
 	golang.org/x/term v0.12.0
+	orus.io/xbus/xbus-stdlib v1.3.3
+	xbus.io/go-xbus/v4 v4.0.0-20230823140355-edd2d56a84db
 )
 
 require (
@@ -28,6 +30,7 @@
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -31,5 +34,6 @@
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
@@ -34,4 +38,5 @@
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
+	github.com/klauspost/compress v1.17.0 // indirect
 	github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
 	github.com/lib/pq v1.10.9 // indirect
@@ -36,6 +41,8 @@
 	github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
 	github.com/lib/pq v1.10.9 // indirect
+	github.com/m4rw3r/uuid v1.0.1 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
@@ -38,6 +45,7 @@
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
@@ -42,7 +50,13 @@
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/nats-io/nats.go v1.30.2 // indirect
+	github.com/nats-io/nkeys v0.4.5 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/nats-rpc/nrpc v0.0.0-20230331165850-dd9d3cdece2a // indirect
+	github.com/nightlyone/lockfile v1.0.0 // indirect
+	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/rs/xid v1.5.0 // indirect
@@ -44,6 +58,12 @@
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/rs/xid v1.5.0 // indirect
+	github.com/spf13/afero v1.10.0 // indirect
+	github.com/spf13/cast v1.5.1 // indirect
+	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/viper v1.16.0 // indirect
+	github.com/subosito/gotenv v1.6.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
@@ -49,4 +69,5 @@
 	go.uber.org/atomic v1.11.0 // indirect
+	golang.org/x/crypto v0.13.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
@@ -50,5 +71,7 @@
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
+	gopkg.in/ini.v1 v1.67.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )