clean up

Author: blinkinglight

event.go

@@ -90,129 +90,3 @@ func Event(ctx context.Context, fn ManagerEventApplier, opts ...eo.Options) erro
 	<-ctx.Done()
 	return nil
 }
-
-/*
-type ProcMgr struct {
-	js        nats.JetStreamContext
-	nc        *nats.Conn
-	aggregate string
-	subject   string
-	durable   string
-	handler   ManagerEventApplier
-	cfg       *eo.Config
-}
-
-// Event is a process manager that listens for events and depending on situaion emits command events.
-func Event(ctx context.Context, handler ManagerEventApplier, opts ...eo.Options) {
-	cfg := &eo.Config{}
-
-	for _, opt := range opts {
-		opt(cfg)
-	}
-
-	if cfg.Aggregate == "" {
-		panic("Aggregate name is required for command processor")
-	}
-
-	subject := fmt.Sprintf(CommandsPrefix+".%s", cfg.Aggregate)
-	if cfg.Subject != "" {
-		subject = cfg.Subject
-	}
-
-	nc, _ := Nats(ctx)
-	js, _ := JetStream(ctx)
-	cp := &ProcMgr{js: js, nc: nc, subject: subject, aggregate: cfg.Aggregate, durable: "default", handler: handler, cfg: cfg}
-	c, cancel := context.WithCancel(ctx)
-	go func() {
-		for {
-			err := cp.init(c, cancel)
-			if err != nil {
-				log.Printf("Error initializing command processor: %v", err)
-				continue
-			}
-		}
-	}()
-	<-c.Done()
-}
-
-func (cp *ProcMgr) init(ctx context.Context, cancel context.CancelFunc) error {
-
-	_, _ = cp.js.AddStream(&nats.StreamConfig{
-		Name:       commandsStream,
-		Subjects:   []string{CommandsPrefix + ".>"},
-		Retention:  nats.WorkQueuePolicy,
-		Storage:    nats.FileStorage,
-		Replicas:   1,
-		Duplicates: 5 * time.Minute,
-	})
-
-	_, err := cp.js.AddConsumer(commandsStream, &nats.ConsumerConfig{
-		Name:          cp.aggregate + "_" + cp.durable + "_cmd",
-		Durable:       cp.aggregate + "_" + cp.durable + "_cmd",
-		FilterSubject: cp.subject,
-		AckPolicy:     nats.AckExplicitPolicy,
-	})
-	if err != nil {
-		return err
-	}
-
-	sub, err := cp.js.PullSubscribe(cp.subject, cp.aggregate+"_"+cp.durable+"_cmd", nats.BindStream(commandsStream), nats.ManualAck(), nats.AckExplicit(), nats.DeliverAll())
-
-	if err != nil {
-		log.Printf("Error subscribing to commands: %v", err)
-		cancel()
-	}
-	defer sub.Unsubscribe()
-
-	for {
-		if ctx.Err() != nil {
-			cancel()
-			return nil
-		}
-		msg, _ := sub.Fetch(1, nats.MaxWait(60*time.Second))
-		if len(msg) == 0 {
-			time.Sleep(50 * time.Millisecond)
-			continue
-		}
-		for _, msg := range msg {
-			var evt gen.EventEnvelope
-			err := proto.Unmarshal(msg.Data, &evt)
-			if err != nil {
-				log.Printf("Error unmarshalling message: %v", err)
-				msg.Ack()
-				continue
-			}
-
-			commands, err := cp.handler.Handle(&evt)
-			if err != nil {
-				msg.Ack()
-				continue
-			}
-
-			for _, command := range commands {
-				if command.Aggregate == "" {
-					command.Aggregate = evt.AggregateType
-				}
-				if command.AggregateId == "" {
-					command.AggregateId = evt.AggregateId
-				}
-				eventSubject := fmt.Sprintf(CommandsPrefix+".%s.%s.%s", command.Aggregate, command.AggregateId, command.CommandType)
-				if len(command.Parents) > 0 {
-					var parents []string
-					for _, parent := range command.Parents {
-						parents = append(parents, fmt.Sprintf("%s.%s", parent.AggregateType, parent.AggregateId))
-					}
-					eventSubject = fmt.Sprintf("events.%s.%s.%s.%s", strings.Join(parents, "."), command.Aggregate, command.AggregateId, command.CommandType)
-				}
-
-				b, _ := proto.Marshal(command)
-				if _, err := cp.js.Publish(eventSubject, b); err != nil {
-					log.Printf("Error publishing event %v", err)
-				}
-			}
-
-			_ = msg.Ack()
-		}
-	}
-}
-*/

