change internals to internal

internal/framing/framer.go

@@ -0,0 +1,7 @@
+package framing
+
+type Framer interface {
+	Decode([]byte) [][]byte
+	Encode([]byte) []byte
+	Clear()
+}

internal/framing/separator.go

@@ -0,0 +1,37 @@
+package framing
+
+import (
+	"bytes"
+)
+
+type ByteSeparatorFramer struct {
+	buffer    []byte
+	separator []byte
+}
+
+func NewByteSeparatorFramer(separator []byte) *ByteSeparatorFramer {
+	return &ByteSeparatorFramer{separator: separator, buffer: []byte{}}
+}
+
+func (bsf *ByteSeparatorFramer) Decode(data []byte) [][]byte {
+	messages := [][]byte{}
+
+	bsf.buffer = append(bsf.buffer, data...)
+
+	parts := bytes.Split(bsf.buffer, bsf.separator)
+
+	if len(parts) > 0 {
+		bsf.buffer = parts[len(parts)-1]
+		messages = parts[:len(parts)-1]
+	}
+
+	return messages
+}
+
+func (bsf *ByteSeparatorFramer) Encode(data []byte) []byte {
+	return append(data, bsf.separator...)
+}
+
+func (bsf *ByteSeparatorFramer) Clear() {
+	bsf.buffer = []byte{}
+}

internal/framing/slip.go

@@ -0,0 +1,76 @@
+package framing
+
+type SlipFramer struct {
+	buffer []byte
+}
+
+func NewSlipFramer() *SlipFramer {
+	return &SlipFramer{buffer: []byte{}}
+}
+
+func (sf *SlipFramer) Decode(data []byte) [][]byte {
+	messages := [][]byte{}
+
+	END := byte(0xc0)
+	ESC := byte(0xdb)
+	ESC_END := byte(0xdc)
+	ESC_ESC := byte(0xdd)
+
+	escapeNext := false
+	for _, packetByte := range data {
+
+		if packetByte == ESC {
+			escapeNext = true
+			continue
+		}
+
+		if escapeNext {
+			if packetByte == ESC_END {
+				sf.buffer = append(sf.buffer, END)
+			} else if packetByte == ESC_ESC {
+				sf.buffer = append(sf.buffer, ESC)
+			}
+			escapeNext = false
+		} else if packetByte == END {
+			if len(sf.buffer) == 0 {
+				// opening END byte, can discard
+				continue
+			} else {
+				message := sf.buffer
+				messages = append(messages, message)
+			}
+			sf.buffer = []byte{}
+		} else {
+			sf.buffer = append(sf.buffer, packetByte)
+		}
+	}
+
+	return messages
+}
+
+func (sf *SlipFramer) Encode(data []byte) []byte {
+	END := byte(0xc0)
+	ESC := byte(0xdb)
+	ESC_END := byte(0xdc)
+	ESC_ESC := byte(0xdd)
+
+	var encodedBytes = []byte{END}
+
+	for _, byteToEncode := range data {
+		switch byteToEncode {
+		case END:
+			encodedBytes = append(encodedBytes, ESC_END)
+		case ESC:
+			encodedBytes = append(encodedBytes, ESC_ESC)
+		default:
+			encodedBytes = append(encodedBytes, byteToEncode)
+		}
+	}
+
+	encodedBytes = append(encodedBytes, END)
+	return encodedBytes
+}
+
+func (sf *SlipFramer) Clear() {
+	sf.buffer = []byte{}
+}

internal/processing/osc-message-create.go

@@ -0,0 +1,73 @@
+package processing
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"text/template"
+
+	"github.com/jwetzell/osc-go"
+)
+
+type OSCMessageCreate struct {
+	config  ProcessorConfig
+	Address *template.Template
+}
+
+func (o *OSCMessageCreate) Process(ctx context.Context, payload any) (any, error) {
+
+	var addressBuffer bytes.Buffer
+	err := o.Address.Execute(&addressBuffer, payload)
+
+	if err != nil {
+		return nil, err
+	}
+
+	addressString := addressBuffer.String()
+
+	if len(addressString) == 0 {
+		return nil, fmt.Errorf("osc.message.create address must not be empty")
+	}
+
+	if addressString[0] != '/' {
+		return nil, fmt.Errorf("osc.message.create address must start with '/'")
+	}
+
+	payloadMessage := osc.OSCMessage{
+		Address: addressString,
+	}
+
+	return payloadMessage, nil
+}
+
+func (o *OSCMessageCreate) Type() string {
+	return o.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "osc.message.create",
+		New: func(config ProcessorConfig) (Processor, error) {
+			params := config.Params
+			address, ok := params["address"]
+
+			if !ok {
+				return nil, fmt.Errorf("osc.message.create requires an address parameter")
+			}
+
+			addressString, ok := address.(string)
+
+			if !ok {
+				return nil, fmt.Errorf("osc.message.create address must be a string")
+			}
+
+			addressTemplate, err := template.New("address").Parse(addressString)
+
+			if err != nil {
+				return nil, err
+			}
+
+			return &OSCMessageCreate{config: config, Address: addressTemplate}, nil
+		},
+	})
+}

internal/processing/osc-message-decode.go

