feat: default stream

Author: bketelsen

cmd/cmd.go

@@ -23,6 +23,7 @@ import (
 	"go-micro.dev/v5/debug/profile/http"
 	"go-micro.dev/v5/debug/profile/pprof"
 	"go-micro.dev/v5/debug/trace"
+	"go-micro.dev/v5/events"
 	"go-micro.dev/v5/logger"
 	mprofile "go-micro.dev/v5/profile"
 	"go-micro.dev/v5/registry"
@@ -293,6 +294,7 @@ var (
 	DefaultCaches = map[string]func(...cache.Option) cache.Cache{
 		"redis": redis.NewRedisCache,
 	}
+	DefaultStreams = map[string]func(...events.Option) (events.Stream, error){}
 )
 
 func init() {
@@ -313,6 +315,7 @@ func newCmd(opts ...Option) Cmd {
 		DebugProfile: &profile.DefaultProfile,
 		Config:       &config.DefaultConfig,
 		Cache:        &cache.DefaultCache,
+		Stream:       &events.DefaultStream,
 
 		Brokers:       DefaultBrokers,
 		Clients:       DefaultClients,
@@ -376,7 +379,10 @@ func (c *cmd) Before(ctx *cli.Context) error {
 	if profileName != "" {
 		switch profileName {
 		case "local":
-			imported := mprofile.LocalProfile()
+			imported, ierr := mprofile.LocalProfile()
+			if ierr != nil {
+				return fmt.Errorf("failed to load local profile: %v", ierr)
+			}
 			*c.opts.Registry = imported.Registry
 			registry.DefaultRegistry = imported.Registry
 			*c.opts.Broker = imported.Broker
@@ -386,7 +392,10 @@ func (c *cmd) Before(ctx *cli.Context) error {
 			*c.opts.Transport = imported.Transport
 			transport.DefaultTransport = imported.Transport
 		case "nats":
-			imported := mprofile.NatsProfile()
+			imported, ierr := mprofile.NatsProfile()
+			if ierr != nil {
+				return fmt.Errorf("failed to load nats profile: %v", ierr)
+			}
 			// Set the registry
 			sopts, clopts := c.setRegistry(imported.Registry)
 			serverOpts = append(serverOpts, sopts...)
@@ -407,6 +416,11 @@ func (c *cmd) Before(ctx *cli.Context) error {
 			serverOpts = append(serverOpts, sopts...)
 			clientOpts = append(clientOpts, clopts...)
 
+			// Set the stream
+			sopts, clopts = c.setStream(imported.Stream)
+			serverOpts = append(serverOpts, sopts...)
+			clientOpts = append(clientOpts, clopts...)
+
 		// Add more profiles as needed
 		default:
 			return fmt.Errorf("unsupported profile: %s", profileName)
@@ -701,6 +715,17 @@ func (c *cmd) setRegistry(r registry.Registry) ([]server.Option, []client.Option
 	registry.DefaultRegistry = *c.opts.Registry
 	return serverOpts, clientOpts
 }
+func (c *cmd) setStream(s events.Stream) ([]server.Option, []client.Option) {
+	var serverOpts []server.Option
+	var clientOpts []client.Option
+	*c.opts.Stream = s
+	// TODO: do server and client need a Stream?
+	// serverOpts = append(serverOpts, server.Registry(*c.opts.Registry))
+	// clientOpts = append(clientOpts, client.Registry(*c.opts.Registry))
+
+	events.DefaultStream = *c.opts.Stream
+	return serverOpts, clientOpts
+}
 
 func (c *cmd) setBroker(b broker.Broker) ([]server.Option, []client.Option) {
 	var serverOpts []server.Option

cmd/options.go

@@ -10,6 +10,7 @@ import (
 	"go-micro.dev/v5/config"
 	"go-micro.dev/v5/debug/profile"
 	"go-micro.dev/v5/debug/trace"
+	"go-micro.dev/v5/events"
 	"go-micro.dev/v5/registry"
 	"go-micro.dev/v5/selector"
 	"go-micro.dev/v5/server"
@@ -42,13 +43,15 @@ type Options struct {
 	Broker     *broker.Broker
 	Auths      map[string]func(...auth.Option) auth.Auth
 	Store      *store.Store
+	Stream     *events.Stream
 	Configs    map[string]func(...config.Option) (config.Config, error)
 	Clients    map[string]func(...client.Option) client.Client
 	Registries map[string]func(...registry.Option) registry.Registry
 	Selectors  map[string]func(...selector.Option) selector.Selector
 	Servers    map[string]func(...server.Option) server.Server
 	Transports map[string]func(...transport.Option) transport.Transport
 	Stores     map[string]func(...store.Option) store.Store
+	Streams    map[string]func(...events.Option) events.Stream
 	Tracers    map[string]func(...trace.Option) trace.Tracer
 	Version    string
 
@@ -141,6 +144,13 @@ func Store(s *store.Store) Option {
 	}
 }
 
+func Stream(s *events.Stream) Option {
+	return func(o *Options) {
+		o.Stream = s
+		events.DefaultStream = *s
+	}
+}
+
 func Tracer(t *trace.Tracer) Option {
 	return func(o *Options) {
 		o.Tracer = t
@@ -169,6 +179,13 @@ func NewBroker(name string, b func(...broker.Option) broker.Broker) Option {
 	}
 }
 
+// New stream func.
+func NewStream(name string, b func(...events.Option) events.Stream) Option {
+	return func(o *Options) {
+		o.Streams[name] = b
+	}
+}
+
 // New cache func.
 func NewCache(name string, c func(...cache.Option) cache.Cache) Option {
 	return func(o *Options) {

events/natsjs/README.md

@@ -1,3 +1,48 @@
 # NATS JetStream
 
 This plugin uses NATS with JetStream to send and receive events.
+
+## Create a stream
+
+```go
+ev, err := natsjs.NewStream(
+  natsjs.Address("nats://10.0.1.46:4222"),
+  natsjs.MaxAge(24*160*time.Minute),
+)
+```
+
+## Consume a stream
+
+```go
+ee, err := events.Consume("test",
+  events.WithAutoAck(false, time.Second*30),
+  events.WithGroup("testgroup"),
+)
+if err != nil {
+  panic(err)
+}
+go func() {
+  for {
+    msg := <-ee
+    // Process the message
+    logger.Info("Received message:", string(msg.Payload))
+    err := msg.Ack()
+    if err != nil {
+      logger.Error("Error acknowledging message:", err)
+    } else {
+      logger.Info("Message acknowledged")
+    }
+  }
+}()
+
+```
+
+## Publish an Event to the stream
+
+```go
+err = ev.Publish("test", []byte("hello world"))
+if err != nil {
+  panic(err)
+}
+```
+

events/natsjs/nats.go

@@ -179,15 +179,23 @@ func (s *stream) Consume(topic string, opts ...events.ConsumeOption) (<-chan eve
 			// not acknowledging the message is the way to indicate an error occurred
 			return
 		}
-
-		if !options.AutoAck {
+		if options.AutoAck {
 			// set up the ack funcs
 			evt.SetAckFunc(func() error {
 				return msg.Ack()
 			})
+
 			evt.SetNackFunc(func() error {
 				return msg.Nak()
 			})
+		} else {
+			// set up the ack funcs
+			evt.SetAckFunc(func() error {
+				return nil
+			})
+			evt.SetNackFunc(func() error {
+				return nil
+			})
 		}
 
 		// push onto the channel and wait for the consumer to take the event off before we acknowledge it.
@@ -198,6 +206,7 @@ func (s *stream) Consume(topic string, opts ...events.ConsumeOption) (<-chan eve
 		}
 
 		if err := msg.Ack(nats.Context(ctx)); err != nil {
+
 			log.Logf(logger.ErrorLevel, "Error acknowledging message: %v", err)
 		}
 	}
@@ -208,6 +217,12 @@ func (s *stream) Consume(topic string, opts ...events.ConsumeOption) (<-chan eve
 		cfg := &nats.StreamConfig{
 			Name: topic,
 		}
+		if s.opts.RetentionPolicy != 0 {
+			cfg.Retention = nats.RetentionPolicy(s.opts.RetentionPolicy)
+		}
+		if s.opts.MaxAge > 0 {
+			cfg.MaxAge = s.opts.MaxAge
+		}
 
 		_, err = s.natsJetStreamCtx.AddStream(cfg)
 		if err != nil {
@@ -223,7 +238,7 @@ func (s *stream) Consume(topic string, opts ...events.ConsumeOption) (<-chan eve
 	}
 
 	if options.AutoAck {
-		subOpts = append(subOpts, nats.AckNone())
+		subOpts = append(subOpts, nats.AckAll())
 	} else {
 		subOpts = append(subOpts, nats.AckExplicit())
 	}

events/natsjs/options.go

@@ -2,6 +2,7 @@ package natsjs
 
 import (
 	"crypto/tls"
+	"time"
 
 	"go-micro.dev/v5/logger"
 )
@@ -19,6 +20,9 @@ type Options struct {
 	DisableDurableStreams bool
 	Username              string
 	Password              string
+	RetentionPolicy       int
+	MaxAge                time.Duration
+	MaxMsgSize            int
 }
 
 // Option is a function which configures options.
@@ -94,3 +98,19 @@ func Authenticate(username, password string) Option {
 		o.Password = password
 	}
 }
+func RetentionPolicy(rp int) Option {
+	return func(o *Options) {
+		o.RetentionPolicy = rp
+	}
+}
+
+func MaxMsgSize(size int) Option {
+	return func(o *Options) {
+		o.MaxMsgSize = size
+	}
+}
+func MaxAge(age time.Duration) Option {
+	return func(o *Options) {
+		o.MaxAge = age
+	}
+}

profile/profile.go

@@ -8,6 +8,8 @@ import (
 	natslib "github.com/nats-io/nats.go"
 	"go-micro.dev/v5/broker"
 	"go-micro.dev/v5/broker/nats"
+	"go-micro.dev/v5/events"
+	nevents "go-micro.dev/v5/events/natsjs"
 	"go-micro.dev/v5/registry"
 	nreg "go-micro.dev/v5/registry/nats"
 	"go-micro.dev/v5/store"
@@ -22,34 +24,46 @@ type Profile struct {
 	Broker    broker.Broker
 	Store     store.Store
 	Transport transport.Transport
+	Stream    events.Stream
 }
 
 // LocalProfile returns a profile with local mDNS as the registry, HTTP as the broker, file as the store, and HTTP as the transport
 // It is used for local development and testing
-func LocalProfile() Profile {
+func LocalProfile() (Profile, error) {
+	stream, err := events.NewStream()
 	return Profile{
 		Registry:  registry.NewMDNSRegistry(),
 		Broker:    broker.NewHttpBroker(),
 		Store:     store.NewFileStore(),
 		Transport: transport.NewHTTPTransport(),
-	}
+		Stream:    stream,
+	}, err
 }
 
 // NatsProfile returns a profile with NATS as the registry, broker, store, and transport
 // It uses the environment variable MICR_NATS_ADDRESS to set the NATS server address
 // If the variable is not set, it defaults to nats://0.0.0.0:4222 which will connect to a local NATS server
-func NatsProfile() Profile {
+func NatsProfile() (Profile, error) {
 	addr := os.Getenv("MICRO_NATS_ADDRESS")
 	if addr == "" {
 		addr = "nats://0.0.0.0:4222"
 	}
 	// Split the address by comma, trim whitespace, and convert to a slice of strings
 	addrs := splitNatsAdressList(addr)
+
 	reg := nreg.NewNatsRegistry(registry.Addrs(addrs...))
-	brok := nats.NewNatsBroker(broker.Addrs(addrs...))
+
+	nopts := natslib.GetDefaultOptions()
+	nopts.Servers = addrs
+	brok := nats.NewNatsBroker(broker.Addrs(addrs...), nats.Options(nopts))
+
 	st := nstore.NewStore(nstore.NatsOptions(natslib.Options{Servers: addrs}))
 	tx := ntx.NewTransport(ntx.Options(natslib.Options{Servers: addrs}))
 
+	stream, err := nevents.NewStream(
+		nevents.Address(addr),
+	)
+
 	registry.DefaultRegistry = reg
 	broker.DefaultBroker = brok
 	store.DefaultStore = st
@@ -59,7 +73,8 @@ func NatsProfile() Profile {
 		Broker:    brok,
 		Store:     st,
 		Transport: tx,
-	}
+		Stream:    stream,
+	}, err
 }
 
 func splitNatsAdressList(addr string) []string {