Merge pull request #3268 from owncloud/nats-stan-fixes

[full-ci] NATS integration fixes
2022-03-07 09:59:08 +01:00
parent 0849563810 37b56d920e
commit eea1e23d1b
4 changed files with 189 additions and 61 deletions
@@ -1,20 +1,16 @@
 package command

 import (
+	"context"
 	"fmt"
-	"os"
-	"os/signal"
-	"syscall"

-	"github.com/cs3org/reva/v2/pkg/events/server"
+	"github.com/oklog/run"
+
 	"github.com/owncloud/ocis/nats/pkg/config"
 	"github.com/owncloud/ocis/nats/pkg/config/parser"
 	"github.com/owncloud/ocis/nats/pkg/logging"
-	"github.com/owncloud/ocis/ocis-pkg/log"
+	"github.com/owncloud/ocis/nats/pkg/server/nats"
 	"github.com/urfave/cli/v2"
-
-	// TODO: .Logger Option on events/server would make this import redundant
-	stanServer "github.com/nats-io/nats-streaming-server/server"
 )

 // Server is the entrypoint for the server command.
@@ -28,65 +24,45 @@ func Server(cfg *config.Config) *cli.Command {
 		},
 		Action: func(c *cli.Context) error {
 			logger := logging.Configure(cfg.Service.Name, cfg.Log)
-			ch := make(chan os.Signal, 1)
-			signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
-			err := server.RunNatsServer(server.Host(cfg.Nats.Host), server.Port(cfg.Nats.Port), server.StanOpts(func(o *stanServer.Options) {
-				o.CustomLogger = &logWrapper{logger}
-			}))
+
+			gr := run.Group{}
+			ctx, cancel := func() (context.Context, context.CancelFunc) {
+				if cfg.Context == nil {
+					return context.WithCancel(context.Background())
+				}
+				return context.WithCancel(cfg.Context)
+			}()
+
+			defer cancel()
+
+			natsServer, err := nats.NewNATSServer(
+				ctx,
+				nats.Host(cfg.Nats.Host),
+				nats.Port(cfg.Nats.Port),
+				nats.Logger(logging.NewLogWrapper(logger)),
+			)
 			if err != nil {
 				return err
 			}
-			for {
+
+			gr.Add(func() error {
+				err := make(chan error)
 				select {
-				case <-ch:
-					// TODO: Should we shut down the NatsServer in a proper way here?
-					// That would require a reference to the StanServer instance for being able to call
-					// StanServer.Shutdown() github.com/cs3org/reva/pkg/events/server doesn't provide that
-					// currently
+				case <-ctx.Done():
 					return nil
+				case err <- natsServer.ListenAndServe():
+					return <-err
 				}
-			}
+
+			}, func(_ error) {
+				logger.Info().
+					Msg("Shutting down server")
+
+				natsServer.Shutdown()
+				cancel()
+			})
+
+			return gr.Run()
 		},
 	}
 }
