func (bfs *buffers) appendProgressBuffer(key string, p chan int64) error {
	bfs.Lock()
	defer bfs.Unlock()
	val, ok := bfs.bufferMap[key]
	if !ok {
		return fmt.Errorf("buffer not found")
	}
	val.progress = append(val.progress, p)
	return nil
}
append progress buffer
package pipeline
import (
	"context"
	"fmt"
	"sync"
)
type buffer struct {
	in chan string
	out      []chan string
	progress []chan int64
}
func (b *buffer) drainBuffer(ctx context.Context) {
loop:
	for l := range b.in {
		for _, o := range b.out {
			select {
			case <-ctx.Done():
				break loop
			case o <- l:
			default:
				<-o
				o <- l
			}
		}
	}
}
type buffers struct {
	bufferMap map[string]*buffer
	sync.RWMutex
}
func (bfs *buffers) get(key string) (*buffer, bool) {
	bfs.Lock()
	defer bfs.Unlock()
	val, ok := bfs.bufferMap[key]
	return val, ok
}
func (bfs *buffers) set(key string, value *buffer) {
	bfs.Lock()
	defer bfs.Unlock()
	bfs.bufferMap[key] = value
}
func (bfs *buffers) remove(key string) {
	bfs.Lock()
	defer bfs.Unlock()
	delete(bfs.bufferMap, key)
}
func (bfs *buffers) appendOutBuffer(key string, o chan string) error {
	bfs.Lock()
	defer bfs.Unlock()
	val, ok := bfs.bufferMap[key]
	if !ok {
		return fmt.Errorf("buffer not found")
	}
	val.out = append(val.out, o)
	return nil
}