diff --git a/app/web_hub.go b/app/web_hub.go index 65ab3cbf2e..e1fc036449 100644 --- a/app/web_hub.go +++ b/app/web_hub.go @@ -343,6 +343,9 @@ func (h *Hub) Unregister(webConn *WebConn) { func (h *Hub) Broadcast(message *model.WebSocketEvent) { if h != nil && h.broadcast != nil && message != nil { + if metrics := h.app.Metrics(); metrics != nil { + metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) + } select { case h.broadcast <- message: case <-h.didStop: @@ -433,6 +436,9 @@ func (h *Hub) Start() { } } case msg := <-h.broadcast: + if metrics := h.app.Metrics(); metrics != nil { + metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) + } candidates := connections.All() if msg.GetBroadcast().UserId != "" { candidates = connections.ForUser(msg.GetBroadcast().UserId) diff --git a/einterfaces/metrics.go b/einterfaces/metrics.go index efb819b883..f018ad0df0 100644 --- a/einterfaces/metrics.go +++ b/einterfaces/metrics.go @@ -37,6 +37,8 @@ type MetricsInterface interface { IncrementWebsocketEvent(eventType string) IncrementWebSocketBroadcast(eventType string) + IncrementWebSocketBroadcastBufferSize(hub string, amount float64) + DecrementWebSocketBroadcastBufferSize(hub string, amount float64) AddMemCacheHitCounter(cacheName string, amount float64) AddMemCacheMissCounter(cacheName string, amount float64) diff --git a/einterfaces/mocks/MetricsInterface.go b/einterfaces/mocks/MetricsInterface.go index b5cf66f44e..c4238699c7 100644 --- a/einterfaces/mocks/MetricsInterface.go +++ b/einterfaces/mocks/MetricsInterface.go @@ -21,6 +21,11 @@ func (_m *MetricsInterface) AddMemCacheMissCounter(cacheName string, amount floa _m.Called(cacheName, amount) } +// DecrementWebSocketBroadcastBufferSize provides a mock function with given fields: hub, amount +func (_m *MetricsInterface) DecrementWebSocketBroadcastBufferSize(hub string, amount float64) { + _m.Called(hub, amount) +} + // IncrementChannelIndexCounter provides a mock function with given fields: func (_m *MetricsInterface) IncrementChannelIndexCounter() { _m.Called() @@ -141,6 +146,11 @@ func (_m *MetricsInterface) IncrementWebSocketBroadcast(eventType string) { _m.Called(eventType) } +// IncrementWebSocketBroadcastBufferSize provides a mock function with given fields: hub, amount +func (_m *MetricsInterface) IncrementWebSocketBroadcastBufferSize(hub string, amount float64) { + _m.Called(hub, amount) +} + // IncrementWebhookPost provides a mock function with given fields: func (_m *MetricsInterface) IncrementWebhookPost() { _m.Called()