diff --git a/go.mod b/go.mod index a32f27b..4080887 100644 --- a/go.mod +++ b/go.mod @@ -20,6 +20,7 @@ require ( github.com/spf13/cobra v1.8.1 github.com/tcolgate/mp3 v0.0.0-20170426193717-e79c5a46d300 golang.org/x/crypto v0.21.0 + golang.org/x/sync v0.5.0 golang.org/x/term v0.18.0 google.golang.org/protobuf v1.33.0 gopkg.in/yaml.v3 v3.0.1 @@ -55,7 +56,6 @@ require ( golang.org/x/exp/shiny v0.0.0-20240719175910-8a7402abbf56 // indirect golang.org/x/image v0.14.0 // indirect golang.org/x/mobile v0.0.0-20231127183840-76ac6878050a // indirect - golang.org/x/sync v0.5.0 // indirect golang.org/x/sys v0.20.0 // indirect golang.org/x/text v0.14.0 // indirect ) diff --git a/pkg/gordio/server/ingest.go b/pkg/gordio/server/ingest.go index 26e8133..8f42950 100644 --- a/pkg/gordio/server/ingest.go +++ b/pkg/gordio/server/ingest.go @@ -6,6 +6,6 @@ import ( "dynatron.me/x/stillbox/pkg/calls" ) -func (s *Server) Ingest(ctx context.Context, call *calls.Call) { - s.sinks.EmitCall(context.Background(), call) +func (s *Server) Ingest(ctx context.Context, call *calls.Call) error { + return s.sinks.EmitCall(context.Background(), call) } diff --git a/pkg/gordio/server/server.go b/pkg/gordio/server/server.go index a2e1373..e51fa6c 100644 --- a/pkg/gordio/server/server.go +++ b/pkg/gordio/server/server.go @@ -41,8 +41,8 @@ func New(cfg *config.Config) (*Server, error) { nex: nexus.New(), } - srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db)) - srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex)) + srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db), true) + srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false) srv.sources.Register("rdio-http", sources.NewRdioHTTP(authenticator, srv)) r.Use(middleware.RequestID) diff --git a/pkg/gordio/sinks/sinks.go b/pkg/gordio/sinks/sinks.go index 0523eba..eb0a8ad 100644 --- a/pkg/gordio/sinks/sinks.go +++ b/pkg/gordio/sinks/sinks.go @@ -2,6 +2,7 @@ package sinks import ( "context" + "golang.org/x/sync/errgroup" "dynatron.me/x/stillbox/pkg/calls" @@ -16,26 +17,41 @@ type Sink interface { type sinkInstance struct { Sink Name string + + // whether call ingest should be considered failed if this sink returns error + Required bool } type Sinks []sinkInstance -func (s *Sinks) Register(name string, toAdd Sink) { +func (s *Sinks) Register(name string, toAdd Sink, required bool) { *s = append(*s, sinkInstance{ - Name: name, - Sink: toAdd, + Name: name, + Sink: toAdd, + Required: required, }) } -func (s *Sinks) EmitCall(ctx context.Context, call *calls.Call) { +func (s *Sinks) EmitCall(ctx context.Context, call *calls.Call) error { + g, ctx := errgroup.WithContext(ctx) for i := range *s { - go (*s)[i].emitCallLogErr(ctx, call) + sink := (*s)[i] + g.Go(sink.callEmitter(ctx, call)) } + + return g.Wait() } -func (sink *sinkInstance) emitCallLogErr(ctx context.Context, call *calls.Call) { - err := sink.Call(ctx, call) - if err != nil { - log.Error().Str("sink", sink.Name).Err(err).Msg("call emit to sink failed") +func (sink *sinkInstance) callEmitter(ctx context.Context, call *calls.Call) func() error { + return func() error { + err := sink.Call(ctx, call) + if err != nil { + log.Error().Str("sink", sink.Name).Err(err).Msg("call emit to sink failed") + if sink.Required { + return err + } + } + + return nil } } diff --git a/pkg/gordio/sources/http.go b/pkg/gordio/sources/http.go index 17a5383..87f0502 100644 --- a/pkg/gordio/sources/http.go +++ b/pkg/gordio/sources/http.go @@ -127,7 +127,12 @@ func (h *RdioHTTP) routeCallUpload(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } - h.ing.Ingest(ctx, call) + err = h.ing.Ingest(ctx, call) + if err != nil { + log.Error().Err(err).Msg("ingest failed") + http.Error(w, "Call ingest failed.", http.StatusInternalServerError) + return + } log.Info().Int("system", cur.System).Int("tgid", cur.Talkgroup).Msg("ingested") diff --git a/pkg/gordio/sources/source.go b/pkg/gordio/sources/source.go index fa49a22..fddc9cc 100644 --- a/pkg/gordio/sources/source.go +++ b/pkg/gordio/sources/source.go @@ -35,7 +35,7 @@ func (s *Sources) PublicRoutes(r chi.Router) { } type Ingestor interface { - Ingest(context.Context, *calls.Call) + Ingest(context.Context, *calls.Call) error } type PublicRouteSource interface {