switch to NATS JetStream

nats/pkg/command/server.go

@@ -37,9 +37,10 @@ func Server(cfg *config.Config) *cli.Command {
 
 			natsServer, err := nats.NewNATSServer(
 				ctx,
+				logging.NewLogWrapper(logger),
 				nats.Host(cfg.Nats.Host),
 				nats.Port(cfg.Nats.Port),
-				nats.Logger(logging.NewLogWrapper(logger)),
+				nats.ClusterID("ocis-cluster"),
 			)
 			if err != nil {
 				return err

nats/pkg/server/nats/nats.go

@@ -4,64 +4,56 @@ import (
 	"context"
 	"time"
 
-	natsServer "github.com/nats-io/nats-server/v2/server"
-	stanServer "github.com/nats-io/nats-streaming-server/server"
+	nserver "github.com/nats-io/nats-server/v2/server"
 )
 
 var NATSListenAndServeLoopTimer = 1 * time.Second
 
 type NATSServer struct {
-	ctx context.Context
-
-	natsOpts *natsServer.Options
-	stanOpts *stanServer.Options
-
-	server *stanServer.StanServer
+	ctx             context.Context
+	jetStreamConfig *nserver.JetStreamConfig
+	server          *nserver.Server
 }
 
-// NewNATSServer returns a new NATSServer
-func NewNATSServer(ctx context.Context, opts ...Option) (*NATSServer, error) {
-
-	server := &NATSServer{
-		ctx:      ctx,
-		natsOpts: &stanServer.DefaultNatsServerOptions,
-		stanOpts: stanServer.GetDefaultOptions(),
-	}
+func NewNATSServer(ctx context.Context, logger nserver.Logger, opts ...Option) (*NATSServer, error) {
+	options := &nserver.Options{}
 
 	for _, o := range opts {
-		o(server.natsOpts, server.stanOpts)
+		o(options)
 	}
 
-	return server, nil
+	server, err := nserver.NewServer(
+		options,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	server.SetLoggerV2(logger, true, true, false)
+
+	c := &nserver.JetStreamConfig{
+		StoreDir: "/tmp/ocis-jetstream", // TODO: configurable
+	}
+
+	return &NATSServer{
+		ctx:             ctx,
+		jetStreamConfig: c,
+		server:          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,
-	)
+	// start NATS first
+	go n.server.Start()
+	// start NATS JetStream second
+	n.server.EnableJetStream(n.jetStreamConfig)
 	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)
-	}
+	<-n.ctx.Done()
+	return nil
 }
 
 func (n *NATSServer) Shutdown() {

nats/pkg/server/nats/options.go

@@ -1,31 +1,29 @@
 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"
+	nserver "github.com/nats-io/nats-server/v2/server"
 )
 
 // Option configures the nats server
-type Option func(*natsServer.Options, *stanServer.Options)
+type Option func(*nserver.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
+	return func(o *nserver.Options) {
+		o.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
+	return func(o *nserver.Options) {
+		o.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
+// ClusterID sets the name for the nats cluster
+func ClusterID(clusterID string) Option {
+	return func(o *nserver.Options) {
+		o.Cluster.Name = clusterID
 	}
 }