Ephemeral observer middleware

eventbus/nats/eventbus.go

@@ -25,6 +25,7 @@ import (
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/codec/json"
+	"github.com/looplab/eventhorizon/middleware/eventhandler/ephemeral"
 )
 
 // EventBus is a NATS Jetstream event bus that delegates handling of published
@@ -36,13 +37,15 @@ type EventBus struct {
 	js           nats.JetStreamContext
 	stream       *nats.StreamInfo
 	connOpts     []nats.Option
+	streamConfig *nats.StreamConfig
 	registered   map[eh.EventHandlerType]struct{}
 	registeredMu sync.RWMutex
 	errCh        chan error
 	cctx         context.Context
 	cancel       context.CancelFunc
 	wg           sync.WaitGroup
 	codec        eh.EventCodec
+	unsubscribe  []func()
 }
 
 // NewEventBus creates an EventBus, with optional settings.
@@ -88,9 +91,15 @@ func NewEventBus(url, appID string, options ...Option) (*EventBus, error) {
 	// Create the stream, which stores messages received on the subject.
 	subjects := b.streamName + ".*.*"
 	cfg := &nats.StreamConfig{
-		Name:     b.streamName,
-		Subjects: []string{subjects},
-		Storage:  nats.FileStorage,
+		Name:      b.streamName,
+		Subjects:  []string{subjects},
+		Storage:   nats.FileStorage,
+		Retention: nats.InterestPolicy,
+	}
+
+	// Use the custom stream config if provided.
+	if b.streamConfig != nil {
+		cfg = b.streamConfig
 	}
 
 	if b.stream, err = b.js.AddStream(cfg); err != nil {
@@ -121,6 +130,14 @@ func WithNATSOptions(opts ...nats.Option) Option {
 	}
 }
 
+// WithStreamConfig can customize the config for created NATS JetStream.
+func WithStreamConfig(opts *nats.StreamConfig) Option {
+	return func(b *EventBus) error {
+		b.streamConfig = opts
+		return nil
+	}
+}
+
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (b *EventBus) HandlerType() eh.EventHandlerType {
 	return "eventbus"
@@ -164,7 +181,6 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	consumerName := fmt.Sprintf("%s_%s", b.appID, h.HandlerType())
 
 	sub, err := b.js.QueueSubscribe(subject, consumerName, b.handler(b.cctx, m, h),
-		nats.Durable(consumerName),
 		nats.DeliverNew(),
 		nats.ManualAck(),
 		nats.AckExplicit(),
@@ -175,6 +191,11 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 		return fmt.Errorf("could not subscribe to queue: %w", err)
 	}
 
+	// capture the subscription of ephemeral consumers so we can unsubscribe when we exit.
+	if b.handlerIsEphemeral(h) {
+		b.unsubscribe = append(b.unsubscribe, func() { sub.Unsubscribe() })
+	}
+
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
@@ -191,11 +212,31 @@ func (b *EventBus) Errors() <-chan error {
 	return b.errCh
 }
 
+// handlerIsEphemeral traverses the middleware chain and checks for the
+// ephemeral middleware and quires it's status.
+func (b *EventBus) handlerIsEphemeral(h eh.EventHandler) bool {
+	for {
+		if obs, ok := h.(ephemeral.EphemeralHandler); ok {
+			return obs.IsEphemeralHandler()
+		} else if c, ok := h.(eh.EventHandlerChain); ok {
+			if h = c.InnerHandler(); h != nil {
+				continue
+			}
+		}
+		return false
+	}
+}
+
 // Close implements the Close method of the eventhorizon.EventBus interface.
 func (b *EventBus) Close() error {
 	b.cancel()
 	b.wg.Wait()
 
+	// unsubscribe any ephemeral subscribers we created.
+	for _, unSub := range b.unsubscribe {
+		unSub()
+	}
+
 	b.conn.Close()
 
 	return nil

go.mod

@@ -3,64 +3,63 @@ module github.com/looplab/eventhorizon
 go 1.17
 
 require (
-	cloud.google.com/go/pubsub v1.17.1
-	github.com/go-redis/redis/v8 v8.11.4
+	cloud.google.com/go/pubsub v1.22.0
+	github.com/go-redis/redis/v8 v8.11.5
 	github.com/google/uuid v1.3.0
 	github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75
-	github.com/gorilla/websocket v1.4.2
+	github.com/gorilla/websocket v1.5.0
 	github.com/jinzhu/copier v0.3.4
 	github.com/jpillora/backoff v1.0.0
 	github.com/kr/pretty v0.3.0
-	github.com/nats-io/nats.go v1.13.0
+	github.com/nats-io/nats.go v1.16.0
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/segmentio/kafka-go v0.4.25
 	github.com/uber/jaeger-client-go v2.29.1+incompatible
-	go.mongodb.org/mongo-driver v1.8.0
-	google.golang.org/api v0.61.0
+	go.mongodb.org/mongo-driver v1.9.1
+	google.golang.org/api v0.81.0
 )
 
 require (
-	cloud.google.com/go v0.97.0 // indirect
+	cloud.google.com/go v0.102.0 // indirect
+	cloud.google.com/go/compute v1.6.1 // indirect
+	cloud.google.com/go/iam v0.3.0 // indirect
 	github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
-	github.com/census-instrumentation/opencensus-proto v0.2.1 // indirect
-	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
-	github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403 // indirect
-	github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
-	github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0 // indirect
-	github.com/envoyproxy/protoc-gen-validate v0.1.0 // indirect
 	github.com/go-stack/stack v1.8.0 // indirect
-	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.3 // indirect
-	github.com/google/go-cmp v0.5.6 // indirect
-	github.com/googleapis/gax-go/v2 v2.1.1 // indirect
+	github.com/google/go-cmp v0.5.8 // indirect
+	github.com/googleapis/gax-go/v2 v2.4.0 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.2.0 // indirect
 	github.com/nats-io/nats-server/v2 v2.6.2 // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pierrec/lz4 v2.6.0+incompatible // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/rogpeppe/go-internal v1.6.1 // indirect
-	github.com/stretchr/testify v1.7.0 // indirect
+	github.com/stretchr/objx v0.1.1 // indirect
+	github.com/stretchr/testify v1.7.1 // indirect
 	github.com/uber/jaeger-lib v2.4.1+incompatible // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.0.2 // indirect
 	github.com/xdg-go/stringprep v1.0.2 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	go.opencensus.io v0.23.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
-	golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e // indirect
-	golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420 // indirect
-	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20211124211545-fe61309f8881 // indirect
-	golang.org/x/text v0.3.6 // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/net v0.0.0-20220520000938-2e3eb7b945c2 // indirect
+	golang.org/x/oauth2 v0.0.0-20220524215830-622c5d57e401 // indirect
+	golang.org/x/sync v0.0.0-20220513210516-0976fa681c29 // indirect
+	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/text v0.3.7 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20211118181313-81c1377c94b1 // indirect
-	google.golang.org/grpc v1.40.0 // indirect
-	google.golang.org/protobuf v1.27.1 // indirect
+	google.golang.org/genproto v0.0.0-20220523171625-347a074981d8 // indirect
+	google.golang.org/grpc v1.47.0 // indirect
+	google.golang.org/protobuf v1.28.0 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 )