// for this node or peers
func (sys *HTTPConsoleLoggerSys) HasLogListeners() bool {
	return sys != nil && sys.pubsub.Subscribers() > 0
}

// Subscribe starts console logging for this node.
func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan log.Info, doneCh <-chan struct{}, node string, last int, logKind madmin.LogMask, filter func(entry log.Info) bool) error {
	// Enable console logging for remote client.

	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")
	}
}

		if sub.types.Contains(Mask(item.Mask())) && (sub.filter == nil || sub.filter(item)) {
			select {
			case sub.ch <- item:
			default:
			}
		}
	}
}

// Subscribe - Adds a subscriber to pubsub system
func (ps *PubSub[T, M]) Subscribe(mask M, subCh chan T, doneCh <-chan struct{}, filter func(entry T) bool) error {
	totalSubs := atomic.AddInt32(&ps.numSubscribers, 1)
	if ps.maxSubscribers > 0 && totalSubs > ps.maxSubscribers {

			case <-ctx.Done():
				grid.PutByteBuffer(buf.Bytes())
				return
			}
		}
	}()
	peers, _ := newPeerRestClients(globalEndpoints)
	err := globalHTTPListen.Subscribe(mask, localCh, ctx.Done(), func(ev event.Event) bool {
		if ev.S3.Bucket.Name != "" && bucketName != "" {
			if ev.S3.Bucket.Name != bucketName {
				return false
			}
		}

	// Use buffered channel to take care of burst sends or slow w.Write()
	ch := make(chan event.Event, globalAPIConfig.getRequestsPoolCapacity())
	err := globalHTTPListen.Subscribe(mask, ch, ctx.Done(), func(ev event.Event) bool {
		if ev.S3.Bucket.Name != "" && values.Get(peerRESTListenBucket) != "" {
			if ev.S3.Bucket.Name != values.Get(peerRESTListenBucket) {
				return false
			}
		}

	}, nil)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
		return
	}

	// Publish bootstrap events that have already occurred before client could subscribe.
	if traceOpts.TraceTypes().Contains(madmin.TraceBootstrap) {
		go globalBootstrapTracer.Publish(ctx, globalTrace)
	}

	for _, peer := range peers {
		if peer == nil {
			continue
		}

		return s.iptablesConfigurator.CreateInpodRules(&HostProbeSNATIP)
	}); err != nil {
		log.Errorf("failed to update POD inpod: %s/%s %v", pod.Namespace, pod.Name, err)
		return err
	}

	log.Debug("notifying subscribed node proxies")
	if err := s.sendPodToZtunnelAndWaitForAck(ctx, pod, openNetns); err != nil {
		// we must return PartialAdd error here. the pod was injected with iptables rules,