Add example HTTP server

buttplug.go

@@ -39,12 +39,76 @@ const Version = 2
 // NewRequestServerInfo creates a new RequestServerInfo with the current client
 // information.
 func NewRequestServerInfo() *RequestServerInfo {
+	v := new(int)
+	*v = Version
 	return &RequestServerInfo{
 		ClientName:     "go-buttplug",
-		MessageVersion: Version,
+		MessageVersion: v,
 	}
 }
 
+// Broadcaster is used for creating multiple event loops on the same Buttplug
+// server.
+type Broadcaster struct {
+	dst  map[chan<- Message]struct{}
+	mut  sync.Mutex
+	void bool
+}
+
+// NewBroadcaster creates a new broadcaster.
+func NewBroadcaster() *Broadcaster {
+	return &Broadcaster{
+		dst: make(map[chan<- Message]struct{}),
+	}
+}
+
+// Start starts the broadcaster.
+func (b *Broadcaster) Start(src <-chan Message) {
+	b.mut.Lock()
+	if b.void {
+		panic("Start called on voided Broadcaster")
+	}
+	b.mut.Unlock()
+
+	go func() {
+		for op := range src {
+			b.mut.Lock()
+
+			for ch := range b.dst {
+				ch <- op
+			}
+
+			b.mut.Unlock()
+		}
+
+		b.mut.Lock()
+		b.void = true
+
+		for ch := range b.dst {
+			close(ch)
+		}
+
+		b.mut.Unlock()
+	}()
+}
+
+// Subscribe subscribes the given channel
+func (b *Broadcaster) Subscribe(ch chan<- Message) {
+	b.mut.Lock()
+	if b.void {
+		panic("Subscribe called on voided Broadcaster")
+	}
+	b.dst[ch] = struct{}{}
+	b.mut.Unlock()
+}
+
+// Listen returns a newly subscribed Op channel.
+func (b *Broadcaster) Listen() <-chan Message {
+	ch := make(chan Message, 1)
+	b.Subscribe(ch)
+	return ch
+}
+
 type command struct {
 	msg   Message
 	reply chan Message
@@ -422,8 +486,8 @@ func (w *Websocket) Send(ctx context.Context, msgs ...Message) {
 // Command sends a message over the websocket and waits for a reply. If the
 // caller calls this method after the websocket is closed, the function will
 // block forever, since a websocket cannot be started back up. The returned
-// message is never nil, but it may be of type InternalError, which the function
-// will unbox into the return type.
+// message is never nil, but it may be of type InternalError or Error, which the
+// function will unbox into the return error type.
 func (w *Websocket) Command(ctx context.Context, msg Message) (Message, error) {
 	msg.SetMessageID(w.id.Next())
 	cmd := command{
@@ -444,10 +508,14 @@ func (w *Websocket) Command(ctx context.Context, msg Message) (Message, error) {
 		err := errors.Wrap(ctx.Err(), "timed out waiting")
 		return &InternalError{Err: err}, err
 	case reply := <-cmd.reply:
-		if err, ok := reply.(*InternalError); ok {
-			return reply, err.Err
+		switch reply := reply.(type) {
+		case *InternalError:
+			return reply, reply.Err
+		case *Error:
+			return reply, reply
+		default:
+			return reply, nil
 		}
-		return reply, nil
 	}
 }
 

cmd/buttplughttp/README.md

@@ -0,0 +1,4 @@
+# buttplughttp
+
+An example application using the buttplug API that exposes a REST API to
+control devices.

cmd/buttplughttp/go.mod

@@ -0,0 +1,14 @@
+module github.com/diamondburned/go-buttplug/cmd/buttplughttp
+
+go 1.17
+
+replace github.com/diamondburned/go-buttplug => ../../
+
+require github.com/diamondburned/go-buttplug v0.0.0-00010101000000-000000000000
+
+require (
+	github.com/go-chi/chi v1.5.4 // indirect
+	github.com/gorilla/schema v1.2.0 // indirect
+	github.com/gorilla/websocket v1.4.2 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+)

cmd/buttplughttp/go.sum

@@ -0,0 +1,15 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-chi/chi v1.5.4 h1:QHdzF2szwjqVV4wmByUnTcsbIg7UGaQ0tPF2t5GcAIs=
+github.com/go-chi/chi v1.5.4/go.mod h1:uaf8YgoFazUOkPBG7fxPftUylNumIev9awIWOENIuEg=
+github.com/gorilla/schema v1.2.0 h1:YufUaxZYCKGFuAq3c96BOhjgd5nmXiOY9NGzF247Tsc=
+github.com/gorilla/schema v1.2.0/go.mod h1:kgLaKoK1FELgZqMAVxx/5cbj0kT+57qxUrAlIO2eleU=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/qri-io/jsonpointer v0.1.1/go.mod h1:DnJPaYgiKu56EuDp8TU5wFLdZIcAnb/uH9v37ZaMV64=
+github.com/qri-io/jsonschema v0.2.1/go.mod h1:g7DPkiOsK1xv6T/Ao5scXRkd+yTFygcANPBaaqW+VrI=
+github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=

cmd/buttplughttp/main.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+
+	"github.com/diamondburned/go-buttplug"
+	"github.com/diamondburned/go-buttplug/device"
+	"github.com/diamondburned/go-buttplug/intiface"
+	"github.com/pkg/errors"
+)
+
+var (
+	wsPort      = 20000
+	httpAddr    = "localhost:8080"
+	intifaceCLI = "intiface-cli"
+)
+
+func main() {
+	flag.IntVar(&wsPort, "ws-port", wsPort, "websocket port to start from")
+	flag.StringVar(&httpAddr, "http-addr", httpAddr, "http address to listen to")
+	flag.StringVar(&intifaceCLI, "exec", intifaceCLI, "command to execute if no --ws-addr")
+	flag.Parse()
+
+	if err := run(); err != nil {
+		log.Fatalln(err)
+	}
+}
+
+func run() error {
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer cancel()
+
+	ws := intiface.NewWebsocket(wsPort, intifaceCLI)
+	broadcaster := buttplug.NewBroadcaster()
+
+	manager := device.NewManager()
+	manager.Listen(broadcaster.Listen())
+
+	msgCh := broadcaster.Listen()
+
+	// Start connecting and broadcasting messages at the same time.
+	broadcaster.Start(ws.Open(ctx))
+
+	httpErr := make(chan error)
+	go func() {
+		server := newServer(ws.Websocket, manager)
+		httpErr <- serveHTTP(ctx, httpAddr, server)
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case msg := <-msgCh:
+			switch msg := msg.(type) {
+			case *buttplug.ServerInfo:
+				// Server is ready. Start scanning and ask for the list of
+				// devices. The device manager will pick up the device messages
+				// for us.
+				ws.Send(ctx,
+					&buttplug.StartScanning{},
+					&buttplug.RequestDeviceList{},
+				)
+			case *buttplug.DeviceList:
+				for _, device := range msg.Devices {
+					log.Printf("listed device %s (index %d)", device.DeviceName, device.DeviceIndex)
+				}
+			case *buttplug.DeviceAdded:
+				log.Printf("added device %s (index %d)", msg.DeviceName, msg.DeviceIndex)
+			case *buttplug.DeviceRemoved:
+				log.Println("removed device", msg.DeviceIndex)
+			case error:
+				log.Println("buttplug error:", msg)
+			}
+		case err := <-httpErr:
+			return errors.Wrap(err, "HTTP error")
+		}
+	}
+}
+
+func serveHTTP(ctx context.Context, addr string, h http.Handler) error {
+	server := http.Server{
+		Addr:    addr,
+		Handler: h,
+	}
+
+	go func() {
+		<-ctx.Done()
+		server.Shutdown(ctx)
+	}()
+
+	log.Print("starting HTTP server at http://", addr)
+
+	if err := server.ListenAndServe(); err != nil {
+		if errors.Is(err, http.ErrServerClosed) {
+			return nil
+		}
+		return err
+	}
+	return nil
+}