| 1381 | } |
| 1382 | |
| 1383 | func (c *streamStateCollector) Collect(ch chan<- prometheus.Metric) { |
| 1384 | var active, totalEvents, maxBufLen, totalSubs int |
| 1385 | c.server.chatStreams.Range(func(_, v any) bool { |
| 1386 | state, ok := v.(*chatStreamState) |
| 1387 | if !ok { |
| 1388 | return true |
| 1389 | } |
| 1390 | active++ |
| 1391 | state.mu.Lock() |
| 1392 | bufLen := len(state.buffer) |
| 1393 | subs := len(state.subscribers) |
| 1394 | state.mu.Unlock() |
| 1395 | totalEvents += bufLen |
| 1396 | totalSubs += subs |
| 1397 | maxBufLen = max(maxBufLen, bufLen) |
| 1398 | return true |
| 1399 | }) |
| 1400 | ch <- prometheus.MustNewConstMetric(streamsActiveDesc, prometheus.GaugeValue, float64(active)) |
| 1401 | ch <- prometheus.MustNewConstMetric(streamBufferSizeMaxDesc, prometheus.GaugeValue, float64(maxBufLen)) |
| 1402 | ch <- prometheus.MustNewConstMetric(streamBufferEventsDesc, prometheus.GaugeValue, float64(totalEvents)) |
| 1403 | ch <- prometheus.MustNewConstMetric(streamSubscribersDesc, prometheus.GaugeValue, float64(totalSubs)) |
| 1404 | } |
| 1405 | |
| 1406 | // MaxQueueSize is the maximum number of queued user messages per chat. |
| 1407 | const MaxQueueSize = 20 |