projector.go

@@ -3,6 +3,7 @@ package bee
 import (
 	"context"
 	"fmt"
+	"log"
 
 	"github.com/blinkinglight/bee/gen"
 	"github.com/blinkinglight/bee/po"
@@ -44,7 +45,7 @@ func Project(ctx context.Context, fn EventApplier, opts ...po.Options) error {
 	}
 
 	if cfg.Aggregate == "" {
-		panic("aggregate is required for projection")
+		return fmt.Errorf("aggregate is required for projection")
 	}
 
 	if cfg.DurableName == "" {
@@ -60,9 +61,12 @@ func Project(ctx context.Context, fn EventApplier, opts ...po.Options) error {
 		prefix = cfg.Prefix + "_"
 	}
 
-	js, _ := JetStream(ctx)
+	js, ok := JetStream(ctx)
+	if !ok {
+		return fmt.Errorf("JetStream not available in context")
+	}
 
-	js.AddStream(&nats.StreamConfig{
+	_, _ = js.AddStream(&nats.StreamConfig{
 		Name:      "EVENTS",
 		Subjects:  []string{EventsPrefix + ".>"},
 		Storage:   nats.FileStorage,
@@ -77,19 +81,24 @@ func Project(ctx context.Context, fn EventApplier, opts ...po.Options) error {
 		}
 		m := &gen.EventEnvelope{}
 		if err := proto.Unmarshal(msg.Data, m); err != nil {
+			log.Printf("Error unmarshalling event: aggregate=%s, aggregateID=%s, eventType=%s, error=%v",
+				cfg.Aggregate, cfg.AggregateID, msg.Subject, err)
 			msg.Ack()
 			return
 		}
 
 		if err := fn.ApplyEvent(m); err != nil {
+			log.Printf("Error applying event: aggregate=%s, aggregateID=%s, eventType=%s, error=%v",
+				m.AggregateType, m.AggregateId, m.EventType, err)
 			msg.Ack()
 			return
 		}
 		msg.Ack()
 	}, nats.DeliverAll(), nats.ManualAck(), nats.Durable("events_"+prefix+cfg.DurableName), nats.BindStream("EVENTS"), nats.ConsumerName(cfg.DurableName))
 	if err != nil {
 		fmt.Printf("Error subscribing to events: %v\n", err)
-		return err
+		return fmt.Errorf("projector: failed to subscribe to subject %s: %w", cfg.Subject, err)
+
 	}
 	defer sub.Unsubscribe()
 

replay.go

@@ -27,7 +27,7 @@ type ReplayHandler interface {
 // ro.WithStartSeq - start from event (if you have snapshot)
 // ro.WtihParent - nests subjects
 // ro.WithTimeout - timeout if no events for stream
-func Replay(ctx context.Context, fn ReplayHandler, opts ...ro.Options) {
+func Replay(ctx context.Context, fn ReplayHandler, opts ...ro.Options) error {
 
 	cfg := &ro.Config{
 		StartSeq: DeliverAll,
@@ -65,17 +65,17 @@ func Replay(ctx context.Context, fn ReplayHandler, opts ...ro.Options) {
 	}, opt, nats.ManualAck())
 	if err != nil {
 		cancel()
-		return
+		return fmt.Errorf("projector: failed to subscribe to subject %s: %w", subject, err)
 	}
 	num, _, err := sub.MaxPending()
 	if err != nil {
 		cancel()
-		return
+		return fmt.Errorf("projector: failed to get max pending for subject %s: %w", subject, err)
 	}
 	_ = num
 	if num <= 0 {
 		cancel()
-		return
+		return fmt.Errorf("projector: no events found for subject %s", subject)
 	}
 
 	defer close(msgs)
@@ -109,6 +109,7 @@ func Replay(ctx context.Context, fn ReplayHandler, opts ...ro.Options) {
 		}
 	}()
 	<-lctx.Done()
+	return nil
 }
 
 // ReplayAndSubscribe replays events for a given aggregate and aggregate ID,