adapters: always loop until explicitly closed

Author: lukasmalkmus

adapters/apex/apex.go

@@ -71,6 +71,7 @@ type Handler struct {
 	ingestOptions []ingest.Option
 
 	eventCh   chan axiom.Event
+	stopCh    chan struct{}
 	closeCh   chan struct{}
 	closeOnce sync.Once
 }
@@ -93,6 +94,7 @@ type Handler struct {
 func New(options ...Option) (*Handler, error) {
 	handler := &Handler{
 		eventCh: make(chan axiom.Event, defaultBatchSize),
+		stopCh:  make(chan struct{}),
 		closeCh: make(chan struct{}),
 	}
 
@@ -127,13 +129,20 @@ func New(options ...Option) (*Handler, error) {
 
 		logger := stdlog.New(os.Stderr, "[AXIOM|APEX]", 0)
 
-		res, err := handler.client.IngestChannel(context.Background(), handler.datasetName, handler.eventCh, handler.ingestOptions...)
-		if err != nil {
-			logger.Printf("failed to ingest events: %s\n", err)
-		} else if res.Failed > 0 {
-			// Best effort on notifying the user about the ingest failure.
-			logger.Printf("event at %s failed to ingest: %s\n",
-				res.Failures[0].Timestamp, res.Failures[0].Error)
+		for {
+			if res, err := handler.client.IngestChannel(context.Background(), handler.datasetName, handler.eventCh, handler.ingestOptions...); err != nil {
+				logger.Printf("failed to ingest events: %s\n", err)
+			} else if res.Failed > 0 {
+				// Best effort on notifying the user about the ingest failure.
+				logger.Printf("event at %s failed to ingest: %s\n",
+					res.Failures[0].Timestamp, res.Failures[0].Error)
+			}
+
+			select {
+			case <-handler.stopCh:
+				return
+			default:
+			}
 		}
 	}()
 
@@ -144,6 +153,7 @@ func New(options ...Option) (*Handler, error) {
 // renders it unusable for further use.
 func (h *Handler) Close() {
 	h.closeOnce.Do(func() {
+		close(h.stopCh)
 		close(h.eventCh)
 		<-h.closeCh
 	})

adapters/logrus/logrus.go

@@ -81,6 +81,7 @@ type Hook struct {
 	levels        []logrus.Level
 
 	eventCh   chan axiom.Event
+	stopCh    chan struct{}
 	closeCh   chan struct{}
 	closeOnce sync.Once
 }
@@ -105,6 +106,7 @@ func New(options ...Option) (*Hook, error) {
 		levels: logrus.AllLevels,
 
 		eventCh: make(chan axiom.Event, defaultBatchSize),
+		stopCh:  make(chan struct{}),
 		closeCh: make(chan struct{}),
 	}
 
@@ -139,13 +141,20 @@ func New(options ...Option) (*Hook, error) {
 
 		logger := log.New(os.Stderr, "[AXIOM|LOGRUS]", 0)
 
-		res, err := hook.client.IngestChannel(context.Background(), hook.datasetName, hook.eventCh, hook.ingestOptions...)
-		if err != nil {
-			logger.Printf("failed to ingest events: %s\n", err)
-		} else if res.Failed > 0 {
-			// Best effort on notifying the user about the ingest failure.
-			logger.Printf("event at %s failed to ingest: %s\n",
-				res.Failures[0].Timestamp, res.Failures[0].Error)
+		for {
+			if res, err := hook.client.IngestChannel(context.Background(), hook.datasetName, hook.eventCh, hook.ingestOptions...); err != nil {
+				logger.Printf("failed to ingest events: %s\n", err)
+			} else if res.Failed > 0 {
+				// Best effort on notifying the user about the ingest failure.
+				logger.Printf("event at %s failed to ingest: %s\n",
+					res.Failures[0].Timestamp, res.Failures[0].Error)
+			}
+
+			select {
+			case <-hook.stopCh:
+				return
+			default:
+			}
 		}
 	}()
 
@@ -157,6 +166,7 @@ func New(options ...Option) (*Hook, error) {
 // unusable for further use.
 func (h *Hook) Close() {
 	h.closeOnce.Do(func() {
+		close(h.stopCh)
 		close(h.eventCh)
 		<-h.closeCh
 	})

adapters/slog/slog.go

@@ -77,6 +77,7 @@ type rootHandler struct {
 	ingestOptions []ingest.Option
 
 	eventCh   chan axiom.Event
+	stopCh    chan struct{}
 	closeCh   chan struct{}
 	closeOnce sync.Once
 }
@@ -108,6 +109,7 @@ type Handler struct {
 func New(options ...Option) (*Handler, error) {
 	root := &rootHandler{
 		eventCh: make(chan axiom.Event, defaultBatchSize),
+		stopCh:  make(chan struct{}),
 		closeCh: make(chan struct{}),
 	}
 
@@ -146,13 +148,20 @@ func New(options ...Option) (*Handler, error) {
 
 		logger := log.New(os.Stderr, "[AXIOM|SLOG]", 0)
 
-		res, err := root.client.IngestChannel(context.Background(), root.datasetName, root.eventCh, root.ingestOptions...)
-		if err != nil {
-			logger.Printf("failed to ingest events: %s\n", err)
-		} else if res.Failed > 0 {
-			// Best effort on notifying the user about the ingest failure.
-			logger.Printf("event at %s failed to ingest: %s\n",
-				res.Failures[0].Timestamp, res.Failures[0].Error)
+		for {
+			if res, err := root.client.IngestChannel(context.Background(), root.datasetName, root.eventCh, root.ingestOptions...); err != nil {
+				logger.Printf("failed to ingest events: %s\n", err)
+			} else if res.Failed > 0 {
+				// Best effort on notifying the user about the ingest failure.
+				logger.Printf("event at %s failed to ingest: %s\n",
+					res.Failures[0].Timestamp, res.Failures[0].Error)
+			}
+
+			select {
+			case <-root.stopCh:
+				return
+			default:
+			}
 		}
 	}()
 
@@ -163,6 +172,7 @@ func New(options ...Option) (*Handler, error) {
 // renders it unusable for further use.
 func (h *Handler) Close() {
 	h.closeOnce.Do(func() {
+		close(h.stopCh)
 		close(h.eventCh)
 		<-h.closeCh
 	})

axiom/datasets.go

@@ -512,16 +512,15 @@ func (s *DatasetsService) IngestChannel(ctx context.Context, id string, events <
 			return fmt.Errorf("failed to ingest events: %w", err)
 		}
 		ingestStatus.Add(res)
-		t.Reset(flushInterval) // Reset the ticker.
 		batch = batch[:0]      // Clear the batch.
+		t.Reset(flushInterval) // Reset the ticker.
 
 		return nil
 	}
 
 	for {
 		select {
 		case <-ctx.Done():
-
 			return &ingestStatus, spanError(span, context.Cause(ctx))
 		case event, ok := <-events:
 			if !ok {