JetStream       struct {
		Enable bool `json:"enable"`
	} `json:"jetStream"`
	Streaming struct {
		Enable             bool   `json:"enable"`
		ClusterID          string `json:"clusterID"`
		Async              bool   `json:"async"`
		MaxPubAcksInflight int    `json:"maxPubAcksInflight"`
	} `json:"streaming"`

	RootCAs *x509.CertPool `json:"-"`
}

// Validate NATSArgs fields
func (n NATSArgs) Validate() error {

	notificationEventsSkippedTotal    = "events_skipped_total"
)

var (
	notificationCurrentSendInProgressMD = NewCounterMD(notificationCurrentSendInProgress, "Number of concurrent async Send calls active to all targets")
	notificationEventsErrorsTotalMD     = NewCounterMD(notificationEventsErrorsTotal, "Events that were failed to be sent to the targets")

		r = bf
	default:
		return fmt.Errorf("Unsupported format %s", f)
	}
	tarReader := tar.NewReader(r)
	n := 0
	asyncWriters := make(chan struct{}, 16)
	var wg sync.WaitGroup

	var asyncErr error
	var asyncErrMu sync.Mutex
	for {
		if !o.ignoreErrs {
			asyncErrMu.Lock()
			err := asyncErr
			asyncErrMu.Unlock()
			if err != nil {
				return err
			}
		}

	eventsSkipped     atomic.Int64
	eventsErrorsTotal atomic.Int64

	sync.RWMutex
	targets map[TargetID]Target
	queue   chan asyncEvent
	ctx     context.Context

	statLock    sync.RWMutex
	targetStats map[TargetID]targetStat
}

type targetStat struct {
	// The number of concurrent async Send calls per targets
	currentSendCalls int64
	// The number of total events per target
	totalEvents int64

					Subsystem: iamSubsystem,
					Name:      "since_last_sync_millis",
					Help:      "Time (in milliseconds) since last successful IAM data sync.",
					Type:      gaugeMetric,
				},
				Value: float64(sinceLastSyncMillis),
			},
			{
				Description: MetricDescription{
					Namespace: nodeMetricNamespace,
					Subsystem: iamSubsystem,
					Name:      "sync_successes",

	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
	close        chan struct{}    // used for shutting down the splitter before end of stream
	readerWg     sync.WaitGroup   // used to keep track of async reader.
}

// queueItem is an item in the queue.
type queueItem struct {

	clnts := newBootstrapRESTClients(endpointServerPools, gm)
	var onlineServers int
	var offlineEndpoints []error
	var incorrectConfigs []error
	var retries int
	var mu sync.Mutex
	for onlineServers < len(clnts)/2 {
		var wg sync.WaitGroup
		wg.Add(len(clnts))
		onlineServers = 0
		for _, clnt := range clnts {
			go func(clnt *bootstrapRESTClient) {
				defer wg.Done()

// The result channel is closed when no more results are expected.
func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) error {
	var mu sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	allAtEOF := true
	var inputs []chan metaCacheEntry
	mu.Lock()
	// Ask all sets and merge entries.
	listCtx, cancelList := context.WithCancel(ctx)
	defer cancelList()

// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package grid

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	xioutil "github.com/minio/minio/internal/ioutil"
	"github.com/zeebo/xxh3"
)

// muxClient is a stateful connection to a remote.
type muxClient struct {
	MuxID            uint64

_verify","value":"off"},{"key":"cert_authority","value":""},{"key":"client_cert","value":""},{"key":"client_key","value":""},{"key":"ping_interval","value":"0"},{"key":"jetstream","value":"off"},{"key":"streaming","value":"off"},{"key":"streaming_async","value":"off"},{"key":"streaming_max_pub_acks_in_flight","value":"0"},{"key":"streaming_cluster_id","value":""},{"key":"queue_dir","value":""},{"key":"queue_limit","value":"0"}]},"notify_nsq":{"_":[{"key":"enable","value":"off"},{"key":"nsqd_addr...