feat: wasmbus

Signed-off-by: Lucas Fontes <[email protected]>

Author: lxfontes

x/wasmbus/bus.go

@@ -0,0 +1,115 @@
+package wasmbus
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"strings"
+)
+
+const (
+	// PatternAll is a wildcard pattern that matches all subjects.
+	PatternAll = "*"
+
+	// NoBackLog is used to indicate that the subscription should not have a backlog.
+	NoBackLog = 0
+
+	// PrefixWadm is the prefix for WasmCloud Admin API.
+	PrefixWadm = "wadm.api"
+	// PrefixEvents is the prefix for Lattice Events.
+	PrefixEvents = "wasmbus.evt"
+	// PrefixControl is the prefix for Lattice RPC.
+	PrefixCtlV1 = "wasmbus.ctl.v1"
+)
+
+var (
+	// ErrEncode is returned when encoding a message fails.
+	ErrEncode = errors.New("encode error")
+	// ErrInternal is returned when an internal error occurs.
+	ErrInternal = errors.New("internal error")
+	// ErrDecode is returned when decoding a message fails.
+	ErrDecode = errors.New("decode error")
+	// ErrTransport is returned when a transport error occurs.
+	ErrTransport = errors.New("transport error")
+	// ErrOperation is returned when an operation error occurs.
+	ErrOperation = errors.New("operation error")
+	// ErrValidation is returned when a validation error occurs.
+	ErrValidation = errors.New("validation error")
+)
+
+// Bus is the interface for the message bus.
+// It provides methods for subscribing to messages and sending messages.
+// It doesn't hold any state and is safe for concurrent use. See `Subscription` for stateful operations.
+// Modeled after the NATS API.
+type Bus interface {
+	// Subscribe creates a subscription for the given subject.
+	// The backlog parameter is the maximum number of messages that can be buffered in memory.
+	Subscribe(subject string, backlog int) (Subscription, error)
+	// QueueSubscribe creates a subscription for the given subject and queue group.
+	// The backlog parameter is the maximum number of messages that can be buffered in memory.
+	QueueSubscribe(subject string, queue string, backlog int) (Subscription, error)
+	// Request sends a request message and waits for a reply.
+	// The context is used for the request timeout.
+	Request(ctx context.Context, msg *Message) (*Message, error)
+	// Publish sends a message to `msg.Subject`.
+	Publish(msg *Message) error
+}
+
+// Subscription is the interface for a message subscription.
+// It provides methods for handling messages and draining the subscription.
+// Subscriptions run in their own goroutine.
+type Subscription interface {
+	// Handle starts the subscription and calls the callback for each message.
+	// Does not block.
+	Handle(callback SubscriptionCallback)
+	// Drain stops the subscription and closes the channel.
+	// Blocks until all messages are processed, releasing the Subscription Thread.
+	Drain() error
+}
+
+// SubscriptionCallback is the callback type for message subscriptions.
+// Subscriptions are single threaded and the callback is called sequentially for each message.
+// Callers should handle concurrency and synchronization themselves.
+type SubscriptionCallback func(msg *Message)
+
+// Header is the Message header type.
+type Header = http.Header
+
+// Message is the message type for the message bus.
+// Modeled after the NATS message type.
+type Message struct {
+	Subject string
+	Reply   string
+	Header  Header
+	Data    []byte
+
+	// For replies
+	bus Bus
+}
+
+// Bus returns the bus that the message was received on or sent to
+// Might be null.
+func (m *Message) Bus() Bus {
+	return m.bus
+}
+
+// LastSubjectPart returns the last part of the subject.
+// Example: "a.b.c" -> "c"
+func (m *Message) LastSubjectPart() string {
+	parts := m.SubjectParts()
+	return parts[len(parts)-1]
+}
+
+// SubjectParts returns the parts of the subject.
+// Example: "a.b.c" -> ["a", "b", "c"]
+func (m *Message) SubjectParts() []string {
+	return strings.Split(m.Subject, ".")
+}
+
+// NewMessage creates a new message with the given subject.
+func NewMessage(subject string) *Message {
+	return &Message{
+		Header:  make(Header),
+		Subject: subject,
+	}
+}

