Workflow management

[Koeng101]

I cannot create two workers for something like "filter data" because the channel can be closed while another worker is processing data.

In reality, it should get passed in a workgroup, something like this:

func FilterData[Data DataTypes](wg *sync.WaitGroup, ctx context.Context, input <-chan Data, output chan<- Data, filter func(a Data) bool) error {
    defer wg.Done()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()

        case data, ok := <-input:
            if !ok {
                return nil // Input channel closed
            }
            if filter(data) {
                output <- data
            }
        }
    }
}

// Usage
var wg sync.WaitGroup

for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go FilterData(&wg, ctx, input, output, filterFunc)
}

// Wait for all workers to finish
wg.Wait()
close(output)

Gotta think of a good way to do this.

[Koeng101]

I think this has been solved in a more recent update. closing