chore: checkin

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

Author: lxfontes

x/wasmbus/bus.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"net/http"
 	"strings"
+
+	"github.com/nats-io/nats.go"
 )
 
 const (
@@ -113,3 +115,7 @@ func NewMessage(subject string) *Message {
 		Subject: subject,
 	}
 }
+
+func NewInbox() string {
+	return nats.NewInbox()
+}

x/wasmbus/client.go

@@ -6,7 +6,7 @@ import (
 	"errors"
 	"fmt"
 
-	"gopkg.in/yaml.v2"
+	yaml "github.com/goccy/go-yaml"
 )
 
 // LatticeRequest encodes the Roundtrip logic for a Bus Request
@@ -74,18 +74,22 @@ func (l *LatticeRequest[T, Y]) Execute(ctx context.Context) (*Y, error) {
 // Encode marshals the payload into a Message.
 // The payload is encoded as json.
 func Encode(subject string, payload any) (*Message, error) {
+	var err error
 	wasmbusMsg := NewMessage(subject)
 	wasmbusMsg.Header.Set("Content-Type", "application/json")
+	wasmbusMsg.Data, err = EncodeMimetype(payload, "application/json")
+	return wasmbusMsg, err
+}
 
-	if payload != nil {
-		var err error
-		wasmbusMsg.Data, err = json.Marshal(payload)
-		if err != nil {
-			return nil, fmt.Errorf("%w: %s", ErrInternal, err)
-		}
+func EncodeMimetype(payload any, mimeType string) ([]byte, error) {
+	switch mimeType {
+	case "application/json", "":
+		return json.Marshal(payload)
+	case "application/yaml":
+		return yaml.Marshal(payload)
+	default:
+		return nil, errors.New("unsupported content type")
 	}
-
-	return wasmbusMsg, nil
 }
 
 // Decode unmarshals the raw response data into the provided struct.

x/wasmbus/cmd/wasmcloud-provider-shim/main.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/joho/godotenv"
+	"github.com/urfave/cli/v3"
+	"go.wasmcloud.dev/wasmbus/control"
+)
+
+func configCommand() *cli.Command {
+	var targetName string
+	var nameArg = &cli.StringArg{
+		Name:        "name",
+		Destination: &targetName,
+		Max:         1,
+	}
+	return &cli.Command{
+		Name:  "config",
+		Usage: "Interact with Lattice Config",
+		Flags: []cli.Flag{},
+		Commands: []*cli.Command{
+			{
+				Name:   "get",
+				Usage:  "Get a config",
+				Action: wrapNamedAction(getConfigCommand, &targetName),
+				Arguments: []cli.Argument{
+					nameArg,
+				},
+			},
+			{
+				Name:   "delete",
+				Usage:  "Delete a config",
+				Action: wrapNamedAction(deleteConfigCommand, &targetName),
+				Arguments: []cli.Argument{
+					nameArg,
+				},
+			},
+			{
+				Name:   "put",
+				Usage:  "Put a config",
+				Action: putConfigCommand,
+				Flags: []cli.Flag{
+					&cli.StringFlag{
+						Name:     "name",
+						Usage:    "Name of the config to store",
+						Required: true,
+					},
+					&cli.StringFlag{
+						Name:     "file",
+						Aliases:  []string{"f"},
+						Usage:    "File to read config from, in dotenv format (one KEY=VALUE per line)",
+						Required: true,
+					},
+				},
+				Arguments: []cli.Argument{
+					nameArg,
+				},
+			},
+		},
+	}
+}
+
+func deleteConfigCommand(ctx context.Context, cmd *cli.Command, name string) error {
+	client, err := controlClientFromCommand(cmd)
+	if err != nil {
+		return err
+	}
+
+	resp, err := client.ConfigDelete(ctx, &control.ConfigDeleteRequest{
+		Name: name,
+	})
+	if err != nil {
+		return err
+	}
+	if !resp.Success {
+		return fmt.Errorf("received error response: %s", resp.Message)
+	}
+
+	fmt.Println(titleStyle.Render("⁜", name), "deleted")
+
+	return nil
+}
+
+func getConfigCommand(ctx context.Context, cmd *cli.Command, name string) error {
+	client, err := controlClientFromCommand(cmd)
+	if err != nil {
+		return err
+	}
+
+	resp, err := client.ConfigGet(ctx, &control.ConfigGetRequest{
+		Name: name,
+	})
+	if err != nil {
+		return err
+	}
+	if !resp.Success {
+		return fmt.Errorf("received error response: %s", resp.Message)
+	}
+
+	// NOTE(lxf): This feels like a bug in hosts' crates/host/src/wasmbus/mod.rs
+	if resp.Message == "Configuration not found" {
+		return fmt.Errorf("configuration not found")
+	}
+
+	config, err := godotenv.Marshal(resp.Response)
+	if err != nil {
+		return err
+	}
+	fmt.Println(config)
+
+	return nil
+}
+
+func putConfigCommand(ctx context.Context, cmd *cli.Command) error {
+	name := cmd.String("name")
+	client, err := controlClientFromCommand(cmd)
+	if err != nil {
+		return err
+	}
+
+	f, err := os.ReadFile(cmd.String("file"))
+	if err != nil {
+		return err
+	}
+
+	var values map[string]string
+	if values, err = godotenv.UnmarshalBytes(f); err != nil {
+		return err
+	}
+
+	resp, err := client.ConfigPut(ctx, &control.ConfigPutRequest{
+		Name:   name,
+		Values: values,
+	})
+	if err != nil {
+		return err
+	}
+	if !resp.Success {
+		return fmt.Errorf("received error response: %s", resp.Message)
+	}
+
+	fmt.Println(titleStyle.Render("⁜", name), "stored")
+
+	return nil
+}

x/wasmbus/cmd/wasmcloud-rpc/config.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/urfave/cli/v3"
+	"go.wasmcloud.dev/wasmbus"
+	"go.wasmcloud.dev/wasmbus/events"
+)
+
+func eventCommand() *cli.Command {
+	return &cli.Command{
+		Name:  "events",
+		Usage: "listen for lattice events",
+		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name:  "pattern",
+				Value: wasmbus.PatternAll,
+				Usage: "The event pattern to subscribe to",
+			},
+			&cli.IntFlag{
+				Name:  "backlog",
+				Value: wasmbus.NoBackLog,
+				Usage: "Bus backlog size. Default is no backlog",
+			},
+		},
+		Action: func(ctx context.Context, cmd *cli.Command) error {
+			bus, err := busFromCommand(cmd)
+			if err != nil {
+				return err
+			}
+
+			lattice := cmd.String("lattice")
+			pattern := cmd.String("pattern")
+			backlog := cmd.Int("backlog")
+
+			callback := func(_ context.Context, ev events.Event) {
+				jsonEv, err := json.MarshalIndent(ev.BusEvent, "", "  ")
+				if err != nil {
+					log.Printf("Error marshalling event: %s", err)
+					return
+				}
+
+				fmt.Println(titleStyle.Render("⁜", ev.CloudEvent.Type()))
+				renderKv("Time", ev.CloudEvent.Time().Format(time.RFC3339))
+				renderKv("Source", ev.CloudEvent.Source())
+				renderKv("Id", ev.CloudEvent.ID())
+				fmt.Println(string(jsonEv))
+			}
+			subscription, err := events.Subscribe(bus, lattice, pattern, int(backlog), events.DiscardErrorsHandler(callback))
+			if err != nil {
+				return err
+			}
+			defer subscription.Drain()
+
+			log.Printf("Listening for events on lattice '%s' with pattern '%s'", lattice, pattern)
+			<-ctx.Done()
+			log.Printf("Shutting down event listener")
+
+			return nil
+		},
+	}
+}