x/wasmbus/client.go

@@ -0,0 +1,109 @@
+package wasmbus
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"gopkg.in/yaml.v2"
+)
+
+// LatticeRequest encodes the Roundtrip logic for a Bus Request
+// This is a generic implementation that can be used with any Bus
+// and any Request/Response pair.
+// Requests are encoded in json and Responses can be either json or yaml.
+// Use Pre & Post Request hooks to modify the request/response before/after.
+// The `T` and `Y` types are used to define the Request and Response types.
+// `T` should be passed by reference (pointer) and `Y` by value (struct).
+type LatticeRequest[T any, Y any] struct {
+	// Request should be a reference to the request struct
+	Request T
+	// Response should be a struct that the response will be unmarshaled into
+	// and should be passed by value
+	Response    Y
+	Subject     string
+	Bus         Bus
+	PreRequest  func(context.Context, T, *Message) error
+	PostRequest func(context.Context, *Y, *Message) error
+}
+
+// NewLatticeRequest returns a `LatticeRequest` for a given type.
+// The `T` and `Y` types are used to define the Request and Response types.
+// `T` should be passed by reference (pointer) and `Y` by value (struct).
+func NewLatticeRequest[T any, Y any](bus Bus, subject string, in T, out Y) *LatticeRequest[T, Y] {
+	return &LatticeRequest[T, Y]{
+		Request:  in,
+		Response: out,
+		Bus:      bus,
+		Subject:  subject,
+	}
+}
+
+// Execute sends the request to the bus and returns the response.
+func (l *LatticeRequest[T, Y]) Execute(ctx context.Context) (*Y, error) {
+	req, err := Encode(l.Subject, l.Request)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %s", ErrEncode, err)
+	}
+
+	if l.PreRequest != nil {
+		if err := l.PreRequest(ctx, l.Request, req); err != nil {
+			return nil, fmt.Errorf("%w: %s", ErrOperation, err)
+		}
+	}
+
+	rawResp, err := l.Bus.Request(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %s", ErrTransport, err)
+	}
+
+	if err := Decode(rawResp, &l.Response); err != nil {
+		return nil, fmt.Errorf("%w: %s", ErrDecode, err)
+	}
+
+	if l.PostRequest != nil {
+		if err := l.PostRequest(ctx, &l.Response, rawResp); err != nil {
+			return nil, fmt.Errorf("%w: %s", ErrOperation, err)
+		}
+	}
+
+	return &l.Response, nil
+}
+
+// Encode marshals the payload into a Message.
+// The payload is encoded as json.
+func Encode(subject string, payload any) (*Message, error) {
+	wasmbusMsg := NewMessage(subject)
+	wasmbusMsg.Header.Set("Content-Type", "application/json")
+
+	if payload != nil {
+		var err error
+		wasmbusMsg.Data, err = json.Marshal(payload)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %s", ErrInternal, err)
+		}
+	}
+
+	return wasmbusMsg, nil
+}
+
+// Decode unmarshals the raw response data into the provided struct.
+// The content type is used to determine the unmarshaling format.
+// If the content type is not supported, an error is returned.
+// Acceptable content types are "application/json" and "application/yaml".
+func Decode(rawResp *Message, into any) error {
+	if len(rawResp.Data) == 0 {
+		return nil
+	}
+
+	contentType := rawResp.Header.Get("Content-Type")
+	switch contentType {
+	case "application/json", "":
+		return json.Unmarshal(rawResp.Data, into)
+	case "application/yaml":
+		return yaml.Unmarshal(rawResp.Data, into)
+	default:
+		return errors.New("unsupported content type")
+	}
+}