eventbus for sending events from backend to frontend. stubbing out a block controller and blockservice

main.go

@@ -10,6 +10,8 @@ import (
 	"log"
 
 	"github.com/wavetermdev/thenextwave/pkg/blockstore"
+	"github.com/wavetermdev/thenextwave/pkg/eventbus"
+	"github.com/wavetermdev/thenextwave/pkg/service/blockservice"
 	"github.com/wavetermdev/thenextwave/pkg/service/fileservice"
 	"github.com/wavetermdev/thenextwave/pkg/wavebase"
 
@@ -22,12 +24,6 @@ var assets embed.FS
 //go:embed build/appicon.png
 var appIcon []byte
 
-type GreetService struct{}
-
-func (g *GreetService) Greet(name string) string {
-	return "Hello " + name + "!"
-}
-
 func main() {
 	err := wavebase.EnsureWaveHomeDir()
 	if err != nil {
@@ -45,8 +41,8 @@ func main() {
 		Name:        "NextWave",
 		Description: "The Next Wave Terminal",
 		Bind: []any{
-			&GreetService{},
 			&fileservice.FileService{},
+			&blockservice.BlockService{},
 		},
 		Icon: appIcon,
 		Assets: application.AssetOptions{
@@ -56,8 +52,9 @@ func main() {
 			ApplicationShouldTerminateAfterLastWindowClosed: true,
 		},
 	})
+	eventbus.RegisterWailsApp(app)
 
-	app.NewWebviewWindowWithOptions(application.WebviewWindowOptions{
+	window := app.NewWebviewWindowWithOptions(application.WebviewWindowOptions{
 		Title: "Wave Terminal",
 		Mac: application.MacWindow{
 			InvisibleTitleBarHeight: 50,
@@ -67,6 +64,10 @@ func main() {
 		BackgroundColour: application.NewRGB(27, 38, 54),
 		URL:              "/public/index.html",
 	})
+	eventbus.RegisterWailsWindow(window)
+
+	eventbus.Start()
+	defer eventbus.Shutdown()
 
 	// blocking
 	err = app.Run()

pkg/blockcontroller/blockcontroller.go

@@ -0,0 +1,95 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package blockcontroller
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/wailsapp/wails/v3/pkg/application"
+	"github.com/wavetermdev/thenextwave/pkg/eventbus"
+)
+
+var globalLock = &sync.Mutex{}
+var blockControllerMap = make(map[string]*BlockController)
+
+type BlockCommand interface {
+	GetType() string
+}
+
+type MessageCommand struct {
+	Message string `json:"message"`
+}
+
+func (mc *MessageCommand) GetType() string {
+	return "message"
+}
+
+type BlockController struct {
+	BlockId string
+	InputCh chan BlockCommand
+}
+
+func ParseCmdMap(cmdMap map[string]any) (BlockCommand, error) {
+	cmdType, ok := cmdMap["type"].(string)
+	if !ok {
+		return nil, fmt.Errorf("no type field in command map")
+	}
+	mapJson, err := json.Marshal(cmdMap)
+	if err != nil {
+		return nil, fmt.Errorf("error marshalling command map: %w", err)
+	}
+	switch cmdType {
+	case "message":
+		var cmd MessageCommand
+		err := json.Unmarshal(mapJson, &cmd)
+		if err != nil {
+			return nil, fmt.Errorf("error unmarshalling message command: %w", err)
+		}
+		return &cmd, nil
+	default:
+		return nil, fmt.Errorf("unknown command type %q", cmdType)
+	}
+}
+
+func (bc *BlockController) Run() {
+	defer func() {
+		eventbus.SendEvent(application.WailsEvent{
+			Name: "block:done",
+			Data: nil,
+		})
+		globalLock.Lock()
+		defer globalLock.Unlock()
+		delete(blockControllerMap, bc.BlockId)
+	}()
+
+	for genCmd := range bc.InputCh {
+		switch cmd := genCmd.(type) {
+		case *MessageCommand:
+			fmt.Printf("MESSAGE: %s | %q\n", bc.BlockId, cmd.Message)
+
+		default:
+			fmt.Printf("unknown command type %T\n", cmd)
+		}
+	}
+}
+
+func NewBlockController(blockId string) *BlockController {
+	globalLock.Lock()
+	defer globalLock.Unlock()
+	bc := &BlockController{
+		BlockId: blockId,
+		InputCh: make(chan BlockCommand),
+	}
+	blockControllerMap[blockId] = bc
+	go bc.Run()
+	return bc
+}
+
+func GetBlockController(blockId string) *BlockController {
+	globalLock.Lock()
+	defer globalLock.Unlock()
+	return blockControllerMap[blockId]
+}

pkg/eventbus/eventbus.go

@@ -0,0 +1,122 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package eventbus
+
+import (
+	"errors"
+	"log"
+	"runtime/debug"
+	"sync"
+
+	"github.com/wailsapp/wails/v3/pkg/application"
+)
+
+const EventBufferSize = 50
+
+var EventCh chan application.WailsEvent = make(chan application.WailsEvent, EventBufferSize)
+var WindowEventCh chan WindowEvent = make(chan WindowEvent, EventBufferSize)
+var shutdownCh chan struct{} = make(chan struct{})
+var ErrQueueFull = errors.New("event queue full")
+
+type WindowEvent struct {
+	WindowId uint
+	Event    application.WailsEvent
+}
+
+var globalLock = &sync.Mutex{}
+var wailsApp *application.App
+var wailsWindowMap = make(map[uint]*application.WebviewWindow)
+
+func Start() {
+	go processEvents()
+}
+
+func Shutdown() {
+	close(shutdownCh)
+}
+
+func RegisterWailsApp(app *application.App) {
+	globalLock.Lock()
+	defer globalLock.Unlock()
+	wailsApp = app
+}
+
+func RegisterWailsWindow(window *application.WebviewWindow) {
+	globalLock.Lock()
+	defer globalLock.Unlock()
+	wailsWindowMap[window.ID()] = window
+}
+
+func UnregisterWailsWindow(windowId uint) {
+	globalLock.Lock()
+	defer globalLock.Unlock()
+	delete(wailsWindowMap, windowId)
+}
+
+func emitEventToWindow(event WindowEvent) {
+	globalLock.Lock()
+	window := wailsWindowMap[event.WindowId]
+	globalLock.Unlock()
+	if window != nil {
+		window.DispatchWailsEvent(&event.Event)
+	}
+}
+
+func SendEvent(event application.WailsEvent) {
+	EventCh <- event
+}
+
+func SendEventNonBlocking(event application.WailsEvent) error {
+	select {
+	case EventCh <- event:
+		return nil
+	default:
+		return ErrQueueFull
+	}
+}
+
+func SendWindowEvent(windowId uint, event application.WailsEvent) {
+	WindowEventCh <- WindowEvent{
+		WindowId: windowId,
+		Event:    event,
+	}
+}
+
+func SendWindowEventNonBlocking(windowId uint, event application.WailsEvent) error {
+	select {
+	case WindowEventCh <- WindowEvent{
+		WindowId: windowId,
+		Event:    event,
+	}:
+		return nil
+	default:
+		return ErrQueueFull
+	}
+}
+
+func processEvents() {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("eventbus panic: %v\n", r)
+			debug.PrintStack()
+		}
+	}()
+
+	log.Printf("eventbus starting\n")
+	for {
+		select {
+		case event := <-EventCh:
+			// no lock needed for wailsApp since it is never updated
+			if wailsApp != nil {
+				wailsApp.Events.Emit(&event)
+			}
+		case windowEvent := <-WindowEventCh:
+			emitEventToWindow(windowEvent)
+
+		case <-shutdownCh:
+			log.Printf("eventbus shutting down\n")
+			return
+		}
+	}
+}

pkg/service/blockservice/blockservice.go

@@ -0,0 +1,25 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package blockservice
+
+import (
+	"fmt"
+
+	"github.com/wavetermdev/thenextwave/pkg/blockcontroller"
+)
+
+type BlockService struct{}
+
+func (bs *BlockService) SendCommand(blockId string, cmdMap map[string]any) error {
+	bc := blockcontroller.GetBlockController(blockId)
+	if bc == nil {
+		return fmt.Errorf("block controller not found for block %q", blockId)
+	}
+	cmd, err := blockcontroller.ParseCmdMap(cmdMap)
+	if err != nil {
+		return fmt.Errorf("error parsing command map: %w", err)
+	}
+	bc.InputCh <- cmd
+	return nil
+}

pkg/wavebase/wavebase.go

@@ -1,3 +1,6 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
 package wavebase
 
 import (