backup/showbridge-go
1
0
Fork 0
mirror of https://github.com/jwetzell/showbridge-go.git synced 2026-04-26 21:05:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

process routes concurrently

Browse Source
This commit is contained in:
Joel Wetzell
2025-12-28 20:42:11 -06:00
parent bd2a68ff6e
commit 59f00c1a32
1 changed files with 34 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
router.go
View File

@@ -131,42 +131,49 @@ func (r *Router) Stop() {
func (r *Router) HandleInput(ctx context.Context, 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
var routeWaitGroup sync.WaitGroup
for routeIndex, routeInstance := range r.RouteInstances { for routeIndex, routeInstance := range r.RouteInstances {
if routeInstance.Input() == sourceId { if routeInstance.Input() == sourceId {
routeFound = true routeWaitGroup.Go(func() {
routeContext := context.WithValue(ctx, route.SourceContextKey, sourceId)
payload, err := routeInstance.ProcessPayload(routeContext, payload) routeFound = true
if err != nil { routeContext := context.WithValue(ctx, route.SourceContextKey, sourceId)
if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{} payload, err := routeInstance.ProcessPayload(routeContext, payload)
if err != nil {
if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{}
}
r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
routeIOErrors = append(routeIOErrors, route.RouteIOError{
Index: routeIndex,
ProcessError: err,
})
return
} }
r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
routeIOErrors = append(routeIOErrors, route.RouteIOError{
Index: routeIndex,
ProcessError: err,
})
continue
}
if payload == nil { if payload == nil {
r.logger.Error("no input after processing", "route", routeIndex, "source", sourceId) r.logger.Error("no input after processing", "route", routeIndex, "source", sourceId)
continue return
}
outputErrors := r.HandleOutput(routeContext, routeInstance.Output(), payload)
if outputErrors != nil {
if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{}
} }
routeIOErrors = append(routeIOErrors, route.RouteIOError{
Index: routeIndex, outputErrors := r.HandleOutput(routeContext, routeInstance.Output(), payload)
OutputErrors: outputErrors, if outputErrors != nil {
}) if routeIOErrors == nil {
} routeIOErrors = []route.RouteIOError{}
}
routeIOErrors = append(routeIOErrors, route.RouteIOError{
Index: routeIndex,
OutputErrors: outputErrors,
})
}
})
} }
} }
routeWaitGroup.Wait()
return routeFound, routeIOErrors return routeFound, routeIOErrors
} }