@@ -0,0 +1,47 @@
+package processing
+
+import (
+	"context"
+	"fmt"
+
+	osc "github.com/jwetzell/osc-go"
+)
+
+type OSCMessageDecode struct {
+	config ProcessorConfig
+}
+
+func (o *OSCMessageDecode) Process(ctx context.Context, payload any) (any, error) {
+	payloadBytes, ok := payload.([]byte)
+
+	if !ok {
+		return nil, fmt.Errorf("osc.message.decode processor only accepts a []byte payload")
+	}
+
+	if len(payloadBytes) == 0 {
+		return nil, fmt.Errorf("osc.message.decode processor can't work on empty []byte")
+	}
+
+	if payloadBytes[0] != '/' {
+		return nil, fmt.Errorf("osc.message.decode processor needs an OSC looking []byte")
+	}
+
+	message, err := osc.MessageFromBytes(payloadBytes)
+	if err != nil {
+		return nil, err
+	}
+	return message, nil
+}
+
+func (o *OSCMessageDecode) Type() string {
+	return o.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "osc.message.decode",
+		New: func(config ProcessorConfig) (Processor, error) {
+			return &OSCMessageDecode{config: config}, nil
+		},
+	})
+}

internal/processing/osc-message-encode.go

@@ -0,0 +1,36 @@
+package processing
+
+import (
+	"context"
+	"fmt"
+
+	osc "github.com/jwetzell/osc-go"
+)
+
+type OSCMessageEncode struct {
+	config ProcessorConfig
+}
+
+func (o *OSCMessageEncode) Process(ctx context.Context, payload any) (any, error) {
+	payloadMessage, ok := payload.(osc.OSCMessage)
+
+	if !ok {
+		return nil, fmt.Errorf("osc.message.encode processor only accepts an OSCMessage")
+	}
+
+	bytes := payloadMessage.ToBytes()
+	return bytes, nil
+}
+
+func (o *OSCMessageEncode) Type() string {
+	return o.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "osc.message.encode",
+		New: func(config ProcessorConfig) (Processor, error) {
+			return &OSCMessageEncode{config: config}, nil
+		},
+	})
+}

internal/processing/osc-message-transform.go

@@ -0,0 +1,77 @@
+package processing
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"text/template"
+
+	osc "github.com/jwetzell/osc-go"
+)
+
+type OSCMessageTransform struct {
+	config  ProcessorConfig
+	Address *template.Template
+}
+
+func (o *OSCMessageTransform) Process(ctx context.Context, payload any) (any, error) {
+	payloadMessage, ok := payload.(osc.OSCMessage)
+
+	if !ok {
+		return nil, fmt.Errorf("osc.message.transform processor only accepts an OSCMessage")
+	}
+
+	var addressBuffer bytes.Buffer
+	//TODO(jwetzell): actually inject data into template
+	err := o.Address.Execute(&addressBuffer, payloadMessage)
+
+	if err != nil {
+		return nil, err
+	}
+
+	addressString := addressBuffer.String()
+
+	if len(addressString) == 0 {
+		return nil, fmt.Errorf("osc.message.transform address must not be empty")
+	}
+
+	if addressString[0] != '/' {
+		return nil, fmt.Errorf("osc.message.transform address must start with '/'")
+	}
+
+	payloadMessage.Address = addressString
+
+	return payloadMessage, nil
+}
+
+func (o *OSCMessageTransform) Type() string {
+	return o.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "osc.message.transform",
+		New: func(config ProcessorConfig) (Processor, error) {
+			params := config.Params
+			address, ok := params["address"]
+
+			if !ok {
+				return nil, fmt.Errorf("osc.message.transform requires an address parameter")
+			}
+
+			addressString, ok := address.(string)
+
+			if !ok {
+				return nil, fmt.Errorf("osc.message.transform address must be a string")
+			}
+
+			addressTemplate, err := template.New("address").Parse(addressString)
+
+			if err != nil {
+				return nil, err
+			}
+
+			return &OSCMessageTransform{config: config, Address: addressTemplate}, nil
+		},
+	})
+}

internal/processing/processor.go

@@ -0,0 +1,45 @@
+package processing
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+type Processor interface {
+	Type() string
+	Process(context.Context, any) (any, error)
+}
+
+type ProcessorConfig struct {
+	Type   string         `json:"type"`
+	Params map[string]any `json:"params"`
+}
+
+type ProcessorRegistration struct {
+	Type string `json:"type"`
+	New  func(ProcessorConfig) (Processor, error)
+}
+
+func RegisterProcessor(processor ProcessorRegistration) {
+
+	if processor.Type == "" {
+		panic("processor type is missing")
+	}
+	if processor.New == nil {
+		panic("missing ProcessorRegistration.New")
+	}
+
+	processorRegistryMu.Lock()
+	defer processorRegistryMu.Unlock()
+
+	if _, ok := ProcessorRegistry[string(processor.Type)]; ok {
+		panic(fmt.Sprintf("processor already registered: %s", processor.Type))
+	}
+	ProcessorRegistry[string(processor.Type)] = processor
+}
+
+var (
+	processorRegistryMu sync.RWMutex
+	ProcessorRegistry   = make(map[string]ProcessorRegistration)
+)