jwetzell/showbridge-go
1
0
Fork 0
mirror of https://github.com/jwetzell/showbridge-go.git synced 2026-09-10 07:29:25 +00:00
Code Issues Packages Projects Releases 10 Wiki Activity

switch modules to accept an InputHandler instead of the full router

Browse Source
This commit is contained in:
Joel Wetzell
2026-05-20 21:03:41 -05:00
parent be0b1c4a5f
commit 4cedd58a76
31 changed files with 301 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/common/module.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
type Module interface { type Module interface {
Id() string Id() string
Type() string Type() string
Start(context.Context, RouteIO) error Start(context.Context, InputHandler) error
Stop() Stop()
} }
internal/common/payload.go
+5 -5
View File
@@ -1,9 +1,9 @@
package common package common
type WrappedPayload struct { type WrappedPayload struct {
Payload any Payload any
Router RouteIO InputHandler InputHandler
Modules map[string]Module Modules map[string]Module
Source string Source string
End bool End bool
} }
internal/common/routing.go
+1 -3
View File
@@ -4,9 +4,7 @@ import (
"context" "context"
) )
type RouteIO interface { type InputHandler func(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)
HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)
}
type RouteIOError struct { type RouteIOError struct {
Index int `json:"index"` Index int `json:"index"`
internal/module/db-sqlite.go
+10 -10
View File
@@ -15,14 +15,14 @@ import (
) )
type DbSqlite struct { type DbSqlite struct {
config config.ModuleConfig config config.ModuleConfig
Dsn string Dsn string
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
db *sql.DB db *sql.DB
logger *slog.Logger logger *slog.Logger
dbMu sync.Mutex dbMu sync.Mutex
cancel context.CancelFunc cancel context.CancelFunc
} }
func init() { func init() {
@@ -61,9 +61,9 @@ func (dbs *DbSqlite) Type() string {
return dbs.config.Type return dbs.config.Type
} }
func (dbs *DbSqlite) Start(ctx context.Context, router common.RouteIO) error { func (dbs *DbSqlite) Start(ctx context.Context, inputHandler common.InputHandler) error {
dbs.logger.Debug("running") dbs.logger.Debug("running")
dbs.router = router dbs.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
dbs.ctx = moduleContext dbs.ctx = moduleContext
dbs.cancel = cancel dbs.cancel = cancel
internal/module/http-server.go
+12 -12
View File
@@ -17,14 +17,14 @@ import (
) )
type HTTPServer struct { type HTTPServer struct {
config config.ModuleConfig config config.ModuleConfig
Port uint16 Port uint16
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
server *http.Server server *http.Server
serverMu sync.Mutex serverMu sync.Mutex
} }
type ResponseIOError struct { type ResponseIOError struct {
@@ -98,9 +98,9 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Message: "routing successful", Message: "routing successful",
Status: "ok", Status: "ok",
} }
if hs.router != nil { if hs.inputHandler != nil {
inputContext := context.WithValue(hs.ctx, httpServerContextKey("responseWriter"), &responseWriter) inputContext := context.WithValue(hs.ctx, httpServerContextKey("responseWriter"), &responseWriter)
aRouteFound, routingErrors := hs.router.HandleInput(inputContext, hs.Id(), r) aRouteFound, routingErrors := hs.inputHandler(inputContext, hs.Id(), r)
if !responseWriter.done { if !responseWriter.done {
if aRouteFound { if aRouteFound {
if routingErrors != nil { if routingErrors != nil {
@@ -147,9 +147,9 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
} }
func (hs *HTTPServer) Start(ctx context.Context, router common.RouteIO) error { func (hs *HTTPServer) Start(ctx context.Context, inputHandler common.InputHandler) error {
hs.logger.Debug("running") hs.logger.Debug("running")
hs.router = router hs.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
hs.ctx = moduleContext hs.ctx = moduleContext
hs.cancel = cancel hs.cancel = cancel
internal/module/midi-input.go
+11 -11
View File
@@ -15,13 +15,13 @@ import (
) )
type MIDIInput struct { type MIDIInput struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Port string Port string
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
stop func() stop func()
} }
func init() { func init() {
@@ -59,9 +59,9 @@ func (mi *MIDIInput) Type() string {
return mi.config.Type return mi.config.Type
} }
func (mi *MIDIInput) Start(ctx context.Context, router common.RouteIO) error { func (mi *MIDIInput) Start(ctx context.Context, inputHandler common.InputHandler) error {
mi.logger.Debug("running") mi.logger.Debug("running")
mi.router = router mi.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
mi.ctx = moduleContext mi.ctx = moduleContext
mi.cancel = cancel mi.cancel = cancel
@@ -72,8 +72,8 @@ func (mi *MIDIInput) Start(ctx context.Context, router common.RouteIO) error {
} }
stop, err := midi.ListenTo(in, func(msg midi.Message, timestampms int32) { stop, err := midi.ListenTo(in, func(msg midi.Message, timestampms int32) {
if mi.router != nil { if mi.inputHandler != nil {
mi.router.HandleInput(mi.ctx, mi.Id(), msg) mi.inputHandler(mi.ctx, mi.Id(), msg)
} }
}, midi.UseSysEx()) }, midi.UseSysEx())
internal/module/midi-output.go
+10 -10
View File
@@ -17,14 +17,14 @@ import (
) )
type MIDIOutput struct { type MIDIOutput struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Port string Port string
sendFunc func(midi.Message) error sendFunc func(midi.Message) error
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
sendFuncMu sync.Mutex sendFuncMu sync.Mutex
} }
func init() { func init() {
@@ -63,9 +63,9 @@ func (mo *MIDIOutput) Type() string {
return mo.config.Type return mo.config.Type
} }
func (mo *MIDIOutput) Start(ctx context.Context, router common.RouteIO) error { func (mo *MIDIOutput) Start(ctx context.Context, inputHandler common.InputHandler) error {
mo.logger.Debug("running") mo.logger.Debug("running")
mo.router = router mo.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
mo.ctx = moduleContext mo.ctx = moduleContext
mo.cancel = cancel mo.cancel = cancel
internal/module/mqtt-client.go
+17 -15
View File
@@ -15,18 +15,18 @@ import (
) )
type MQTTClient struct { type MQTTClient struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Broker string Broker string
ClientID string ClientID string
Topic string Topic string
QoS byte QoS byte
Retained bool Retained bool
client mqtt.Client client mqtt.Client
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
clientMu sync.Mutex clientMu sync.Mutex
} }
func init() { func init() {
@@ -117,9 +117,9 @@ func (mc *MQTTClient) Type() string {
return mc.config.Type return mc.config.Type
} }
func (mc *MQTTClient) Start(ctx context.Context, router common.RouteIO) error { func (mc *MQTTClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
mc.logger.Debug("running") mc.logger.Debug("running")
mc.router = router mc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
mc.ctx = moduleContext mc.ctx = moduleContext
mc.cancel = cancel mc.cancel = cancel
@@ -132,7 +132,9 @@ func (mc *MQTTClient) Start(ctx context.Context, router common.RouteIO) error {
opts.OnConnect = func(c mqtt.Client) { opts.OnConnect = func(c mqtt.Client) {
token := mc.client.Subscribe(mc.Topic, 1, func(c mqtt.Client, m mqtt.Message) { token := mc.client.Subscribe(mc.Topic, 1, func(c mqtt.Client, m mqtt.Message) {
mc.router.HandleInput(mc.ctx, mc.Id(), m) if mc.inputHandler != nil {
mc.inputHandler(mc.ctx, mc.Id(), m)
}
}) })
token.Wait() token.Wait()
} }
internal/module/nats-client.go
+15 -15
View File
@@ -13,17 +13,17 @@ import (
) )
type NATSClient struct { type NATSClient struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
URL string URL string
Subject string Subject string
client *nats.Conn client *nats.Conn
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
sub *nats.Subscription sub *nats.Subscription
subMu sync.Mutex subMu sync.Mutex
clientMu sync.Mutex clientMu sync.Mutex
} }
func init() { func init() {
@@ -71,9 +71,9 @@ func (nc *NATSClient) Type() string {
return nc.config.Type return nc.config.Type
} }
func (nc *NATSClient) Start(ctx context.Context, router common.RouteIO) error { func (nc *NATSClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
nc.logger.Debug("running") nc.logger.Debug("running")
nc.router = router nc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
nc.ctx = moduleContext nc.ctx = moduleContext
nc.cancel = cancel nc.cancel = cancel
@@ -89,8 +89,8 @@ func (nc *NATSClient) Start(ctx context.Context, router common.RouteIO) error {
nc.clientMu.Unlock() nc.clientMu.Unlock()
sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) { sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) {
if nc.router != nil { if nc.inputHandler != nil {
nc.router.HandleInput(nc.ctx, nc.Id(), msg) nc.inputHandler(nc.ctx, nc.Id(), msg)
} }
}) })
internal/module/nats-server.go
+11 -11
View File
@@ -17,15 +17,15 @@ import (
) )
type NATSServer struct { type NATSServer struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
Ip string Ip string
Port int Port int
router common.RouteIO inputHandler common.InputHandler
server *server.Server server *server.Server
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
serverMu sync.Mutex serverMu sync.Mutex
} }
func init() { func init() {
@@ -88,9 +88,9 @@ func (ns *NATSServer) Type() string {
return ns.config.Type return ns.config.Type
} }
func (ns *NATSServer) Start(ctx context.Context, router common.RouteIO) error { func (ns *NATSServer) Start(ctx context.Context, inputHandler common.InputHandler) error {
ns.logger.Debug("running") ns.logger.Debug("running")
ns.router = router ns.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
ns.ctx = moduleContext ns.ctx = moduleContext
ns.cancel = cancel ns.cancel = cancel
internal/module/psn-client.go
+13 -13
View File
@@ -13,14 +13,14 @@ import (
) )
type PSNClient struct { type PSNClient struct {
config config.ModuleConfig config config.ModuleConfig
conn *net.UDPConn conn *net.UDPConn
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
decoder *psn.Decoder decoder *psn.Decoder
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
connMu sync.Mutex connMu sync.Mutex
} }
func init() { func init() {
@@ -41,9 +41,9 @@ func (pc *PSNClient) Type() string {
return pc.config.Type return pc.config.Type
} }
func (pc *PSNClient) Start(ctx context.Context, router common.RouteIO) error { func (pc *PSNClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
pc.logger.Debug("running") pc.logger.Debug("running")
pc.router = router pc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
pc.ctx = moduleContext pc.ctx = moduleContext
pc.cancel = cancel pc.cancel = cancel
@@ -88,13 +88,13 @@ func (pc *PSNClient) Start(ctx context.Context, router common.RouteIO) error {
pc.logger.Error("problem decoding psn traffic", "error", err) pc.logger.Error("problem decoding psn traffic", "error", err)
} }
if pc.router != nil { if pc.inputHandler != nil {
// TODO(jwetzell): better input handling // TODO(jwetzell): better input handling
for _, tracker := range pc.decoder.Trackers { for _, tracker := range pc.decoder.Trackers {
pc.router.HandleInput(pc.ctx, pc.Id(), tracker) pc.inputHandler(pc.ctx, pc.Id(), tracker)
} }
} else { } else {
pc.logger.Error("has no router") pc.logger.Error("has no input handler")
} }
} }
} }
internal/module/redis-client.go
+11 -11
View File
@@ -14,15 +14,15 @@ import (
) )
type RedisClient struct { type RedisClient struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Host string Host string
Port uint16 Port uint16
client *redis.Client client *redis.Client
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
clientMu sync.Mutex clientMu sync.Mutex
} }
func init() { func init() {
@@ -75,10 +75,10 @@ func (rc *RedisClient) Printf(ctx context.Context, format string, v ...any) {
rc.logger.Debug(msg) rc.logger.Debug(msg)
} }
func (rc *RedisClient) Start(ctx context.Context, router common.RouteIO) error { func (rc *RedisClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
redis.SetLogger(rc) redis.SetLogger(rc)
rc.logger.Debug("running") rc.logger.Debug("running")
rc.router = router rc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
rc.ctx = moduleContext rc.ctx = moduleContext
rc.cancel = cancel rc.cancel = cancel
internal/module/serial-client.go
+14 -14
View File
@@ -18,16 +18,16 @@ import (
) )
type SerialClient struct { type SerialClient struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Port string Port string
Framer framer.Framer Framer framer.Framer
Mode *serial.Mode Mode *serial.Mode
port serial.Port port serial.Port
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
portMu sync.Mutex portMu sync.Mutex
} }
func init() { func init() {
@@ -107,9 +107,9 @@ func (sc *SerialClient) SetupPort() error {
return nil return nil
} }
func (sc *SerialClient) Start(ctx context.Context, router common.RouteIO) error { func (sc *SerialClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
sc.logger.Debug("running") sc.logger.Debug("running")
sc.router = router sc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
sc.ctx = moduleContext sc.ctx = moduleContext
sc.cancel = cancel sc.cancel = cancel
@@ -147,8 +147,8 @@ func (sc *SerialClient) Start(ctx context.Context, router common.RouteIO) error
if byteCount > 0 { if byteCount > 0 {
messages := sc.Framer.Decode(buffer[0:byteCount]) messages := sc.Framer.Decode(buffer[0:byteCount])
for _, message := range messages { for _, message := range messages {
if sc.router != nil { if sc.inputHandler != nil {
sc.router.HandleInput(sc.ctx, sc.Id(), message) sc.inputHandler(sc.ctx, sc.Id(), message)
} else { } else {
sc.logger.Error("input received but no router is configured") sc.logger.Error("input received but no router is configured")
} }
internal/module/sip-call-server.go
+18 -16
View File
@@ -22,17 +22,17 @@ import (
) )
type SIPCallServer struct { type SIPCallServer struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
IP string IP string
Port int Port int
Transport string Transport string
UserAgent string UserAgent string
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
ua *sipgo.UserAgent ua *sipgo.UserAgent
uaMu sync.Mutex uaMu sync.Mutex
} }
type SIPCallMessage struct { type SIPCallMessage struct {
@@ -132,9 +132,9 @@ func (scs *SIPCallServer) Type() string {
return scs.config.Type return scs.config.Type
} }
func (scs *SIPCallServer) Start(ctx context.Context, router common.RouteIO) error { func (scs *SIPCallServer) Start(ctx context.Context, inputHandler common.InputHandler) error {
scs.logger.Debug("running") scs.logger.Debug("running")
scs.router = router scs.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
scs.ctx = moduleContext scs.ctx = moduleContext
scs.cancel = cancel scs.cancel = cancel
@@ -179,9 +179,11 @@ func (scs *SIPCallServer) HandleCall(inDialog *diago.DialogServerSession) {
dialogContext := context.WithValue(scs.ctx, sipCallContextKey("call"), &SIPCall{ dialogContext := context.WithValue(scs.ctx, sipCallContextKey("call"), &SIPCall{
inDialog: inDialog, inDialog: inDialog,
}) })
scs.router.HandleInput(dialogContext, scs.Id(), SIPCallMessage{ if scs.inputHandler != nil {
To: inDialog.ToUser(), scs.inputHandler(dialogContext, scs.Id(), SIPCallMessage{
}) To: inDialog.ToUser(),
})
}
} }
func (scs *SIPCallServer) Output(ctx context.Context, payload any) error { func (scs *SIPCallServer) Output(ctx context.Context, payload any) error {
internal/module/sip-dtmf-server.go
+16 -16
View File
@@ -23,18 +23,18 @@ import (
) )
type SIPDTMFServer struct { type SIPDTMFServer struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
IP string IP string
Port int Port int
Transport string Transport string
UserAgent string UserAgent string
Separator string Separator string
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
ua *sipgo.UserAgent ua *sipgo.UserAgent
uaMu sync.Mutex uaMu sync.Mutex
} }
type SIPDTMFMessage struct { type SIPDTMFMessage struct {
@@ -152,9 +152,9 @@ func (sds *SIPDTMFServer) Type() string {
return sds.config.Type return sds.config.Type
} }
func (sds *SIPDTMFServer) Start(ctx context.Context, router common.RouteIO) error { func (sds *SIPDTMFServer) Start(ctx context.Context, inputHandler common.InputHandler) error {
sds.logger.Debug("running") sds.logger.Debug("running")
sds.router = router sds.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
sds.ctx = moduleContext sds.ctx = moduleContext
sds.cancel = cancel sds.cancel = cancel
@@ -203,11 +203,11 @@ func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error
return reader.Listen(func(dtmf rune) error { return reader.Listen(func(dtmf rune) error {
if dtmf == rune(sds.Separator[0]) { if dtmf == rune(sds.Separator[0]) {
if sds.router != nil { if sds.inputHandler != nil {
dialogContext := context.WithValue(sds.ctx, sipCallContextKey("call"), &SIPDTMFCall{ dialogContext := context.WithValue(sds.ctx, sipCallContextKey("call"), &SIPDTMFCall{
inDialog: inDialog, inDialog: inDialog,
}) })
sds.router.HandleInput(dialogContext, sds.Id(), SIPDTMFMessage{ sds.inputHandler(dialogContext, sds.Id(), SIPDTMFMessage{
To: inDialog.ToUser(), To: inDialog.ToUser(),
Digits: userString, Digits: userString,
}) })
internal/module/tcp-client.go
+14 -14
View File
@@ -16,15 +16,15 @@ import (
) )
type TCPClient struct { type TCPClient struct {
config config.ModuleConfig config config.ModuleConfig
framer framer.Framer framer framer.Framer
conn *net.TCPConn conn *net.TCPConn
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Addr *net.TCPAddr Addr *net.TCPAddr
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
connMu sync.Mutex connMu sync.Mutex
} }
func init() { func init() {
@@ -93,9 +93,9 @@ func (tc *TCPClient) Type() string {
return tc.config.Type return tc.config.Type
} }
func (tc *TCPClient) Start(ctx context.Context, router common.RouteIO) error { func (tc *TCPClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
tc.logger.Debug("running") tc.logger.Debug("running")
tc.router = router tc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
tc.ctx = moduleContext tc.ctx = moduleContext
tc.cancel = cancel tc.cancel = cancel
@@ -133,10 +133,10 @@ func (tc *TCPClient) Start(ctx context.Context, router common.RouteIO) error {
if byteCount > 0 { if byteCount > 0 {
messages := tc.framer.Decode(buffer[0:byteCount]) messages := tc.framer.Decode(buffer[0:byteCount])
for _, message := range messages { for _, message := range messages {
if tc.router != nil { if tc.inputHandler != nil {
tc.router.HandleInput(tc.ctx, tc.Id(), message) tc.inputHandler(tc.ctx, tc.Id(), message)
} else { } else {
tc.logger.Error("input received but no router is configured") tc.logger.Error("input received but no input handler is configured")
} }
} }
} }
internal/module/tcp-server.go
+6 -11
View File
@@ -24,7 +24,7 @@ type TCPServer struct {
Addr *net.TCPAddr Addr *net.TCPAddr
Framer framer.Framer Framer framer.Framer
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
wg sync.WaitGroup wg sync.WaitGroup
connections []*net.TCPConn connections []*net.TCPConn
connectionsMu sync.RWMutex connectionsMu sync.RWMutex
@@ -166,15 +166,10 @@ ClientRead:
if byteCount > 0 { if byteCount > 0 {
messages := ts.Framer.Decode(buffer[0:byteCount]) messages := ts.Framer.Decode(buffer[0:byteCount])
for _, message := range messages { for _, message := range messages {
if ts.router != nil { if ts.inputHandler != nil {
_, ok := client.RemoteAddr().(*net.TCPAddr) ts.inputHandler(ts.ctx, ts.Id(), message)
if ok {
ts.router.HandleInput(ts.ctx, ts.Id(), message)
} else {
ts.router.HandleInput(ts.ctx, ts.Id(), message)
}
} else { } else {
ts.logger.Error("input received but no router is configured") ts.logger.Error("input received but no input handler is configured")
} }
} }
} }
@@ -183,9 +178,9 @@ ClientRead:
} }
} }
func (ts *TCPServer) Start(ctx context.Context, router common.RouteIO) error { func (ts *TCPServer) Start(ctx context.Context, inputHandler common.InputHandler) error {
ts.logger.Debug("running") ts.logger.Debug("running")
ts.router = router ts.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
ts.ctx = moduleContext ts.ctx = moduleContext
ts.cancel = cancel ts.cancel = cancel
internal/module/time-interval.go
+11 -11
View File
@@ -12,13 +12,13 @@ import (
) )
type TimeInterval struct { type TimeInterval struct {
config config.ModuleConfig config config.ModuleConfig
Duration uint32 Duration uint32
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
ticker *time.Ticker ticker *time.Ticker
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
} }
func init() { func init() {
@@ -57,9 +57,9 @@ func (i *TimeInterval) Type() string {
return i.config.Type return i.config.Type
} }
func (i *TimeInterval) Start(ctx context.Context, router common.RouteIO) error { func (i *TimeInterval) Start(ctx context.Context, inputHandler common.InputHandler) error {
i.logger.Debug("running") i.logger.Debug("running")
i.router = router i.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
i.ctx = moduleContext i.ctx = moduleContext
i.cancel = cancel i.cancel = cancel
@@ -72,8 +72,8 @@ func (i *TimeInterval) Start(ctx context.Context, router common.RouteIO) error {
case <-i.ctx.Done(): case <-i.ctx.Done():
return nil return nil
case <-ticker.C: case <-ticker.C:
if i.router != nil { if i.inputHandler != nil {
i.router.HandleInput(i.ctx, i.Id(), time.Now()) i.inputHandler(i.ctx, i.Id(), time.Now())
} }
} }
} }
internal/module/time-timer.go
+11 -11
View File
@@ -12,13 +12,13 @@ import (
) )
type TimeTimer struct { type TimeTimer struct {
config config.ModuleConfig config config.ModuleConfig
Duration uint32 Duration uint32
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
timer *time.Timer timer *time.Timer
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
} }
func init() { func init() {
@@ -58,9 +58,9 @@ func (t *TimeTimer) Type() string {
return t.config.Type return t.config.Type
} }
func (t *TimeTimer) Start(ctx context.Context, router common.RouteIO) error { func (t *TimeTimer) Start(ctx context.Context, inputHandler common.InputHandler) error {
t.logger.Debug("running") t.logger.Debug("running")
t.router = router t.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
t.ctx = moduleContext t.ctx = moduleContext
t.cancel = cancel t.cancel = cancel
@@ -71,8 +71,8 @@ func (t *TimeTimer) Start(ctx context.Context, router common.RouteIO) error {
case <-t.ctx.Done(): case <-t.ctx.Done():
return nil return nil
case time := <-t.timer.C: case time := <-t.timer.C:
if t.router != nil { if t.inputHandler != nil {
t.router.HandleInput(t.ctx, t.Id(), time) t.inputHandler(t.ctx, t.Id(), time)
} }
} }
} }
internal/module/udp-client.go
+11 -11
View File
@@ -14,15 +14,15 @@ import (
) )
type UDPClient struct { type UDPClient struct {
config config.ModuleConfig config config.ModuleConfig
Addr *net.UDPAddr Addr *net.UDPAddr
Port uint16 Port uint16
conn *net.UDPConn conn *net.UDPConn
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
connMu sync.Mutex connMu sync.Mutex
} }
func init() { func init() {
@@ -83,9 +83,9 @@ func (uc *UDPClient) SetupConn() error {
return err return err
} }
func (uc *UDPClient) Start(ctx context.Context, router common.RouteIO) error { func (uc *UDPClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
uc.logger.Debug("running") uc.logger.Debug("running")
uc.router = router uc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
uc.ctx = moduleContext uc.ctx = moduleContext
uc.cancel = cancel uc.cancel = cancel
internal/module/udp-multicast.go
+13 -13
View File
@@ -15,14 +15,14 @@ import (
) )
type UDPMulticast struct { type UDPMulticast struct {
config config.ModuleConfig config config.ModuleConfig
conn *net.UDPConn conn *net.UDPConn
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
Addr *net.UDPAddr Addr *net.UDPAddr
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
connMu sync.Mutex connMu sync.Mutex
} }
func init() { func init() {
@@ -75,9 +75,9 @@ func (um *UDPMulticast) Type() string {
return um.config.Type return um.config.Type
} }
func (um *UDPMulticast) Start(ctx context.Context, router common.RouteIO) error { func (um *UDPMulticast) Start(ctx context.Context, inputHandler common.InputHandler) error {
um.logger.Debug("running") um.logger.Debug("running")
um.router = router um.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
um.ctx = moduleContext um.ctx = moduleContext
um.cancel = cancel um.cancel = cancel
@@ -114,10 +114,10 @@ func (um *UDPMulticast) Start(ctx context.Context, router common.RouteIO) error
if numBytes > 0 { if numBytes > 0 {
message := buffer[:numBytes] message := buffer[:numBytes]
if um.router != nil { if um.inputHandler != nil {
um.router.HandleInput(um.ctx, um.Id(), message) um.inputHandler(um.ctx, um.Id(), message)
} else { } else {
um.logger.Error("input received but no router is configured") um.logger.Error("input received but no input handler is configured")
} }
} }
} }
internal/module/udp-server.go
+14 -14
View File
@@ -16,15 +16,15 @@ import (
) )
type UDPServer struct { type UDPServer struct {
Addr *net.UDPAddr Addr *net.UDPAddr
BufferSize int BufferSize int
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
router common.RouteIO inputHandler common.InputHandler
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
listener *net.UDPConn listener *net.UDPConn
listenerMu sync.Mutex listenerMu sync.Mutex
} }
func init() { func init() {
@@ -98,9 +98,9 @@ func (us *UDPServer) Type() string {
return us.config.Type return us.config.Type
} }
func (us *UDPServer) Start(ctx context.Context, router common.RouteIO) error { func (us *UDPServer) Start(ctx context.Context, inputHandler common.InputHandler) error {
us.logger.Debug("running") us.logger.Debug("running")
us.router = router us.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
us.ctx = moduleContext us.ctx = moduleContext
us.cancel = cancel us.cancel = cancel
@@ -129,10 +129,10 @@ func (us *UDPServer) Start(ctx context.Context, router common.RouteIO) error {
return err return err
} }
message := buffer[:numBytes] message := buffer[:numBytes]
if us.router != nil { if us.inputHandler != nil {
us.router.HandleInput(us.ctx, us.Id(), message) us.inputHandler(us.ctx, us.Id(), message)
} else { } else {
us.logger.Error("input received but no router is configured") us.logger.Error("input received but no input handler is configured")
} }
} }
} }
internal/module/websocket-client.go
+14 -14
View File
@@ -17,14 +17,14 @@ import (
) )
type WebSocketClient struct { type WebSocketClient struct {
config config.ModuleConfig config config.ModuleConfig
URL url.URL URL url.URL
ctx context.Context ctx context.Context
conn *websocket.Conn conn *websocket.Conn
router common.RouteIO inputHandler common.InputHandler
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
connMu sync.Mutex connMu sync.Mutex
} }
func init() { func init() {
@@ -79,9 +79,9 @@ func (wc *WebSocketClient) SetupConn() error {
return err return err
} }
func (wc *WebSocketClient) Start(ctx context.Context, router common.RouteIO) error { func (wc *WebSocketClient) Start(ctx context.Context, inputHandler common.InputHandler) error {
wc.logger.Debug("running") wc.logger.Debug("running")
wc.router = router wc.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
wc.ctx = moduleContext wc.ctx = moduleContext
wc.cancel = cancel wc.cancel = cancel
@@ -124,17 +124,17 @@ func (wc *WebSocketClient) readLoop() {
wc.logger.Error("websocket read error", "error", err) wc.logger.Error("websocket read error", "error", err)
return return
} }
if wc.router != nil { if wc.inputHandler != nil {
switch messageType { switch messageType {
case websocket.TextMessage: case websocket.TextMessage:
wc.router.HandleInput(wc.ctx, wc.Id(), string(message)) wc.inputHandler(wc.ctx, wc.Id(), string(message))
case websocket.BinaryMessage: case websocket.BinaryMessage:
wc.router.HandleInput(wc.ctx, wc.Id(), message) wc.inputHandler(wc.ctx, wc.Id(), message)
default: default:
wc.logger.Warn("unsupported message type received", "messageType", messageType) wc.logger.Warn("unsupported message type received", "messageType", messageType)
} }
} else { } else {
wc.logger.Error("input received but no router is configured") wc.logger.Error("input received but no input handler is configured")
continue continue
} }
} }
internal/processor/router-input.go
+3 -3
View File
@@ -20,12 +20,12 @@ type RouterInput struct {
func (ro *RouterInput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) { func (ro *RouterInput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
payload := wrappedPayload.Payload payload := wrappedPayload.Payload
if wrappedPayload.Router == nil { if wrappedPayload.InputHandler == nil {
wrappedPayload.End = true wrappedPayload.End = true
return wrappedPayload, errors.New("router.input no router found") return wrappedPayload, errors.New("router.input no input handler found")
} }
_, err := wrappedPayload.Router.HandleInput(ctx, ro.SourceId, payload) _, err := wrappedPayload.InputHandler(ctx, ro.SourceId, payload)
if err != nil { if err != nil {
wrappedPayload.End = true wrappedPayload.End = true
internal/processor/test/kv-set_test.go
+1 -1
View File
@@ -113,7 +113,7 @@ func TestBadKvSet(t *testing.T) {
name: "no module param", name: "no module param",
payload: test.TestStruct{Data: "hello"}, payload: test.TestStruct{Data: "hello"},
params: map[string]any{ params: map[string]any{
"key": "test", "key": "test",
}, },
wrappedPayloadModules: map[string]common.Module{ wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{}, "test": &test.TestKVModule{},
internal/processor/test/module-output_test.go
+3 -1
View File
@@ -34,8 +34,10 @@ func TestModuleOutputFromRegistry(t *testing.T) {
payload := "test" payload := "test"
expected := "test" expected := "test"
router := test.GetNewTestRouter()
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ got, err := processorInstance.Process(t.Context(), common.WrappedPayload{
Router: test.GetNewTestRouter(), InputHandler: router.HandleInput,
Modules: map[string]common.Module{"test": &test.TestOutputModule{}}, Modules: map[string]common.Module{"test": &test.TestOutputModule{}},
Payload: payload, Payload: payload,
}) })
internal/processor/test/router-input_test.go
+12 -11
View File
@@ -35,7 +35,7 @@ func TestRouterInputFromRegistry(t *testing.T) {
expected := "test" expected := "test"
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ got, err := processorInstance.Process(t.Context(), common.WrappedPayload{
Router: test.GetNewTestRouter(), InputHandler: test.GetNewTestRouter().HandleInput,
Payload: payload, Payload: payload,
}) })
if err != nil { if err != nil {
@@ -86,18 +86,19 @@ func TestGoodRouterInput(t *testing.T) {
} }
func TestBadRouterInput(t *testing.T) { func TestBadRouterInput(t *testing.T) {
router := test.GetNewTestRouter()
testCases := []struct { testCases := []struct {
name string name string
params map[string]any params map[string]any
payload any payload any
router common.RouteIO inputHandler common.InputHandler
errorString string errorString string
}{ }{
{ {
name: "no source param", name: "no source param",
params: map[string]any{}, params: map[string]any{},
payload: "test", payload: "test",
router: test.GetNewTestRouter(), inputHandler: router.HandleInput,
errorString: "router.input source error: not found", errorString: "router.input source error: not found",
}, },
{ {
@@ -106,7 +107,7 @@ func TestBadRouterInput(t *testing.T) {
"source": 123, "source": 123,
}, },
payload: "test", payload: "test",
router: test.GetNewTestRouter(), inputHandler: router.HandleInput,
errorString: "router.input source error: not a string", errorString: "router.input source error: not a string",
}, },
{ {
@@ -115,8 +116,8 @@ func TestBadRouterInput(t *testing.T) {
"source": "test", "source": "test",
}, },
payload: "test", payload: "test",
router: nil, inputHandler: nil,
errorString: "router.input no router found", errorString: "router.input no input handler found",
}, },
} }
@@ -140,7 +141,7 @@ func TestBadRouterInput(t *testing.T) {
return return
} }
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Router: testCase.router, Payload: testCase.payload}) got, err := processorInstance.Process(t.Context(), common.WrappedPayload{InputHandler: testCase.inputHandler, Payload: testCase.payload})
if err == nil { if err == nil {
t.Fatalf("router.input expected to fail but succeeded, got: %v", got) t.Fatalf("router.input expected to fail but succeeded, got: %v", got)
internal/route/route_test.go
+8 -4
View File
@@ -52,8 +52,9 @@ func TestGoodRouteHandleInput(t *testing.T) {
} }
inputData := "test input data" inputData := "test input data"
testRouter := test.GetNewTestRouter()
payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{}, InputHandler: testRouter.HandleInput,
Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: inputData, Payload: inputData,
}) })
@@ -91,8 +92,9 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) {
} }
inputData := "test input data" inputData := "test input data"
testRouter := test.GetNewTestRouter()
_, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ _, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{}, InputHandler: testRouter.HandleInput,
Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: inputData, Payload: inputData,
}) })
@@ -120,8 +122,9 @@ func TestRouteHandleNilPayload(t *testing.T) {
return return
} }
testRouter := test.GetNewTestRouter()
payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{}, InputHandler: testRouter.HandleInput,
Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: nil, Payload: nil,
}) })
@@ -152,8 +155,9 @@ func TestRouteHandleNilPayloadFromProcessor(t *testing.T) {
t.Fatalf("route failed to create: %v", err) t.Fatalf("route failed to create: %v", err)
} }
testRouter := test.GetNewTestRouter()
_, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ _, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{}, InputHandler: testRouter.HandleInput,
Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: "test", Payload: "test",
}) })
internal/test/module.go
+5 -5
View File
@@ -18,7 +18,7 @@ type TestModule struct {
id string id string
} }
func (m *TestModule) Start(ctx context.Context, router common.RouteIO) error { func (m *TestModule) Start(ctx context.Context, inputHandler common.InputHandler) error {
<-ctx.Done() <-ctx.Done()
return nil return nil
} }
@@ -43,7 +43,7 @@ type TestOutputModule struct {
id string id string
} }
func (m *TestOutputModule) Start(ctx context.Context, router common.RouteIO) error { func (m *TestOutputModule) Start(ctx context.Context, inputHandler common.InputHandler) error {
<-ctx.Done() <-ctx.Done()
return nil return nil
} }
@@ -74,7 +74,7 @@ type TestKVModule struct {
kvData map[string]any kvData map[string]any
} }
func (m *TestKVModule) Start(ctx context.Context, router common.RouteIO) error { func (m *TestKVModule) Start(ctx context.Context, inputHandler common.InputHandler) error {
<-ctx.Done() <-ctx.Done()
return nil return nil
} }
@@ -119,7 +119,7 @@ type TestDBModule struct {
db *sql.DB db *sql.DB
} }
func (m *TestDBModule) Start(ctx context.Context, router common.RouteIO) error { func (m *TestDBModule) Start(ctx context.Context, inputHandler common.InputHandler) error {
<-ctx.Done() <-ctx.Done()
return nil return nil
} }
@@ -167,7 +167,7 @@ type TestPubSubModule struct {
id string id string
} }
func (m *TestPubSubModule) Start(ctx context.Context, router common.RouteIO) error { func (m *TestPubSubModule) Start(ctx context.Context, inputHandler common.InputHandler) error {
<-ctx.Done() <-ctx.Done()
return nil return nil
} }
router.go
+2 -2
View File
@@ -70,7 +70,7 @@ func (r *Router) startModule(ctx context.Context, moduleId string) error {
return errors.New("module id not found") return errors.New("module id not found")
} }
r.moduleWait.Go(func() { r.moduleWait.Go(func() {
err := moduleInstance.Start(ctx, r) err := moduleInstance.Start(ctx, r.HandleInput)
if err != nil { if err != nil {
// TODO(jwetzell): propagate module run errors better // TODO(jwetzell): propagate module run errors better
r.logger.Error("error encountered running module", "moduleId", moduleId, "error", err) r.logger.Error("error encountered running module", "moduleId", moduleId, "error", err)
@@ -211,7 +211,7 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
Payload: payload, Payload: payload,
Source: sourceId, Source: sourceId,
Modules: r.ModuleInstances, Modules: r.ModuleInstances,
Router: r, InputHandler: r.HandleInput,
End: false, End: false,
}) })
if err != nil { if err != nil {
router_test.go
+8 -8
View File
@@ -15,12 +15,12 @@ import (
) )
type MockCounterModule struct { type MockCounterModule struct {
config config.ModuleConfig config config.ModuleConfig
ctx context.Context ctx context.Context
outputCount int outputCount int
router common.RouteIO inputHandler common.InputHandler
logger *slog.Logger logger *slog.Logger
cancel context.CancelFunc cancel context.CancelFunc
} }
func (mcm *MockCounterModule) Id() string { func (mcm *MockCounterModule) Id() string {
@@ -32,8 +32,8 @@ func (mcm *MockCounterModule) Output(context.Context, any) error {
return nil return nil
} }
func (mcm *MockCounterModule) Start(ctx context.Context, router common.RouteIO) error { func (mcm *MockCounterModule) Start(ctx context.Context, inputHandler common.InputHandler) error {
mcm.router = router mcm.inputHandler = inputHandler
moduleContext, cancel := context.WithCancel(ctx) moduleContext, cancel := context.WithCancel(ctx)
mcm.ctx = moduleContext mcm.ctx = moduleContext
mcm.cancel = cancel mcm.cancel = cancel