res := make(map[string]string, 8)
		defer func() {
			stopFn(res)
		}()

		doneSz := globalScannerMetrics.timeSize(scannerMetricReadMetadata)
		buf, err := s.readMetadata(ctx, item.Path)
		doneSz(len(buf))
		res["metasize"] = strconv.Itoa(len(buf))
		if err != nil {
			res["err"] = err.Error()
			return sizeSummary{}, errSkipFile
		}

	rc   io.ReadCloser
	done chan struct{}
}

func (c *closeNotifier) Read(p []byte) (n int, err error) {
	n, err = c.rc.Read(p)
	if err != nil {
		if c.done != nil {
			xioutil.SafeClose(c.done)
			c.done = nil
		}
	}
	return n, err
}

func (c *closeNotifier) Close() error {
	if c.done != nil {
		xioutil.SafeClose(c.done)
		c.done = nil
	}
	return c.rc.Close()
}

		}

		if !done {
			// Drain channel.
			for range s.responses {
			}
		}
	}()
	doneCh := s.ctx.Done()
	for {
		select {
		case <-doneCh:
			if err := context.Cause(s.ctx); !errors.Is(err, errStreamEOF) {
				return err
			}
			// Fall through to be sure we have returned all responses.
			doneCh = nil
		case resp, ok := <-s.responses:
			if !ok {
				done = true
				return nil
			}

func (writer *messageWriter) Finish(bytesScanned, bytesProcessed int64) error {
	select {
	case <-writer.doneCh:
		return fmt.Errorf("messageWriter is done")
	default:
		writer.finBytesScanned = bytesScanned
		writer.finBytesProcessed = bytesProcessed
		close(writer.payloadCh)
		// Wait until the `start` go-routine is done.
		<-writer.doneCh
		return nil
	}
}

		enc := json.NewEncoder(&buf)
		for {
			select {
			case <-doneCh:
				return
			case v, ok := <-subChT:
				if !ok {
					return
				}
				buf.Reset()
				err := enc.Encode(v)
				if err != nil {
					return
				}

				select {
				case subCh <- append(GetByteBuffer()[:0], buf.Bytes()...):
					continue
				case <-doneCh:
					return
				}
			}
		}
	}()

		return ps.Participating && ps.Info.Status == rebalStarted
	}
	return false
}

func (z *erasureServerPools) rebalanceBuckets(ctx context.Context, poolIdx int) (err error) {
	doneCh := make(chan error, 1)
	defer xioutil.SafeClose(doneCh)

	// Save rebalance.bin periodically.
	go func() {
		// Update rebalance.bin periodically once every 5-10s, chosen randomly

func StreamItems[I any](store Store[I], target Target, doneCh <-chan struct{}, logger logger) {
	go func() {
		keyCh := replayItems(store, doneCh, logger, target.Name())
		sendItems(target, keyCh, doneCh, logger)
	}()

		size := actualSz
		if i.heal.enabled {
			done := globalScannerMetrics.time(scannerMetricHealCheck)
			size = i.applyHealing(ctx, objAPI, oi)
			done()

			if healDeleteDangling {
				done := globalScannerMetrics.time(scannerMetricCleanAbandoned)
				err := objAPI.CheckAbandonedParts(ctx, i.bucket, i.objectPath(), madmin.HealOpts{Remove: healDeleteDangling})
				done()
				if err != nil {

	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 err := ps.Subscribe(MaskAll, ch2, doneCh, nil); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if err := ps.Subscribe(MaskAll, ch3, doneCh, nil); err == nil {
		t.Fatalf("unexpected nil err")
	}
}

		for i := range last {
			entry := lastN[(cnt+i)%last]
			if (entry == log.Info{}) {
				continue
			}
			select {
			case subCh <- entry:
			case <-doneCh:
				return nil
			}
		}
	}
	return sys.pubsub.Subscribe(madmin.LogMaskAll, subCh, doneCh, filter)
}

// Init if HTTPConsoleLoggerSys is valid, always returns nil right now
func (sys *HTTPConsoleLoggerSys) Init(_ context.Context) error {
	return nil