send_to_worker_routines_with_cancel.go

package main

import (
	"context"
	"os"
	"os/signal"
	"sync"
)


func mapEachWithWorkers[In any, Out any](ctx context.Context, workers int, in <-chan In, out chan<- Out, fn func(context.Context, In) (Out, bool)) {
	wg := sync.WaitGroup{}
	for i := 0; i < workers; i++ {
		wg.Add(1)

		go func() {
			defer wg.Done()
			for input := range in {
				output, ok := fn(ctx, input)
				if !ok {
					continue
				}
				out <- output
			}
		}()
	}
	go func() {
		wg.Wait()
		close(out)
	}()
}

func sendElementsAndCloseOnContextCancelOrNotifyInterrupt[T any](ctx context.Context, objects []T, in chan<- T, sigs ...os.Signal) {
	exit := make(chan os.Signal, 1)
	signal.Notify(exit, sigs...)
	defer close(in)
	for _, object := range objects {
		select {
		case <-ctx.Done():
			return
		case <-exit:
			return
		default:
			in <- object
		}
	}
}