Add iplist_size metric

events/event_stream.go

@@ -102,6 +102,8 @@ func eventStreamProcessor(ctx context.Context, eventChan <-chan mtglib.Event, ob
 				observer.EventConcurrencyLimited(typedEvt)
 			case mtglib.EventReplayAttack:
 				observer.EventReplayAttack(typedEvt)
+			case mtglib.EventIPListSize:
+				observer.EventIPListSize(typedEvt)
 			}
 		}
 	}

events/event_stream_test.go

@@ -204,6 +204,27 @@ func (suite *EventStreamTestSuite) TestEventReplayAttack() {
 	time.Sleep(100 * time.Millisecond)
 }
 
+func (suite *EventStreamTestSuite) TestEventIPListSize() {
+	evt := mtglib.NewEventIPListSize(10, true)
+
+	for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} {
+		v.
+			On("EventIPListSize", mock.Anything).
+			Once().
+			Run(func(args mock.Arguments) {
+				caught, ok := args.Get(0).(mtglib.EventIPListSize)
+
+				suite.True(ok)
+				suite.Equal(evt.Timestamp(), caught.Timestamp())
+				suite.Equal(evt.Size, caught.Size)
+				suite.Equal(evt.IsBlockList, caught.IsBlockList)
+			})
+	}
+
+	suite.stream.Send(suite.ctx, evt)
+	time.Sleep(100 * time.Millisecond)
+}
+
 func (suite *EventStreamTestSuite) TearDownTest() {
 	suite.stream.Shutdown()
 	suite.ctxCancel()

events/init.go

@@ -53,6 +53,9 @@ type Observer interface {
 	// EventReplayAttack reacts on incoming mtglib.EventReplayAttack event.
 	EventReplayAttack(mtglib.EventReplayAttack)
 
+	// EventIPListSize reacts on incoming mtglib.EventIPListSize
+	EventIPListSize(mtglib.EventIPListSize)
+
 	// Shutdown stop observer. Default event stream guarantees:
 	//   1. If shutdown is executed, it is executed only once
 	//   2. Observer won't receieve any new message after this

events/init_test.go

@@ -41,6 +41,10 @@ func (o *ObserverMock) EventReplayAttack(evt mtglib.EventReplayAttack) {
 	o.Called(evt)
 }
 
+func (o *ObserverMock) EventIPListSize(evt mtglib.EventIPListSize) {
+	o.Called(evt)
+}
+
 func (o *ObserverMock) Shutdown() {
 	o.Called()
 }

events/multi_observer.go

@@ -130,6 +130,21 @@ func (m multiObserver) EventReplayAttack(evt mtglib.EventReplayAttack) {
 	wg.Wait()
 }
 
+func (m multiObserver) EventIPListSize(evt mtglib.EventIPListSize) {
+	wg := &sync.WaitGroup{}
+	wg.Add(len(m.observers))
+
+	for _, v := range m.observers {
+		go func(obs Observer) {
+			defer wg.Done()
+
+			obs.EventIPListSize(evt)
+		}(v)
+	}
+
+	wg.Wait()
+}
+
 func (m multiObserver) Shutdown() {
 	for _, v := range m.observers {
 		v.Shutdown()

events/noop.go

@@ -25,6 +25,7 @@ func (n noopObserver) EventFinish(_ mtglib.EventFinish)
 func (n noopObserver) EventConcurrencyLimited(_ mtglib.EventConcurrencyLimited) {}
 func (n noopObserver) EventIPBlocklisted(_ mtglib.EventIPBlocklisted)           {}
 func (n noopObserver) EventReplayAttack(_ mtglib.EventReplayAttack)             {}
+func (n noopObserver) EventIPListSize(_ mtglib.EventIPListSize)                 {}
 func (n noopObserver) Shutdown()                                                {}
 
 // NewNoopObserver creates an observer which discards each message.

events/noop_test.go

@@ -27,6 +27,7 @@ func (suite *NoopTestSuite) SetupSuite() {
 		"concurrency-limited": mtglib.NewEventConcurrencyLimited(),
 		"ip-blacklisted":      mtglib.NewEventIPBlocklisted(net.ParseIP("10.0.0.10")),
 		"replay-attack":       mtglib.NewEventReplayAttack("connID"),
+		"ip-list-size":        mtglib.NewEventIPListSize(10, true),
 	}
 	suite.ctx = context.Background()
 }
@@ -65,6 +66,8 @@ func (suite *NoopTestSuite) TestObserver() {
 				observer.EventIPBlocklisted(typedEvt)
 			case mtglib.EventReplayAttack:
 				observer.EventReplayAttack(typedEvt)
+			case mtglib.EventIPListSize:
+				observer.EventIPListSize(typedEvt)
 			}
 		})
 	}