-
-// we need to wrap our logger so we can pass it to the nats server
-type logWrapper struct {
-	logger log.Logger
-}
-
-// Noticef logs a notice statement
-func (l *logWrapper) Noticef(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	l.logger.Info().Msg(msg)
-}
-
-// Warnf logs a warning statement
-func (l *logWrapper) Warnf(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	l.logger.Warn().Msg(msg)
-}
-
-// Fatalf logs a fatal statement
-func (l *logWrapper) Fatalf(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	l.logger.Fatal().Msg(msg)
-}
-
-// Errorf logs an error statement
-func (l *logWrapper) Errorf(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	l.logger.Error().Msg(msg)
-}
-
-// Debugf logs a debug statement
-func (l *logWrapper) Debugf(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	l.logger.Debug().Msg(msg)
-}
-
-// Tracef logs a trace statement
-func (l *logWrapper) Tracef(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	l.logger.Trace().Msg(msg)
-}
@@ -0,0 +1,52 @@
+package logging
+
+import (
+	"fmt"
+
+	"github.com/owncloud/ocis/ocis-pkg/log"
+)
+
+func NewLogWrapper(logger log.Logger) *LogWrapper {
+	return &LogWrapper{logger}
+}
+
+// we need to wrap our logger so we can pass it to the nats server
+type LogWrapper struct {
+	logger log.Logger
+}
+
+// Noticef logs a notice statement
+func (l *LogWrapper) Noticef(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	l.logger.Info().Msg(msg)
+}
+
+// Warnf logs a warning statement
+func (l *LogWrapper) Warnf(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	l.logger.Warn().Msg(msg)
+}
+
+// Fatalf logs a fatal statement
+func (l *LogWrapper) Fatalf(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	l.logger.Fatal().Msg(msg)
+}
+
+// Errorf logs an error statement
+func (l *LogWrapper) Errorf(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	l.logger.Error().Msg(msg)
+}
+
+// Debugf logs a debug statement
+func (l *LogWrapper) Debugf(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	l.logger.Debug().Msg(msg)
+}
+
+// Tracef logs a trace statement
+func (l *LogWrapper) Tracef(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	l.logger.Trace().Msg(msg)
+}
@@ -0,0 +1,69 @@
+package nats
+
+import (
+	"context"
+	"time"
+
+	natsServer "github.com/nats-io/nats-server/v2/server"
+	stanServer "github.com/nats-io/nats-streaming-server/server"
+)
+
+var NATSListenAndServeLoopTimer = 1 * time.Second
+
+type NATSServer struct {
+	ctx context.Context
+
+	natsOpts *natsServer.Options
+	stanOpts *stanServer.Options
+
+	server *stanServer.StanServer
+}
+
+// NewNATSServer returns a new NATSServer
+func NewNATSServer(ctx context.Context, opts ...Option) (*NATSServer, error) {
+
+	server := &NATSServer{
+		ctx:      ctx,
+		natsOpts: &stanServer.DefaultNatsServerOptions,
+		stanOpts: stanServer.GetDefaultOptions(),
+	}
+
+	for _, o := range opts {
+		o(server.natsOpts, server.stanOpts)
+	}
+
+	return server, nil
+}
+
+// ListenAndServe runs the NATSServer in a blocking way until the server is shutdown or an error occurs
+func (n *NATSServer) ListenAndServe() (err error) {
+	n.server, err = stanServer.RunServerWithOpts(
+		n.stanOpts,
+		n.natsOpts,
+	)
+	if err != nil {
+		return err
+	}
+
+	defer n.Shutdown()
+
+	for {
+		// check if NATs server has an encountered an error
+		if err := n.server.LastError(); err != nil {
+			return err
+		}
+		// check if the NATs server is still running
+		if n.server.State() == stanServer.Shutdown {
+			return nil
+		}
+		// check if context was cancelled
+		if n.ctx.Err() != nil {
+			return nil
+		}
+		time.Sleep(NATSListenAndServeLoopTimer)
+	}
+}
+
+func (n *NATSServer) Shutdown() {
+	n.server.Shutdown()
+}
@@ -0,0 +1,31 @@
+package nats
+
+import (
+	natsServer "github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats-streaming-server/logger"
+	stanServer "github.com/nats-io/nats-streaming-server/server"
+)
+
+// Option configures the nats server
+type Option func(*natsServer.Options, *stanServer.Options)
+
+// Host sets the host URL for the nats server
+func Host(url string) Option {
+	return func(no *natsServer.Options, _ *stanServer.Options) {
+		no.Host = url
+	}
+}
+
+// Port sets the host URL for the nats server
+func Port(port int) Option {
+	return func(no *natsServer.Options, _ *stanServer.Options) {
+		no.Port = port
+	}
+}
+
+// Port sets the host URL for the nats server
+func Logger(logger logger.Logger) Option {
+	return func(no *natsServer.Options, so *stanServer.Options) {
+		so.CustomLogger = logger
+	}
+}