runtime.Gosched()
		stats := c.Stats()
		if stats.IncomingStreams != 0 {
			if i > 0 {
				time.Sleep(100 * time.Millisecond)
				continue
			}
			var found []uint64
			c.inStream.Range(func(key uint64, value *muxServer) bool {
				found = append(found, key)
				return true
			})
			t.Errorf("expected no active streams, got %d incoming: %v", stats.IncomingStreams, found)
		}

	KeepAlive            time.Duration  `json:"keepAliveInterval"`
	RootCAs              *x509.CertPool `json:"-"`
	QueueDir             string         `json:"queueDir"`
	QueueLimit           uint64         `json:"queueLimit"`
}

// Validate MQTTArgs fields
func (m MQTTArgs) Validate() error {
	if !m.Enable {
		return nil
	}
	u, err := xnet.ParseURL(m.Broker.String())
	if err != nil {
		return err

	Enable             bool          `json:"enable"`
	Brokers            []xnet.Host   `json:"brokers"`
	Topic              string        `json:"topic"`
	QueueDir           string        `json:"queueDir"`
	QueueLimit         uint64        `json:"queueLimit"`
	Version            string        `json:"version"`
	BatchSize          uint32        `json:"batchSize"`
	BatchCommitTimeout time.Duration `json:"batchCommitTimeout"`
	TLS                struct {

	time.Sleep(100 * time.Millisecond)
	ps.Lock()
	if len(ps.subs) != 1 {
		t.Fatal("expected 1 subscriber")
	}
	ps.Unlock()
	close(doneCh2)
}

type maskString string

func (m maskString) Mask() uint64 {
	return 1
}

func TestPubSub(t *testing.T) {
	ps := New[Maskable, Maskable](1)
	ch1 := make(chan Maskable, 1)
	doneCh1 := make(chan struct{})
	defer close(doneCh1)

type IAMSys struct {
	// Need to keep them here to keep alignment - ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
	// metrics
	LastRefreshTimeUnixNano         uint64
	LastRefreshDurationMilliseconds uint64
	TotalRefreshSuccesses           uint64
	TotalRefreshFailures            uint64

	sync.Mutex

	iamRefreshInterval time.Duration

	LDAPConfig   xldap.Config  // only valid if usersSysType is LDAPUsers

	}
	return &KMS{
		Type:       Builtin,
		DefaultKey: keyID,
		conn: secretKey{
			keyID: keyID,
			key:   key,
		},
		latencyBuckets: defaultLatencyBuckets,
		latency:        make([]atomic.Uint64, len(defaultLatencyBuckets)),
	}, nil
}

// secretKey is a KMS implementation that derives new DEKs
// from a single key.
type secretKey struct {
	keyID string
	key   []byte
}

		return -1
	}
	// use the faster version as per siphash docs
	// https://github.com/dchest/siphash#usage
	k0, k1 := binary.LittleEndian.Uint64(id[0:8]), binary.LittleEndian.Uint64(id[8:16])
	sum64 := siphash.Hash(k0, k1, []byte(key))
	return int(sum64 % uint64(cardinality))
}

func crcHashMod(key string, cardinality int) int {
	if cardinality <= 0 {
		return -1
	}

	Password           string    `json:"password"` // default: no password
	Database           string    `json:"database"` // default: same as user
	QueueDir           string    `json:"queueDir"`
	QueueLimit         uint64    `json:"queueLimit"`
	MaxOpenConnections int       `json:"maxOpenConnections"`
}

// Validate PostgreSQLArgs fields
func (p PostgreSQLArgs) Validate() error {
	if !p.Enable {
		return nil
	}

	typedef struct
	{
		int _mp_alloc;
		int _mp_size;
		mp_limb_t *_mp_d;
	} __mpz_struct;

	typedef __mpz_struct mpz_t[1];

Cgo generates:

	type _C_int int32
	type _C_mp_limb_t uint64
	type _C___mpz_struct struct {
		_mp_alloc _C_int;
		_mp_size _C_int;
		_mp_d *_C_mp_limb_t;
	}
	type _C_mpz_t [1]_C___mpz_struct

and then replaces each occurrence of a type C.xxx with _C_xxx.

	return dataUsageInfo, nil
}

func (s *xlStorage) getDeleteAttribute() uint64 {
	attr := "user.total_deletes"
	buf, err := xattr.LGet(s.formatFile, attr)
	if err != nil {
		// We start off with '0' if we can read the attributes
		return 0
	}
	return binary.LittleEndian.Uint64(buf[:8])
}

func (s *xlStorage) getWriteAttribute() uint64 {
	attr := "user.total_writes"
	buf, err := xattr.LGet(s.formatFile, attr)