backup/showbridge-go
1
0
Fork 0
mirror of https://github.com/jwetzell/showbridge-go.git synced 2026-04-29 06:15:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

propagate a ctx all the way from input to output of a route

Browse Source
This commit is contained in:
Joel Wetzell
2025-12-28 12:21:58 -06:00
parent 12de947f3d
commit f97f9b9fc9
18 changed files with 21 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/module/http-client.go
View File

@@ -73,7 +73,7 @@ func (hc *HTTPClient) Output(ctx context.Context, payload any) error {
} }
if hc.router != nil { if hc.router != nil {
hc.router.HandleInput(hc.Id(), response) hc.router.HandleInput(hc.ctx, hc.Id(), response)
} }
return nil return nil

2
internal/module/http-server.go
View File

@@ -77,7 +77,7 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
if hs.router != nil { if hs.router != nil {
aRouteFound, routingErrors := hs.router.HandleInput(hs.Id(), r) aRouteFound, routingErrors := hs.router.HandleInput(hs.ctx, hs.Id(), r)
if aRouteFound { if aRouteFound {
if routingErrors != nil { if routingErrors != nil {
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)

2
internal/module/midi-input.go
View File

@@ -69,7 +69,7 @@ func (mi *MIDIInput) Run() 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.router != nil {
mi.router.HandleInput(mi.Id(), msg) mi.router.HandleInput(mi.ctx, mi.Id(), msg)
} }
}, midi.UseSysEx()) }, midi.UseSysEx())

2
internal/module/mqtt-client.go
View File

@@ -90,7 +90,7 @@ func (mc *MQTTClient) Run() 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.Id(), m) mc.router.HandleInput(mc.ctx, mc.Id(), m)
}) })
token.Wait() token.Wait()
} }

2
internal/module/nats-client.go
View File

@@ -83,7 +83,7 @@ func (nc *NATSClient) Run() error {
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.router != nil {
nc.router.HandleInput(nc.Id(), msg) nc.router.HandleInput(nc.ctx, nc.Id(), msg)
} }
}) })

2
internal/module/psn-client.go
View File

@@ -87,7 +87,7 @@ func (pc *PSNClient) Run() error {
if pc.router != nil { if pc.router != nil {
for _, tracker := range pc.decoder.Trackers { for _, tracker := range pc.decoder.Trackers {
pc.router.HandleInput(pc.Id(), tracker) pc.router.HandleInput(pc.ctx, pc.Id(), tracker)
} }
} else { } else {
pc.logger.Error("has no router") pc.logger.Error("has no router")

2
internal/module/serial-client.go
View File

@@ -155,7 +155,7 @@ func (sc *SerialClient) Run() error {
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.router != nil {
sc.router.HandleInput(sc.Id(), message) sc.router.HandleInput(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")
} }

2
internal/module/sip-call-server.go
View File

@@ -143,7 +143,7 @@ func (scs *SIPCallServer) HandleCall(inDialog *diago.DialogServerSession) {
inDialog.Trying() inDialog.Trying()
inDialog.Ringing() inDialog.Ringing()
inDialog.Answer() inDialog.Answer()
scs.router.HandleInput(scs.Id(), SIPCallMessage{ scs.router.HandleInput(scs.ctx, scs.Id(), SIPCallMessage{
To: inDialog.ToUser(), To: inDialog.ToUser(),
}) })
<-inDialog.Context().Done() <-inDialog.Context().Done()

2
internal/module/sip-dtmf-server.go
View File

@@ -151,7 +151,7 @@ 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.router != nil {
sds.router.HandleInput(sds.Id(), SIPDTMFMessage{ sds.router.HandleInput(sds.ctx, sds.Id(), SIPDTMFMessage{
To: inDialog.ToUser(), To: inDialog.ToUser(),
Digits: userString, Digits: userString,
}) })

2
internal/module/tcp-client.go
View File

@@ -142,7 +142,7 @@ func (tc *TCPClient) Run() error {
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.router != nil {
tc.router.HandleInput(tc.Id(), message) tc.router.HandleInput(tc.ctx, tc.Id(), message)
} else { } else {
tc.logger.Error("input received but no router is configured") tc.logger.Error("input received but no router is configured")
} }

2
internal/module/tcp-server.go
View File

@@ -155,7 +155,7 @@ ClientRead:
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.router != nil {
ts.router.HandleInput(ts.Id(), message) 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 router is configured")
} }

2
internal/module/time-interval.go
View File

@@ -67,7 +67,7 @@ func (i *TimeInterval) Run() error {
return nil return nil
case <-ticker.C: case <-ticker.C:
if i.router != nil { if i.router != nil {
i.router.HandleInput(i.Id(), time.Now()) i.router.HandleInput(i.ctx, i.Id(), time.Now())
} }
} }
} }

2
internal/module/time-timer.go
View File

@@ -66,7 +66,7 @@ func (t *TimeTimer) Run() error {
return nil return nil
case time := <-t.timer.C: case time := <-t.timer.C:
if t.router != nil { if t.router != nil {
t.router.HandleInput(t.Id(), time) t.router.HandleInput(t.ctx, t.Id(), time)
} }
} }
} }

