xbus_serve.go

package cmd

import (
	"os"
	"os/signal"
	"syscall"

	"github.com/jessevdk/go-flags"
	"xbus.io/go-xbus/v4"
)

func SetupXbusServeCmd[E any](program *Program[E], cmd *flags.Command) *XbusServeCmd[E] {
	serveCmd := XbusServeCmd[E]{
		program: program,
	}

	_, err := cmd.AddCommand("serve", "Start a standalone xbus client", "", &serveCmd)
	if err != nil {
		panic(err)
	}

	return &serveCmd
}

//nolint:lll
type XbusServeCmd[E any] struct {
	program *Program[E]

	NoReconnect bool `long:"no-reconnect" ini-name:"no-reconnect" description:"When the connection to xbus is lost, stops the program instead of attempting a reconnection"`
}

func (cmd *XbusServeCmd[E]) 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
}