From 4cedd58a760187d4b88878761deb735528fd221f Mon Sep 17 00:00:00 2001 From: Joel Wetzell Date: Wed, 20 May 2026 21:03:41 -0500 Subject: [PATCH] switch modules to accept an InputHandler instead of the full router --- internal/common/module.go | 2 +- internal/common/payload.go | 10 +++--- internal/common/routing.go | 4 +-- internal/module/db-sqlite.go | 20 +++++------ internal/module/http-server.go | 24 ++++++------- internal/module/midi-input.go | 22 ++++++------ internal/module/midi-output.go | 20 +++++------ internal/module/mqtt-client.go | 32 +++++++++-------- internal/module/nats-client.go | 30 ++++++++-------- internal/module/nats-server.go | 22 ++++++------ internal/module/psn-client.go | 26 +++++++------- internal/module/redis-client.go | 22 ++++++------ internal/module/serial-client.go | 28 +++++++-------- internal/module/sip-call-server.go | 34 ++++++++++--------- internal/module/sip-dtmf-server.go | 32 ++++++++--------- internal/module/tcp-client.go | 28 +++++++-------- internal/module/tcp-server.go | 17 ++++------ internal/module/time-interval.go | 22 ++++++------ internal/module/time-timer.go | 22 ++++++------ internal/module/udp-client.go | 22 ++++++------ internal/module/udp-multicast.go | 26 +++++++------- internal/module/udp-server.go | 28 +++++++-------- internal/module/websocket-client.go | 28 +++++++-------- internal/processor/router-input.go | 6 ++-- internal/processor/test/kv-set_test.go | 2 +- internal/processor/test/module-output_test.go | 4 ++- internal/processor/test/router-input_test.go | 23 +++++++------ internal/route/route_test.go | 12 ++++--- internal/test/module.go | 10 +++--- router.go | 4 +-- router_test.go | 16 ++++----- 31 files changed, 301 insertions(+), 297 deletions(-) diff --git a/internal/common/module.go b/internal/common/module.go index 47d133c..e9616ef 100644 --- a/internal/common/module.go +++ b/internal/common/module.go @@ -8,7 +8,7 @@ import ( type Module interface { Id() string Type() string - Start(context.Context, RouteIO) error + Start(context.Context, InputHandler) error Stop() } diff --git a/internal/common/payload.go b/internal/common/payload.go index ac28bf8..86b3ac6 100644 --- a/internal/common/payload.go +++ b/internal/common/payload.go @@ -1,9 +1,9 @@ package common type WrappedPayload struct { - Payload any - Router RouteIO - Modules map[string]Module - Source string - End bool + Payload any + InputHandler InputHandler + Modules map[string]Module + Source string + End bool } diff --git a/internal/common/routing.go b/internal/common/routing.go index d470cfa..3efa98a 100644 --- a/internal/common/routing.go +++ b/internal/common/routing.go @@ -4,9 +4,7 @@ import ( "context" ) -type RouteIO interface { - HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError) -} +type InputHandler func(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError) type RouteIOError struct { Index int `json:"index"` diff --git a/internal/module/db-sqlite.go b/internal/module/db-sqlite.go index aeeae46..db83186 100644 --- a/internal/module/db-sqlite.go +++ b/internal/module/db-sqlite.go @@ -15,14 +15,14 @@ import ( ) type DbSqlite struct { - config config.ModuleConfig - Dsn string - ctx context.Context - router common.RouteIO - db *sql.DB - logger *slog.Logger - dbMu sync.Mutex - cancel context.CancelFunc + config config.ModuleConfig + Dsn string + ctx context.Context + inputHandler common.InputHandler + db *sql.DB + logger *slog.Logger + dbMu sync.Mutex + cancel context.CancelFunc } func init() { @@ -61,9 +61,9 @@ func (dbs *DbSqlite) Type() string { 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.router = router + dbs.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) dbs.ctx = moduleContext dbs.cancel = cancel diff --git a/internal/module/http-server.go b/internal/module/http-server.go index ef5bedb..25917e5 100644 --- a/internal/module/http-server.go +++ b/internal/module/http-server.go @@ -17,14 +17,14 @@ import ( ) type HTTPServer struct { - config config.ModuleConfig - Port uint16 - ctx context.Context - router common.RouteIO - logger *slog.Logger - cancel context.CancelFunc - server *http.Server - serverMu sync.Mutex + config config.ModuleConfig + Port uint16 + ctx context.Context + inputHandler common.InputHandler + logger *slog.Logger + cancel context.CancelFunc + server *http.Server + serverMu sync.Mutex } type ResponseIOError struct { @@ -98,9 +98,9 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { Message: "routing successful", Status: "ok", } - if hs.router != nil { + if hs.inputHandler != nil { 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 aRouteFound { 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.router = router + hs.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) hs.ctx = moduleContext hs.cancel = cancel diff --git a/internal/module/midi-input.go b/internal/module/midi-input.go index 86f7e99..55a440f 100644 --- a/internal/module/midi-input.go +++ b/internal/module/midi-input.go @@ -15,13 +15,13 @@ import ( ) type MIDIInput struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Port string - logger *slog.Logger - cancel context.CancelFunc - stop func() + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + Port string + logger *slog.Logger + cancel context.CancelFunc + stop func() } func init() { @@ -59,9 +59,9 @@ func (mi *MIDIInput) Type() string { 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.router = router + mi.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) mi.ctx = moduleContext 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) { - if mi.router != nil { - mi.router.HandleInput(mi.ctx, mi.Id(), msg) + if mi.inputHandler != nil { + mi.inputHandler(mi.ctx, mi.Id(), msg) } }, midi.UseSysEx()) diff --git a/internal/module/midi-output.go b/internal/module/midi-output.go index 5eb7d97..5c60a42 100644 --- a/internal/module/midi-output.go +++ b/internal/module/midi-output.go @@ -17,14 +17,14 @@ import ( ) type MIDIOutput struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Port string - sendFunc func(midi.Message) error - logger *slog.Logger - cancel context.CancelFunc - sendFuncMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + Port string + sendFunc func(midi.Message) error + logger *slog.Logger + cancel context.CancelFunc + sendFuncMu sync.Mutex } func init() { @@ -63,9 +63,9 @@ func (mo *MIDIOutput) Type() string { 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.router = router + mo.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) mo.ctx = moduleContext mo.cancel = cancel diff --git a/internal/module/mqtt-client.go b/internal/module/mqtt-client.go index 2c56f63..5fb9f69 100644 --- a/internal/module/mqtt-client.go +++ b/internal/module/mqtt-client.go @@ -15,18 +15,18 @@ import ( ) type MQTTClient struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Broker string - ClientID string - Topic string - QoS byte - Retained bool - client mqtt.Client - logger *slog.Logger - cancel context.CancelFunc - clientMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + Broker string + ClientID string + Topic string + QoS byte + Retained bool + client mqtt.Client + logger *slog.Logger + cancel context.CancelFunc + clientMu sync.Mutex } func init() { @@ -117,9 +117,9 @@ func (mc *MQTTClient) Type() string { 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.router = router + mc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) mc.ctx = moduleContext mc.cancel = cancel @@ -132,7 +132,9 @@ func (mc *MQTTClient) Start(ctx context.Context, router common.RouteIO) error { opts.OnConnect = func(c mqtt.Client) { 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() } diff --git a/internal/module/nats-client.go b/internal/module/nats-client.go index 5f3a647..8f82dbf 100644 --- a/internal/module/nats-client.go +++ b/internal/module/nats-client.go @@ -13,17 +13,17 @@ import ( ) type NATSClient struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - URL string - Subject string - client *nats.Conn - logger *slog.Logger - cancel context.CancelFunc - sub *nats.Subscription - subMu sync.Mutex - clientMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + URL string + Subject string + client *nats.Conn + logger *slog.Logger + cancel context.CancelFunc + sub *nats.Subscription + subMu sync.Mutex + clientMu sync.Mutex } func init() { @@ -71,9 +71,9 @@ func (nc *NATSClient) Type() string { 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.router = router + nc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) nc.ctx = moduleContext nc.cancel = cancel @@ -89,8 +89,8 @@ func (nc *NATSClient) Start(ctx context.Context, router common.RouteIO) error { nc.clientMu.Unlock() sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) { - if nc.router != nil { - nc.router.HandleInput(nc.ctx, nc.Id(), msg) + if nc.inputHandler != nil { + nc.inputHandler(nc.ctx, nc.Id(), msg) } }) diff --git a/internal/module/nats-server.go b/internal/module/nats-server.go index 1a3131b..2a16f94 100644 --- a/internal/module/nats-server.go +++ b/internal/module/nats-server.go @@ -17,15 +17,15 @@ import ( ) type NATSServer struct { - config config.ModuleConfig - ctx context.Context - Ip string - Port int - router common.RouteIO - server *server.Server - logger *slog.Logger - cancel context.CancelFunc - serverMu sync.Mutex + config config.ModuleConfig + ctx context.Context + Ip string + Port int + inputHandler common.InputHandler + server *server.Server + logger *slog.Logger + cancel context.CancelFunc + serverMu sync.Mutex } func init() { @@ -88,9 +88,9 @@ func (ns *NATSServer) Type() string { 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.router = router + ns.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) ns.ctx = moduleContext ns.cancel = cancel diff --git a/internal/module/psn-client.go b/internal/module/psn-client.go index 3f87d1f..86a9ef7 100644 --- a/internal/module/psn-client.go +++ b/internal/module/psn-client.go @@ -13,14 +13,14 @@ import ( ) type PSNClient struct { - config config.ModuleConfig - conn *net.UDPConn - ctx context.Context - router common.RouteIO - decoder *psn.Decoder - logger *slog.Logger - cancel context.CancelFunc - connMu sync.Mutex + config config.ModuleConfig + conn *net.UDPConn + ctx context.Context + inputHandler common.InputHandler + decoder *psn.Decoder + logger *slog.Logger + cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -41,9 +41,9 @@ func (pc *PSNClient) Type() string { 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.router = router + pc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) pc.ctx = moduleContext 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) } - if pc.router != nil { + if pc.inputHandler != nil { // TODO(jwetzell): better input handling for _, tracker := range pc.decoder.Trackers { - pc.router.HandleInput(pc.ctx, pc.Id(), tracker) + pc.inputHandler(pc.ctx, pc.Id(), tracker) } } else { - pc.logger.Error("has no router") + pc.logger.Error("has no input handler") } } } diff --git a/internal/module/redis-client.go b/internal/module/redis-client.go index 9212fa7..5b5f7a4 100644 --- a/internal/module/redis-client.go +++ b/internal/module/redis-client.go @@ -14,15 +14,15 @@ import ( ) type RedisClient struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Host string - Port uint16 - client *redis.Client - logger *slog.Logger - cancel context.CancelFunc - clientMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + Host string + Port uint16 + client *redis.Client + logger *slog.Logger + cancel context.CancelFunc + clientMu sync.Mutex } func init() { @@ -75,10 +75,10 @@ func (rc *RedisClient) Printf(ctx context.Context, format string, v ...any) { 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) rc.logger.Debug("running") - rc.router = router + rc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) rc.ctx = moduleContext rc.cancel = cancel diff --git a/internal/module/serial-client.go b/internal/module/serial-client.go index b788573..e18e611 100644 --- a/internal/module/serial-client.go +++ b/internal/module/serial-client.go @@ -18,16 +18,16 @@ import ( ) type SerialClient struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Port string - Framer framer.Framer - Mode *serial.Mode - port serial.Port - logger *slog.Logger - cancel context.CancelFunc - portMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + Port string + Framer framer.Framer + Mode *serial.Mode + port serial.Port + logger *slog.Logger + cancel context.CancelFunc + portMu sync.Mutex } func init() { @@ -107,9 +107,9 @@ func (sc *SerialClient) SetupPort() error { 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.router = router + sc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) sc.ctx = moduleContext sc.cancel = cancel @@ -147,8 +147,8 @@ func (sc *SerialClient) Start(ctx context.Context, router common.RouteIO) error if byteCount > 0 { messages := sc.Framer.Decode(buffer[0:byteCount]) for _, message := range messages { - if sc.router != nil { - sc.router.HandleInput(sc.ctx, sc.Id(), message) + if sc.inputHandler != nil { + sc.inputHandler(sc.ctx, sc.Id(), message) } else { sc.logger.Error("input received but no router is configured") } diff --git a/internal/module/sip-call-server.go b/internal/module/sip-call-server.go index f407dc7..d86ddea 100644 --- a/internal/module/sip-call-server.go +++ b/internal/module/sip-call-server.go @@ -22,17 +22,17 @@ import ( ) type SIPCallServer struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - IP string - Port int - Transport string - UserAgent string - logger *slog.Logger - cancel context.CancelFunc - ua *sipgo.UserAgent - uaMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + IP string + Port int + Transport string + UserAgent string + logger *slog.Logger + cancel context.CancelFunc + ua *sipgo.UserAgent + uaMu sync.Mutex } type SIPCallMessage struct { @@ -132,9 +132,9 @@ func (scs *SIPCallServer) Type() string { 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.router = router + scs.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) scs.ctx = moduleContext scs.cancel = cancel @@ -179,9 +179,11 @@ func (scs *SIPCallServer) HandleCall(inDialog *diago.DialogServerSession) { dialogContext := context.WithValue(scs.ctx, sipCallContextKey("call"), &SIPCall{ inDialog: inDialog, }) - scs.router.HandleInput(dialogContext, scs.Id(), SIPCallMessage{ - To: inDialog.ToUser(), - }) + if scs.inputHandler != nil { + scs.inputHandler(dialogContext, scs.Id(), SIPCallMessage{ + To: inDialog.ToUser(), + }) + } } func (scs *SIPCallServer) Output(ctx context.Context, payload any) error { diff --git a/internal/module/sip-dtmf-server.go b/internal/module/sip-dtmf-server.go index 9ef4f95..5036c2f 100644 --- a/internal/module/sip-dtmf-server.go +++ b/internal/module/sip-dtmf-server.go @@ -23,18 +23,18 @@ import ( ) type SIPDTMFServer struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - IP string - Port int - Transport string - UserAgent string - Separator string - logger *slog.Logger - cancel context.CancelFunc - ua *sipgo.UserAgent - uaMu sync.Mutex + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + IP string + Port int + Transport string + UserAgent string + Separator string + logger *slog.Logger + cancel context.CancelFunc + ua *sipgo.UserAgent + uaMu sync.Mutex } type SIPDTMFMessage struct { @@ -152,9 +152,9 @@ func (sds *SIPDTMFServer) Type() string { 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.router = router + sds.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) sds.ctx = moduleContext sds.cancel = cancel @@ -203,11 +203,11 @@ func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error return reader.Listen(func(dtmf rune) error { if dtmf == rune(sds.Separator[0]) { - if sds.router != nil { + if sds.inputHandler != nil { dialogContext := context.WithValue(sds.ctx, sipCallContextKey("call"), &SIPDTMFCall{ inDialog: inDialog, }) - sds.router.HandleInput(dialogContext, sds.Id(), SIPDTMFMessage{ + sds.inputHandler(dialogContext, sds.Id(), SIPDTMFMessage{ To: inDialog.ToUser(), Digits: userString, }) diff --git a/internal/module/tcp-client.go b/internal/module/tcp-client.go index d1d71a9..a41a799 100644 --- a/internal/module/tcp-client.go +++ b/internal/module/tcp-client.go @@ -16,15 +16,15 @@ import ( ) type TCPClient struct { - config config.ModuleConfig - framer framer.Framer - conn *net.TCPConn - ctx context.Context - router common.RouteIO - Addr *net.TCPAddr - logger *slog.Logger - cancel context.CancelFunc - connMu sync.Mutex + config config.ModuleConfig + framer framer.Framer + conn *net.TCPConn + ctx context.Context + inputHandler common.InputHandler + Addr *net.TCPAddr + logger *slog.Logger + cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -93,9 +93,9 @@ func (tc *TCPClient) Type() string { 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.router = router + tc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) tc.ctx = moduleContext tc.cancel = cancel @@ -133,10 +133,10 @@ func (tc *TCPClient) Start(ctx context.Context, router common.RouteIO) error { if byteCount > 0 { messages := tc.framer.Decode(buffer[0:byteCount]) for _, message := range messages { - if tc.router != nil { - tc.router.HandleInput(tc.ctx, tc.Id(), message) + if tc.inputHandler != nil { + tc.inputHandler(tc.ctx, tc.Id(), message) } else { - tc.logger.Error("input received but no router is configured") + tc.logger.Error("input received but no input handler is configured") } } } diff --git a/internal/module/tcp-server.go b/internal/module/tcp-server.go index 5efc383..88f5cd1 100644 --- a/internal/module/tcp-server.go +++ b/internal/module/tcp-server.go @@ -24,7 +24,7 @@ type TCPServer struct { Addr *net.TCPAddr Framer framer.Framer ctx context.Context - router common.RouteIO + inputHandler common.InputHandler wg sync.WaitGroup connections []*net.TCPConn connectionsMu sync.RWMutex @@ -166,15 +166,10 @@ ClientRead: if byteCount > 0 { messages := ts.Framer.Decode(buffer[0:byteCount]) for _, message := range messages { - if ts.router != nil { - _, ok := client.RemoteAddr().(*net.TCPAddr) - if ok { - ts.router.HandleInput(ts.ctx, ts.Id(), message) - } else { - ts.router.HandleInput(ts.ctx, ts.Id(), message) - } + if ts.inputHandler != nil { + ts.inputHandler(ts.ctx, ts.Id(), message) } 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.router = router + ts.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) ts.ctx = moduleContext ts.cancel = cancel diff --git a/internal/module/time-interval.go b/internal/module/time-interval.go index c9e7b8f..0bf1204 100644 --- a/internal/module/time-interval.go +++ b/internal/module/time-interval.go @@ -12,13 +12,13 @@ import ( ) type TimeInterval struct { - config config.ModuleConfig - Duration uint32 - ctx context.Context - router common.RouteIO - ticker *time.Ticker - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + Duration uint32 + ctx context.Context + inputHandler common.InputHandler + ticker *time.Ticker + logger *slog.Logger + cancel context.CancelFunc } func init() { @@ -57,9 +57,9 @@ func (i *TimeInterval) Type() string { 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.router = router + i.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) i.ctx = moduleContext i.cancel = cancel @@ -72,8 +72,8 @@ func (i *TimeInterval) Start(ctx context.Context, router common.RouteIO) error { case <-i.ctx.Done(): return nil case <-ticker.C: - if i.router != nil { - i.router.HandleInput(i.ctx, i.Id(), time.Now()) + if i.inputHandler != nil { + i.inputHandler(i.ctx, i.Id(), time.Now()) } } } diff --git a/internal/module/time-timer.go b/internal/module/time-timer.go index 1be1fa1..d466621 100644 --- a/internal/module/time-timer.go +++ b/internal/module/time-timer.go @@ -12,13 +12,13 @@ import ( ) type TimeTimer struct { - config config.ModuleConfig - Duration uint32 - ctx context.Context - router common.RouteIO - timer *time.Timer - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + Duration uint32 + ctx context.Context + inputHandler common.InputHandler + timer *time.Timer + logger *slog.Logger + cancel context.CancelFunc } func init() { @@ -58,9 +58,9 @@ func (t *TimeTimer) Type() string { 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.router = router + t.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) t.ctx = moduleContext t.cancel = cancel @@ -71,8 +71,8 @@ func (t *TimeTimer) Start(ctx context.Context, router common.RouteIO) error { case <-t.ctx.Done(): return nil case time := <-t.timer.C: - if t.router != nil { - t.router.HandleInput(t.ctx, t.Id(), time) + if t.inputHandler != nil { + t.inputHandler(t.ctx, t.Id(), time) } } } diff --git a/internal/module/udp-client.go b/internal/module/udp-client.go index 273dab9..b8471b3 100644 --- a/internal/module/udp-client.go +++ b/internal/module/udp-client.go @@ -14,15 +14,15 @@ import ( ) type UDPClient struct { - config config.ModuleConfig - Addr *net.UDPAddr - Port uint16 - conn *net.UDPConn - ctx context.Context - router common.RouteIO - logger *slog.Logger - cancel context.CancelFunc - connMu sync.Mutex + config config.ModuleConfig + Addr *net.UDPAddr + Port uint16 + conn *net.UDPConn + ctx context.Context + inputHandler common.InputHandler + logger *slog.Logger + cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -83,9 +83,9 @@ func (uc *UDPClient) SetupConn() error { 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.router = router + uc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) uc.ctx = moduleContext uc.cancel = cancel diff --git a/internal/module/udp-multicast.go b/internal/module/udp-multicast.go index b85554d..4177aab 100644 --- a/internal/module/udp-multicast.go +++ b/internal/module/udp-multicast.go @@ -15,14 +15,14 @@ import ( ) type UDPMulticast struct { - config config.ModuleConfig - conn *net.UDPConn - ctx context.Context - router common.RouteIO - Addr *net.UDPAddr - logger *slog.Logger - cancel context.CancelFunc - connMu sync.Mutex + config config.ModuleConfig + conn *net.UDPConn + ctx context.Context + inputHandler common.InputHandler + Addr *net.UDPAddr + logger *slog.Logger + cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -75,9 +75,9 @@ func (um *UDPMulticast) Type() string { 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.router = router + um.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) um.ctx = moduleContext um.cancel = cancel @@ -114,10 +114,10 @@ func (um *UDPMulticast) Start(ctx context.Context, router common.RouteIO) error if numBytes > 0 { message := buffer[:numBytes] - if um.router != nil { - um.router.HandleInput(um.ctx, um.Id(), message) + if um.inputHandler != nil { + um.inputHandler(um.ctx, um.Id(), message) } else { - um.logger.Error("input received but no router is configured") + um.logger.Error("input received but no input handler is configured") } } } diff --git a/internal/module/udp-server.go b/internal/module/udp-server.go index 2623473..fbb3650 100644 --- a/internal/module/udp-server.go +++ b/internal/module/udp-server.go @@ -16,15 +16,15 @@ import ( ) type UDPServer struct { - Addr *net.UDPAddr - BufferSize int - config config.ModuleConfig - ctx context.Context - router common.RouteIO - logger *slog.Logger - cancel context.CancelFunc - listener *net.UDPConn - listenerMu sync.Mutex + Addr *net.UDPAddr + BufferSize int + config config.ModuleConfig + ctx context.Context + inputHandler common.InputHandler + logger *slog.Logger + cancel context.CancelFunc + listener *net.UDPConn + listenerMu sync.Mutex } func init() { @@ -98,9 +98,9 @@ func (us *UDPServer) Type() string { 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.router = router + us.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) us.ctx = moduleContext us.cancel = cancel @@ -129,10 +129,10 @@ func (us *UDPServer) Start(ctx context.Context, router common.RouteIO) error { return err } message := buffer[:numBytes] - if us.router != nil { - us.router.HandleInput(us.ctx, us.Id(), message) + if us.inputHandler != nil { + us.inputHandler(us.ctx, us.Id(), message) } else { - us.logger.Error("input received but no router is configured") + us.logger.Error("input received but no input handler is configured") } } } diff --git a/internal/module/websocket-client.go b/internal/module/websocket-client.go index 2f0488d..bd3586e 100644 --- a/internal/module/websocket-client.go +++ b/internal/module/websocket-client.go @@ -17,14 +17,14 @@ import ( ) type WebSocketClient struct { - config config.ModuleConfig - URL url.URL - ctx context.Context - conn *websocket.Conn - router common.RouteIO - logger *slog.Logger - cancel context.CancelFunc - connMu sync.Mutex + config config.ModuleConfig + URL url.URL + ctx context.Context + conn *websocket.Conn + inputHandler common.InputHandler + logger *slog.Logger + cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -79,9 +79,9 @@ func (wc *WebSocketClient) SetupConn() error { 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.router = router + wc.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) wc.ctx = moduleContext wc.cancel = cancel @@ -124,17 +124,17 @@ func (wc *WebSocketClient) readLoop() { wc.logger.Error("websocket read error", "error", err) return } - if wc.router != nil { + if wc.inputHandler != nil { switch messageType { case websocket.TextMessage: - wc.router.HandleInput(wc.ctx, wc.Id(), string(message)) + wc.inputHandler(wc.ctx, wc.Id(), string(message)) case websocket.BinaryMessage: - wc.router.HandleInput(wc.ctx, wc.Id(), message) + wc.inputHandler(wc.ctx, wc.Id(), message) default: wc.logger.Warn("unsupported message type received", "messageType", messageType) } } else { - wc.logger.Error("input received but no router is configured") + wc.logger.Error("input received but no input handler is configured") continue } } diff --git a/internal/processor/router-input.go b/internal/processor/router-input.go index 9626279..4d2ae0c 100644 --- a/internal/processor/router-input.go +++ b/internal/processor/router-input.go @@ -20,12 +20,12 @@ type RouterInput struct { func (ro *RouterInput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) { payload := wrappedPayload.Payload - if wrappedPayload.Router == nil { + if wrappedPayload.InputHandler == nil { 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 { wrappedPayload.End = true diff --git a/internal/processor/test/kv-set_test.go b/internal/processor/test/kv-set_test.go index 45f1373..e758d63 100644 --- a/internal/processor/test/kv-set_test.go +++ b/internal/processor/test/kv-set_test.go @@ -113,7 +113,7 @@ func TestBadKvSet(t *testing.T) { name: "no module param", payload: test.TestStruct{Data: "hello"}, params: map[string]any{ - "key": "test", + "key": "test", }, wrappedPayloadModules: map[string]common.Module{ "test": &test.TestKVModule{}, diff --git a/internal/processor/test/module-output_test.go b/internal/processor/test/module-output_test.go index 8ea7ffa..b35908d 100644 --- a/internal/processor/test/module-output_test.go +++ b/internal/processor/test/module-output_test.go @@ -34,8 +34,10 @@ func TestModuleOutputFromRegistry(t *testing.T) { payload := "test" expected := "test" + router := test.GetNewTestRouter() + got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ - Router: test.GetNewTestRouter(), + InputHandler: router.HandleInput, Modules: map[string]common.Module{"test": &test.TestOutputModule{}}, Payload: payload, }) diff --git a/internal/processor/test/router-input_test.go b/internal/processor/test/router-input_test.go index b3859bf..8aca9e0 100644 --- a/internal/processor/test/router-input_test.go +++ b/internal/processor/test/router-input_test.go @@ -35,7 +35,7 @@ func TestRouterInputFromRegistry(t *testing.T) { expected := "test" got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ - Router: test.GetNewTestRouter(), + InputHandler: test.GetNewTestRouter().HandleInput, Payload: payload, }) if err != nil { @@ -86,18 +86,19 @@ func TestGoodRouterInput(t *testing.T) { } func TestBadRouterInput(t *testing.T) { + router := test.GetNewTestRouter() testCases := []struct { - name string - params map[string]any - payload any - router common.RouteIO - errorString string + name string + params map[string]any + payload any + inputHandler common.InputHandler + errorString string }{ { name: "no source param", params: map[string]any{}, payload: "test", - router: test.GetNewTestRouter(), + inputHandler: router.HandleInput, errorString: "router.input source error: not found", }, { @@ -106,7 +107,7 @@ func TestBadRouterInput(t *testing.T) { "source": 123, }, payload: "test", - router: test.GetNewTestRouter(), + inputHandler: router.HandleInput, errorString: "router.input source error: not a string", }, { @@ -115,8 +116,8 @@ func TestBadRouterInput(t *testing.T) { "source": "test", }, payload: "test", - router: nil, - errorString: "router.input no router found", + inputHandler: nil, + errorString: "router.input no input handler found", }, } @@ -140,7 +141,7 @@ func TestBadRouterInput(t *testing.T) { 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 { t.Fatalf("router.input expected to fail but succeeded, got: %v", got) diff --git a/internal/route/route_test.go b/internal/route/route_test.go index 9075126..f7ee655 100644 --- a/internal/route/route_test.go +++ b/internal/route/route_test.go @@ -52,8 +52,9 @@ func TestGoodRouteHandleInput(t *testing.T) { } inputData := "test input data" + testRouter := test.GetNewTestRouter() payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ - Router: &MockRouter{}, + InputHandler: testRouter.HandleInput, Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Payload: inputData, }) @@ -91,8 +92,9 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) { } inputData := "test input data" + testRouter := test.GetNewTestRouter() _, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ - Router: &MockRouter{}, + InputHandler: testRouter.HandleInput, Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Payload: inputData, }) @@ -120,8 +122,9 @@ func TestRouteHandleNilPayload(t *testing.T) { return } + testRouter := test.GetNewTestRouter() payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ - Router: &MockRouter{}, + InputHandler: testRouter.HandleInput, Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Payload: nil, }) @@ -152,8 +155,9 @@ func TestRouteHandleNilPayloadFromProcessor(t *testing.T) { t.Fatalf("route failed to create: %v", err) } + testRouter := test.GetNewTestRouter() _, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{ - Router: &MockRouter{}, + InputHandler: testRouter.HandleInput, Modules: map[string]common.Module{"output": &test.TestOutputModule{}}, Payload: "test", }) diff --git a/internal/test/module.go b/internal/test/module.go index 52c4646..21a25fc 100644 --- a/internal/test/module.go +++ b/internal/test/module.go @@ -18,7 +18,7 @@ type TestModule struct { 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() return nil } @@ -43,7 +43,7 @@ type TestOutputModule struct { 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() return nil } @@ -74,7 +74,7 @@ type TestKVModule struct { 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() return nil } @@ -119,7 +119,7 @@ type TestDBModule struct { 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() return nil } @@ -167,7 +167,7 @@ type TestPubSubModule struct { 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() return nil } diff --git a/router.go b/router.go index a4fc3c8..3b9f24c 100644 --- a/router.go +++ b/router.go @@ -70,7 +70,7 @@ func (r *Router) startModule(ctx context.Context, moduleId string) error { return errors.New("module id not found") } r.moduleWait.Go(func() { - err := moduleInstance.Start(ctx, r) + err := moduleInstance.Start(ctx, r.HandleInput) if err != nil { // TODO(jwetzell): propagate module run errors better 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, Source: sourceId, Modules: r.ModuleInstances, - Router: r, + InputHandler: r.HandleInput, End: false, }) if err != nil { diff --git a/router_test.go b/router_test.go index 982e493..223b76d 100644 --- a/router_test.go +++ b/router_test.go @@ -15,12 +15,12 @@ import ( ) type MockCounterModule struct { - config config.ModuleConfig - ctx context.Context - outputCount int - router common.RouteIO - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + ctx context.Context + outputCount int + inputHandler common.InputHandler + logger *slog.Logger + cancel context.CancelFunc } func (mcm *MockCounterModule) Id() string { @@ -32,8 +32,8 @@ func (mcm *MockCounterModule) Output(context.Context, any) error { return nil } -func (mcm *MockCounterModule) Start(ctx context.Context, router common.RouteIO) error { - mcm.router = router +func (mcm *MockCounterModule) Start(ctx context.Context, inputHandler common.InputHandler) error { + mcm.inputHandler = inputHandler moduleContext, cancel := context.WithCancel(ctx) mcm.ctx = moduleContext mcm.cancel = cancel