}
		}
		atomic.StoreUint64(&ps.types, uint64(remainTypes))
		atomic.AddInt32(&ps.numSubscribers, -1)
	}()

	return nil
}

// SubscribeJSON - Adds a subscriber to pubsub system and returns results with JSON encoding.
func (ps *PubSub[T, M]) SubscribeJSON(mask M, subCh chan<- []byte, doneCh <-chan struct{}, filter func(entry T) bool, wg *sync.WaitGroup) error {
	totalSubs := atomic.AddInt32(&ps.numSubscribers, 1)

		duration := time.Since(startTime)

		atomic.AddUint64(&p.operations[s], 1)
		if s < scannerMetricLastRealtime {
			p.latency[s].addSize(duration, int64(sz))
		}
	}
}

// incTime will increment time on metric s with a specific duration.
// Use for s < scannerMetricLastRealtime
func (p *scannerMetrics) incTime(s scannerMetric, d time.Duration) {
	atomic.AddUint64(&p.operations[s], 1)

			w.WriteHeader(http.StatusServiceUnavailable)
			w.Write([]byte(http.ErrServerClosed.Error()))
			return
		}

		atomic.AddInt32(&srv.requestCount, 1)
		defer atomic.AddInt32(&srv.requestCount, -1)

		// Handle request using passed handler.
		handler.ServeHTTP(w, r)
	})

	srv.listenerMutex.Lock()
	srv.Handler = wrappedHandler
	srv.listener = listener
	srv.listenerMutex.Unlock()

		ID:             target.id,
		ActiveRequests: atomic.LoadInt64(&target.activeRequests),
		TotalRequests:  atomic.LoadInt64(&target.totalRequests),
		FailedRequests: atomic.LoadInt64(&target.failedRequests),
	}
}

// Send - sends an event to the webhook.
func (target *WebhookTarget) Send(eventData event.Event) (resp *http.Response, err error) {
	atomic.AddInt64(&target.activeRequests, 1)

func (s *scanner) remaining() int64 {
	if atomic.LoadInt64(&s.end) == maxInt {
		return maxInt
	}
	return atomic.LoadInt64(&s.end) - s.pos
}

// read byte at current position (without advancing)
func (s *scanner) cur() byte { return s.buf[s.ipos] }

// read next byte
func (s *scanner) next() byte {
	if s.pos >= atomic.LoadInt64(&s.end) {
		return nullByte
	}
	s.ipos++

func GetRPCStats() RPCStats {
	s := RPCStats{
		Errs:     atomic.LoadUint64(&globalStats.errs),
		DialErrs: atomic.LoadUint64(&globalStats.tcpDialErrs),
	}
	if v := atomic.LoadUint64(&globalStats.tcpDialCount); v > 0 {
		s.DialAvgDuration = atomic.LoadUint64(&globalStats.tcpDialTotalDur) / v
		s.TTFBAvgDuration = atomic.LoadUint64(&globalStats.tcpTimeForFirstByteTotalDur) / v
	}
	return s
}

	lastSyncTime := atomic.LoadUint64(&globalIAMSys.LastRefreshTimeUnixNano)
	if lastSyncTime != 0 {
		m.Set(sinceLastSyncMillis, float64((uint64(time.Now().UnixNano())-lastSyncTime)/uint64(time.Millisecond)))
	}
	m.Set(syncFailures, float64(atomic.LoadUint64(&globalIAMSys.TotalRefreshFailures)))
	m.Set(syncSuccesses, float64(atomic.LoadUint64(&globalIAMSys.TotalRefreshSuccesses)))
	return nil

			n, err := rr.ReadAt(p.buf[bufIdx], p.offset)
			if err != nil {
				switch {
				case errors.Is(err, errFileNotFound):
					atomic.StoreInt32(&missingPartsHeal, 1)
				case errors.Is(err, errFileCorrupt):
					atomic.StoreInt32(&bitrotHeal, 1)
				case errors.Is(err, errDiskNotFound):
					atomic.AddInt32(&disksNotFound, 1)
				}

				// This will be communicated upstream.
				p.orgReaders[bufIdx] = nil

	r      io.Reader
}

func (s *safeCloser) Read(p []byte) (n int, err error) {
	if atomic.LoadUint32(&s.closed) == 1 {
		return 0, io.EOF
	}
	n, err = s.r.Read(p)
	if atomic.LoadUint32(&s.closed) == 1 {
		return 0, io.EOF
	}
	return n, err
}

func (s *safeCloser) Close() error {
	atomic.CompareAndSwapUint32(&s.closed, 0, 1)
	return nil

	for range numIterations {
		if rwm.GetRLock(context.Background(), nil, id, source, Options{Timeout: time.Second}) {
			n := atomic.AddInt32(activity, 1)
			if n < 1 || n >= 10000 {
				panic(fmt.Sprintf("wlock(%d)\n", n))
			}
			for range 100 {
			}
			atomic.AddInt32(activity, -1)
			rwm.RUnlock(context.Background())
		}
	}
	cdone <- true
}

// Borrowed from rwmutex_test.go