finish ingest loop

2024-04-15 14:19:53 -06:00 · 2024-04-15 14:19:53 -06:00 · acac2a60b0
parent eef78d6e39
commit acac2a60b0
1 changed files with 3 additions and 1 deletions
--- 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()
 }