kernel v0.20.1 with minor improvements (#94)

bag.go

@@ -25,6 +25,9 @@ type Msg[T any] struct {
 	// Error on the message processing
 	Error error
 
+	// I/O Context of the message, as obtained from broker
+	IOContext any
+
 	// Message decoded content
 	Object T
 }
@@ -47,16 +50,20 @@ type Bag struct {
 	// Error on the message processing
 	Error error
 
+	// I/O Context of the message, as obtained from broker
+	IOContext any
+
 	// Message raw content
 	Object []byte
 }
 
 func ToMsg[T any](bag Bag, object T) Msg[T] {
 	return Msg[T]{
-		Category: bag.Category,
-		Digest:   bag.Digest,
-		Error:    bag.Error,
-		Object:   object,
+		Category:  bag.Category,
+		Digest:    bag.Digest,
+		Error:     bag.Error,
+		IOContext: bag.IOContext,
+		Object:    object,
 	}
 }
 

config.go

@@ -17,6 +17,13 @@ import (
 	"github.com/fogfish/swarm/kernel/backoff"
 )
 
+// Environment variable to config kernel
+const (
+	EnvConfigPollFrequency  = "CONFIG_SWARM_POLL_FREQUENCY"
+	EnvConfigTimeToFlight   = "CONFIG_SWARM_TIME_TO_FLIGHT"
+	EnvConfigNetworkTimeout = "CONFIG_SWARM_NETWORK_TIMEOUT"
+)
+
 // Grade of Service Policy
 type Policy int
 
@@ -64,22 +71,26 @@ type Config struct {
 	// Timeout for any network operations
 	NetworkTimeout time.Duration
 
-	// Codec for binary packets
-	Codec Codec
+	// Fail fast the message if category is not known to kernel.
+	FailOnUnknownCategory bool
+
+	// PacketCodec for binary packets
+	PacketCodec Codec
 }
 
 func NewConfig() Config {
 	return Config{
-		Source:         "github.com/fogfish/swarm",
-		Policy:         PolicyAtLeastOnce,
-		CapOut:         0,
-		CapDlq:         0,
-		CapRcv:         0,
-		CapAck:         0,
-		Backoff:        backoff.Exp(10*time.Millisecond, 10, 0.5),
-		PollFrequency:  10 * time.Millisecond,
-		TimeToFlight:   5 * time.Second,
-		NetworkTimeout: 5 * time.Second,
+		Source:                "github.com/fogfish/swarm",
+		Policy:                PolicyAtLeastOnce,
+		CapOut:                0,
+		CapDlq:                0,
+		CapRcv:                0,
+		CapAck:                0,
+		Backoff:               backoff.Exp(10*time.Millisecond, 10, 0.5),
+		PollFrequency:         10 * time.Millisecond,
+		TimeToFlight:          5 * time.Second,
+		NetworkTimeout:        5 * time.Second,
+		FailOnUnknownCategory: false,
 	}
 }
 
@@ -179,9 +190,9 @@ func WithNetworkTimeout(t time.Duration) Option {
 // - CONFIG_SWARM_NETWORK_TIMEOUT
 func WithConfigFromEnv() Option {
 	return func(conf *Config) {
-		conf.PollFrequency = durationFromEnv("CONFIG_SWARM_POLL_FREQUENCY", conf.PollFrequency)
-		conf.TimeToFlight = durationFromEnv("CONFIG_SWARM_TIME_TO_FLIGHT", conf.TimeToFlight)
-		conf.NetworkTimeout = durationFromEnv("CONFIG_SWARM_NETWORK_TIMEOUT", conf.NetworkTimeout)
+		conf.PollFrequency = durationFromEnv(EnvConfigPollFrequency, conf.PollFrequency)
+		conf.TimeToFlight = durationFromEnv(EnvConfigTimeToFlight, conf.TimeToFlight)
+		conf.NetworkTimeout = durationFromEnv(EnvConfigNetworkTimeout, conf.NetworkTimeout)
 	}
 }
 
@@ -225,3 +236,17 @@ func WithPolicyAtLeastOnce(n int) Option {
 		conf.CapAck = n
 	}
 }
+
+// Configure codec for binary packets
+func WithPacketCodec(codec Codec) Option {
+	return func(conf *Config) {
+		conf.PacketCodec = codec
+	}
+}
+
+// Fail fast the message if category is not known to kernel.
+func WithFailOnUnknownCategory() Option {
+	return func(conf *Config) {
+		conf.FailOnUnknownCategory = true
+	}
+}

dequeue/dequeue.go

@@ -33,8 +33,8 @@ func Event[M, T any](q *kernel.Dequeuer, category ...string) (<-chan swarm.Msg[s
 
 // Create pair of channels to receive and acknowledge pure binary
 func Bytes(q *kernel.Dequeuer, cat string) (<-chan swarm.Msg[[]byte], chan<- swarm.Msg[[]byte]) {
-	if q.Config.Codec != nil {
-		return kernel.Dequeue(q, cat, q.Config.Codec)
+	if q.Config.PacketCodec != nil {
+		return kernel.Dequeue(q, cat, q.Config.PacketCodec)
 	}
 
 	return kernel.Dequeue(q, cat, encoding.NewCodecByte())

enqueue/enqueue.go

@@ -33,8 +33,8 @@ func Event[M, T any](q *kernel.Enqueuer, category ...string) (snd chan<- swarm.E
 
 // Create pair of channels to emit pure binaries
 func Bytes(q *kernel.Enqueuer, cat string) (snd chan<- []byte, dlq <-chan []byte) {
-	if q.Config.Codec != nil {
-		return kernel.Enqueue(q, cat, q.Config.Codec)
+	if q.Config.PacketCodec != nil {
+		return kernel.Enqueue(q, cat, q.Config.PacketCodec)
 	}
 
 	return kernel.Enqueue(q, cat, encoding.NewCodecByte())

enqueue/writer.go

@@ -119,8 +119,8 @@ type EmitterBytes struct {
 // Creates synchronous emitter
 func NewBytes(q *kernel.Enqueuer, category string) *EmitterBytes {
 	var codec swarm.Codec
-	if q.Config.Codec != nil {
-		codec = q.Config.Codec
+	if q.Config.PacketCodec != nil {
+		codec = q.Config.PacketCodec
 	} else {
 		codec = encoding.NewCodecByte()
 	}

errors.go

@@ -8,10 +8,35 @@
 
 package swarm
 
-import "github.com/fogfish/faults"
+import (
+	"fmt"
+	"time"
+
+	"github.com/fogfish/faults"
+)
 
 const (
 	ErrServiceIO = faults.Type("service i/o failed")
 	ErrEnqueue   = faults.Type("enqueue is failed")
 	ErrDequeue   = faults.Type("dequeue is failed")
 )
+
+type errTimeout struct {
+	op    string
+	timer time.Duration
+}
+
+func ErrTimeout(op string, timer time.Duration) error {
+	return errTimeout{
+		op:    op,
+		timer: timer,
+	}
+}
+
+func (err errTimeout) Error() string {
+	return fmt.Sprintf("timeout %s after %s", err.op, err.timer)
+}
+
+func (err errTimeout) Timeout() time.Duration {
+	return err.timer
+}

kernel/bridge.go

@@ -10,7 +10,6 @@ package kernel
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/fogfish/swarm"
@@ -52,7 +51,7 @@ func (s *Bridge) Dispatch(seq []swarm.Bag) error {
 	case err := <-s.session:
 		return err
 	case <-time.After(s.timeToFlight):
-		return fmt.Errorf("ack timeout")
+		return swarm.ErrTimeout("ack", s.timeToFlight)
 	}
 }
 

kernel/cathode.go

@@ -10,6 +10,7 @@ package kernel
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"sync"
 	"time"
@@ -102,6 +103,13 @@ func (k *Dequeuer) receive() {
 					k.Config.StdErr <- err
 					return
 				}
+			} else {
+				if k.Config.FailOnUnknownCategory {
+					slog.Error("Unknown category", "cat", bag.Category, "kernel", k.Config.Source)
+					k.Cathode.Err(k.context, bag.Digest, swarm.ErrDequeue.New(fmt.Errorf("unknown category %s ", bag.Category)))
+				} else {
+					slog.Warn("Unknown category", "cat", bag.Category, "kernel", k.Config.Source)
+				}
 			}
 		}
 	}

kernel/cathode_test.go

@@ -22,7 +22,14 @@ func TestDequeuer(t *testing.T) {
 	codec := encoding.NewCodecJson[string]()
 	none := mockCathode(nil, nil)
 	pass := mockCathode(make(chan string),
-		[]swarm.Bag{{Category: "test", Digest: "1", Object: []byte(`"1"`)}},
+		[]swarm.Bag{
+			{
+				Category:  "test",
+				Digest:    "1",
+				IOContext: "context",
+				Object:    []byte(`"1"`),
+			},
+		},
 	)
 
 	t.Run("Kernel", func(t *testing.T) {
@@ -60,6 +67,21 @@ func TestDequeuer(t *testing.T) {
 		k.Close()
 	})
 
+	t.Run("Dequeue.1.Context", func(t *testing.T) {
+		k := NewDequeuer(pass, swarm.Config{PollFrequency: 10 * time.Millisecond})
+		rcv, ack := Dequeue(k, "test", codec)
+		go k.Await()
+
+		msg := <-rcv
+		ack <- msg
+		it.Then(t).Should(
+			it.Equal(string(<-pass.ack), `1`),
+			it.Equal(msg.IOContext.(string), "context"),
+		)
+
+		k.Close()
+	})
+
 	t.Run("Backlog", func(t *testing.T) {
 		k := NewDequeuer(pass, swarm.Config{CapAck: 4, PollFrequency: 1 * time.Millisecond})
 		rcv, ack := Dequeue(k, "test", codec)

version.go

@@ -8,4 +8,4 @@
 
 package swarm
 
-const Version = "v0.20.0-alpha"
+const Version = "v0.20.1"