return err
		}
		for e := range mCh {
			select {
			case <-ctx.Done():
				return err
			case mrfCh <- e:
			}
		}
		return nil
	}(mrfCh)
	go func(wg *sync.WaitGroup) {
		wg.Wait()
		xioutil.SafeClose(mrfCh)
	}(&wg)
	return mrfCh, nil

	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
	defer keepAliveTicker.Stop()

	mrfCh, err := globalNotificationSys.GetReplicationMRF(ctx, bucket, node)
	if err != nil {
		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
		return
	}
	enc := json.NewEncoder(w)
	for {
		select {
		case entry, ok := <-mrfCh:
			if !ok {
				return
			}
			if err := enc.Encode(entry); err != nil {

	mrfRec, err := p.loadMRF()
	if err != nil {
		return nil, err
	}

	mrfCh := make(chan madmin.ReplicationMRF, 100)
	go func() {
		defer xioutil.SafeClose(mrfCh)
		for vID, e := range mrfRec.Entries {
			if bucket != "" && e.Bucket != bucket {
				continue
			}
			select {
			case mrfCh <- madmin.ReplicationMRF{
				NodeName:   globalLocalNodeName,
				Object:     e.Object,