2
internal/module/udp-multicast.go
View File

@@ -105,7 +105,7 @@ func (um *UDPMulticast) Run() error {
message := buffer[:numBytes] message := buffer[:numBytes]
if um.router != nil { if um.router != nil {
um.router.HandleInput(um.Id(), message) um.router.HandleInput(um.ctx, um.Id(), message)
} else { } else {
um.logger.Error("input received but no router is configured") um.logger.Error("input received but no router is configured")
} }

2
internal/module/udp-server.go
View File

@@ -115,7 +115,7 @@ func (us *UDPServer) Run() error {
} }
message := buffer[:numBytes] message := buffer[:numBytes]
if us.router != nil { if us.router != nil {
us.router.HandleInput(us.Id(), message) us.router.HandleInput(us.ctx, us.Id(), message)
} else { } else {
us.logger.Error("input received but no router is configured") us.logger.Error("input received but no router is configured")
} }

2
internal/route/route.go
View File

@@ -27,7 +27,7 @@ type RouteIOError struct {
} }
type RouteIO interface { type RouteIO interface {
HandleInput(sourceId string, payload any) (bool, []RouteIOError) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)
HandleOutput(ctx context.Context, destinationId string, payload any) []error HandleOutput(ctx context.Context, destinationId string, payload any) []error
} }

4
internal/route/route_test.go
View File

@@ -30,8 +30,8 @@ func TestRouteCreate(t *testing.T) {
type MockRouter struct{} type MockRouter struct{}
func (mr *MockRouter) HandleInput(sourceId string, payload any) []route.RouteIOError { func (mr *MockRouter) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
return nil return false, []route.RouteIOError{}
} }
func (mr *MockRouter) HandleOutput(ctx context.Context, destinationId string, payload any) error { func (mr *MockRouter) HandleOutput(ctx context.Context, destinationId string, payload any) error {

7
router.go
View File

@@ -3,7 +3,6 @@ package showbridge
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"log/slog" "log/slog"
"sync" "sync"
@@ -129,13 +128,13 @@ func (r *Router) Stop() {
r.contextCancel() r.contextCancel()
} }
func (r *Router) HandleInput(sourceId string, payload any) (bool, []route.RouteIOError) { func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
var routeIOErrors []route.RouteIOError var routeIOErrors []route.RouteIOError
routeFound := false routeFound := false
for routeIndex, routeInstance := range r.RouteInstances { for routeIndex, routeInstance := range r.RouteInstances {
if routeInstance.Input() == sourceId { if routeInstance.Input() == sourceId {
routeFound = true routeFound = true
routeContext := context.WithValue(r.Context, route.SourceContextKey, sourceId) routeContext := context.WithValue(ctx, route.SourceContextKey, sourceId)
payload, err := routeInstance.ProcessPayload(routeContext, payload) payload, err := routeInstance.ProcessPayload(routeContext, payload)
if err != nil { if err != nil {
@@ -177,8 +176,8 @@ func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload
outputErrors = []error{} outputErrors = []error{}
} }
outputErrors = append(outputErrors, err) outputErrors = append(outputErrors, err)
r.logger.Error("unable to route output", "module", moduleInstance.Id(), "error", err)
} }
// r.logger.Error("unable to route output", "module", moduleInstance.Id(), "error", err)
} }
} }
return outputErrors return outputErrors