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

rework Route struct

Browse Source
This commit is contained in:
Joel Wetzell
2026-02-16 19:13:04 -06:00
parent 8c3f93b601
commit abd37439c5
2 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
internal/route/route.go
View File

@@ -35,19 +35,13 @@ type RouteIO interface {
HandleOutput(ctx context.Context, destinationId string, payload any) error HandleOutput(ctx context.Context, destinationId string, payload any) error
} }
type Route interface { type Route struct {
Input() string
Output() string
ProcessPayload(ctx context.Context, payload any) (any, error)
}
type ProcessorRoute struct {
input string input string
processors []processor.Processor processors []processor.Processor
output string output string
} }
func NewRoute(config config.RouteConfig) (Route, error) { func NewRoute(config config.RouteConfig) (*Route, error) {
processors := []processor.Processor{} processors := []processor.Processor{}
if len(config.Processors) > 0 { if len(config.Processors) > 0 {
@@ -65,18 +59,18 @@ func NewRoute(config config.RouteConfig) (Route, error) {
} }
} }
return &ProcessorRoute{input: config.Input, processors: processors, output: config.Output}, nil return &Route{input: config.Input, processors: processors, output: config.Output}, nil
} }
func (r *ProcessorRoute) Input() string { func (r *Route) Input() string {
return r.input return r.input
} }
func (r *ProcessorRoute) Output() string { func (r *Route) Output() string {
return r.output return r.output
} }
func (r *ProcessorRoute) ProcessPayload(ctx context.Context, payload any) (any, error) { func (r *Route) ProcessPayload(ctx context.Context, payload any) (any, error) {
tracer := otel.Tracer("route") tracer := otel.Tracer("route")
processCtx, processSpan := tracer.Start(ctx, "ProcessPayload") processCtx, processSpan := tracer.Start(ctx, "ProcessPayload")
defer processSpan.End() defer processSpan.End()

8
router.go
View File

@@ -21,7 +21,7 @@ type Router struct {
Context context.Context Context context.Context
ModuleInstances map[string]module.Module ModuleInstances map[string]module.Module
// TODO(jwetzell): change to something easier to lookup // TODO(jwetzell): change to something easier to lookup
RouteInstances []route.Route RouteInstances []*route.Route
moduleWait sync.WaitGroup moduleWait sync.WaitGroup
logger *slog.Logger logger *slog.Logger
runningConfig config.Config runningConfig config.Config
@@ -105,7 +105,7 @@ func NewRouter(config config.Config) (*Router, []module.ModuleError, []route.Rou
router := Router{ router := Router{
ModuleInstances: make(map[string]module.Module), ModuleInstances: make(map[string]module.Module),
RouteInstances: []route.Route{}, RouteInstances: []*route.Route{},
logger: slog.Default().With("component", "router"), logger: slog.Default().With("component", "router"),
runningConfig: config, runningConfig: config,
} }
@@ -180,6 +180,10 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
var routeWaitGroup sync.WaitGroup var routeWaitGroup sync.WaitGroup
for routeIndex, routeInstance := range r.RouteInstances { for routeIndex, routeInstance := range r.RouteInstances {
if routeInstance == nil {
r.logger.Error("nil route instance found", "routeIndex", routeIndex)
continue
}
if routeInstance.Input() == sourceId { if routeInstance.Input() == sourceId {
routeWaitGroup.Go(func() { routeWaitGroup.Go(func() {