diff --git a/ingest.go b/ingest.go index 85b6b1b..3f07f58 100644 --- a/ingest.go +++ b/ingest.go @@ -23,7 +23,7 @@ func (i Ingester) Process(ctx context.Context) error { ctx, can := context.WithCancel(ctx) defer can() - for { + for ctx.Err() == nil { reservation, read, err := i.reader.Syn(ctx) if err != nil { return err @@ -39,4 +39,6 @@ func (i Ingester) Process(ctx context.Context) error { return err } } + + return ctx.Err() }