type Key struct {
	Name   string
	IsLast bool
}

// replayItems - Reads the items from the store and replays.
func replayItems[I any](store Store[I], doneCh <-chan struct{}, log logger, id string) <-chan Key {
	keyCh := make(chan Key)

	go func() {
		defer xioutil.SafeClose(keyCh)

		retryTicker := time.NewTicker(retryInterval)
		defer retryTicker.Stop()

		for {
			names, err := store.List()

		t.Fatalf("want %d subscribers, got %d", want, got)
	}
}

func TestSubscribeExceedingLimit(t *testing.T) {
	ps := New[Maskable, Maskable](2)
	ch1 := make(chan Maskable, 1)
	ch2 := make(chan Maskable, 1)
	ch3 := make(chan Maskable, 1)
	doneCh := make(chan struct{})
	defer close(doneCh)
	if err := ps.Subscribe(MaskAll, ch1, doneCh, nil); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}

			select {
			case sub.ch <- item:
			default:
			}
		}
	}
}

// Subscribe - Adds a subscriber to pubsub system
func (ps *PubSub[T, M]) Subscribe(mask M, subCh chan T, doneCh <-chan struct{}, filter func(entry T) bool) error {
	totalSubs := atomic.AddInt32(&ps.numSubscribers, 1)
	if ps.maxSubscribers > 0 && totalSubs > ps.maxSubscribers {
		atomic.AddInt32(&ps.numSubscribers, -1)

		if debugPrint {
			fmt.Println("connected stream mux:", ack.MuxID)
		}
	}()

	// Data inbound to the handler
	var handlerIn chan []byte
	if inboundCap > 0 {
		m.inbound = make(chan []byte, inboundCap)
		handlerIn = make(chan []byte, 1)
		go func(inbound chan []byte) {
			wg.Wait()
			defer xioutil.SafeClose(handlerIn)
			m.handleInbound(c, inbound, handlerIn)
		}(m.inbound)
	}

	globalServiceFreezeMu.Lock()
	// Close when we reach 0
	globalServiceFreezeCnt--
	if globalServiceFreezeCnt <= 0 {
		// Set to a nil channel.
		var _ch chan struct{}
		if val := globalServiceFreeze.Swap(_ch); val != nil {
			if ch, ok := val.(chan struct{}); ok && ch != nil {
				// Close previous non-nil channel.
				xioutil.SafeClose(ch)
			}
		}
		globalServiceFreezeCnt = 0 // Don't risk going negative.
	}

	}()

	cleanReqs := make(chan struct{})
	gotCall := make(chan struct{})
	defer close(cleanReqs)
	// 1: Block forever
	h1 := func(payload []byte) ([]byte, *RemoteErr) {
		gotCall <- struct{}{}
		<-cleanReqs
		return nil, nil
	}
	// 2: Also block, but with streaming.
	h2 := StreamHandler{
		Handle: func(ctx context.Context, payload []byte, request <-chan []byte, resp chan<- []byte) *RemoteErr {

	input        chan *queueItem  // input for workers
	queue        chan *queueItem  // output from workers in order
	err          error            // global error state, only touched by Reader.Read
	bufferPool   sync.Pool        // pool of []byte objects for input
	csvDstPool   sync.Pool        // pool of [][]string used for output

	"encoding/json"
	"fmt"
	"os"

	"github.com/fsnotify/fsnotify"

	"istio.io/istio/pkg/file"
	"istio.io/istio/pkg/log"
)

type Watcher struct {
	watcher *fsnotify.Watcher
	Events  chan struct{}
	Errors  chan error
}

// Waits until a file is modified (returns nil), the context is cancelled (returns context error), or returns error
func (w *Watcher) Wait(ctx context.Context) error {
	select {
	case <-w.Events:

	opts      madmin.HealOpts
	// Healing response will be sent here
	respCh chan healResult
}

// healResult represents a healing result with a possible error
type healResult struct {
	result madmin.HealResultItem
	err    error
}

// healRoutine receives heal tasks, to heal buckets, objects and format.json
type healRoutine struct {
	tasks   chan healTask
	workers int
}

func activeListeners() int {

type WaitGroup struct {
	count int32
	done  chan struct{}
}

func NewWaitGroup() *WaitGroup {
	return &WaitGroup{
		done: make(chan struct{}),
	}
}

func NewWaitForNCalls(t *testing.T, n int32) (*WaitGroup, func()) {
	wg := &WaitGroup{
		done: make(chan struct{}),
	}

	wg.Add(n)
	return wg, func() {
		select {
		case <-wg.C():