)

// Reader - JSON record reader for S3Select.
type Reader struct {
	args    *json.ReaderArgs
	input   chan simdjson.Stream
	decoded chan simdjson.Object

	// err will only be returned after decoded has been closed.
	err          *error
	readCloser   io.ReadCloser
	onReaderExit func()

	exitReader chan struct{}
	readerWg   sync.WaitGroup
}

// Read - reads single record.

			}
			consolidatedReport.BucketStats[opts] = d
		}
	}
	return consolidatedReport
}

func (sys *NotificationSys) collectPeerMetrics(ctx context.Context, peerChannels []<-chan MetricV2, g *errgroup.Group) <-chan MetricV2 {
	ch := make(chan MetricV2)
	var wg sync.WaitGroup
	for index, err := range g.Wait() {
		if err != nil {
			if sys.peerClients[index] != nil {

		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)
	}

		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)
	}

	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
	kvDstPool   sync.Pool       // pool of []jstream.KV used for output
	close       chan struct{}   // used for shutting down the splitter before end of stream

	if m.GetRLock(ctx, "", "", time.Second) {
		clocked <- true
		<-cunlock
		m.RUnlock()
		cdone <- true
	}
}

// Borrowed from rwmutex_test.go
func doTestParallelReaders(numReaders, gomaxprocs int) {
	runtime.GOMAXPROCS(gomaxprocs)
	m := NewLRWMutex()

	clocked := make(chan bool)
	cunlock := make(chan bool)
	cdone := make(chan bool)

	// has never been replaced.
	Healing() *healingTracker
	DiskInfo(ctx context.Context, opts DiskInfoOptions) (info DiskInfo, err error)
	NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, shouldSleep func() bool) (dataUsageCache, error)

	// Volume operations.
	MakeVol(ctx context.Context, volume string) (err error)

// Updates are sent on a regular basis and the caller *must* consume them.
func (er erasureObjects) nsScanner(ctx context.Context, buckets []BucketInfo, wantCycle uint32, updates chan<- dataUsageCache, healScanMode madmin.HealScanMode) error {
	if len(buckets) == 0 {
		return nil
	}

	// Collect disks we can use.
	disks, healing := er.getOnlineDisksWithHealing(false)
	if len(disks) == 0 {

	defer n.Unlock()
	n.RX = 0
	n.RXSample = 0
	n.lastToConnect = time.Time{}
	n.firstToDisconnect = time.Time{}
}

// Reader to read random data.
type netperfReader struct {
	n   uint64
	eof chan struct{}
	buf []byte
}

func (m *netperfReader) Read(b []byte) (int, error) {
	select {
	case <-m.eof:
		return 0, io.EOF
	default:
	}
	n := copy(b, m.buf)
	atomic.AddUint64(&m.n, uint64(n))

	}()

	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 {