Add emitter to generic utils (#993)

.changeset/healthy-toes-confess.md

@@ -0,0 +1,4 @@
+---
+'@segment/analytics-core': minor
+---
+Consume Emitter module from `@segment/analytics-generic-utils`

.changeset/stale-seals-impress.md

@@ -0,0 +1,5 @@
+---
+'@segment/analytics-generic-utils': minor
+---
+
+Add Emitter library. Log default warning if a listeners exceeds 10 for a specific event type (configurable)

packages/browser/package.json

@@ -44,12 +44,13 @@
   "size-limit": [
     {
       "path": "dist/umd/index.js",
-      "limit": "29.1 KB"
+      "limit": "29.2 KB"
     }
   ],
   "dependencies": {
     "@lukeed/uuid": "^2.0.0",
     "@segment/analytics-core": "1.3.2",
+    "@segment/analytics-generic-utils": "1.0.0",
     "@segment/analytics.js-video-plugins": "^0.2.1",
     "@segment/facade": "^3.4.9",
     "@segment/tsub": "^2.0.0",

packages/browser/src/browser/index.ts

@@ -8,7 +8,7 @@ import { Plan } from '../core/events'
 import { Plugin } from '../core/plugin'
 import { MetricsOptions } from '../core/stats/remote-metrics'
 import { mergedOptions } from '../lib/merged-options'
-import { createDeferred } from '../lib/create-deferred'
+import { createDeferred } from '@segment/analytics-generic-utils'
 import { envEnrichment } from '../plugins/env-enrichment'
 import {
   PluginFactory,

packages/browser/src/core/analytics/index.ts

@@ -13,7 +13,8 @@ import {
 import type { FormArgs, LinkArgs } from '../auto-track'
 import { isOffline } from '../connection'
 import { Context } from '../context'
-import { dispatch, Emitter } from '@segment/analytics-core'
+import { dispatch } from '@segment/analytics-core'
+import { Emitter } from '@segment/analytics-generic-utils'
 import {
   Callback,
   EventFactory,

packages/browser/src/plugins/ajs-destination/types.ts

@@ -1,6 +1,6 @@
 import { Group, Identify, Track, Page, Alias } from '@segment/facade'
 import { Analytics } from '../../core/analytics'
-import { Emitter } from '@segment/analytics-core'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { User } from '../../core/user'
 
 export interface LegacyIntegration extends Emitter {

packages/core/package.json

@@ -34,6 +34,7 @@
   "packageManager": "[email protected]",
   "dependencies": {
     "@lukeed/uuid": "^2.0.0",
+    "@segment/analytics-generic-utils": "1.0.0",
     "dset": "^3.1.2",
     "tslib": "^2.4.1"
   }

packages/core/src/analytics/__tests__/dispatch.test.ts

@@ -16,7 +16,7 @@ jest.mock('../../callback', () => ({
 }))
 
 import { CoreEventQueue } from '../../queue/event-queue'
-import { Emitter } from '../../emitter'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { dispatch, getDelay } from '../dispatch'
 import { CoreContext } from '../../context'
 import { TestCtx, TestEventQueue } from '../../../test-helpers'

packages/core/src/analytics/dispatch.ts

@@ -2,7 +2,7 @@ import { CoreContext } from '../context'
 import { Callback } from '../events/interfaces'
 import { CoreEventQueue } from '../queue/event-queue'
 import { invokeCallback } from '../callback'
-import { Emitter } from '../emitter'
+import { Emitter } from '@segment/analytics-generic-utils'
 
 export type DispatchOptions<Ctx extends CoreContext = CoreContext> = {
   timeout?: number

packages/core/src/index.ts

@@ -1,4 +1,3 @@
-export * from './emitter'
 export * from './emitter/interface'
 export * from './plugins'
 export * from './events/interfaces'

packages/core/src/priority-queue/index.ts

@@ -1,4 +1,4 @@
-import { Emitter } from '../emitter'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { backoff } from './backoff'
 
 /**

packages/core/src/queue/event-queue.ts

@@ -3,7 +3,7 @@ import { groupBy } from '../utils/group-by'
 import { ON_REMOVE_FROM_FUTURE, PriorityQueue } from '../priority-queue'
 
 import { CoreContext, ContextCancelation } from '../context'
-import { Emitter } from '../emitter'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { Integrations, JSONObject } from '../events/interfaces'
 import { CorePlugin } from '../plugins'
 import { createTaskGroup, TaskGroup } from '../task/task-group'

packages/core/src/emitter/__tests__/emitter.test.ts → packages/generic-utils/src/emitter/__tests__/emitter.test.ts

@@ -1,4 +1,4 @@
-import { Emitter } from '../'
+import { Emitter } from '../emitter'
 
 describe(Emitter, () => {
   it('emits events', () => {
@@ -71,4 +71,36 @@ describe(Emitter, () => {
 
     expect(fn).toHaveBeenCalledTimes(2)
   })
+
+  it('has a default max listeners of 10', () => {
+    const em = new Emitter()
+    expect(em.maxListeners).toBe(10)
+  })
+
+  it('should warn if possible memory leak', () => {
+    const fn = jest.fn()
+    const warnSpy = jest.spyOn(console, 'warn').mockImplementation(() => {})
+    const em = new Emitter({ maxListeners: 3 })
+    em.on('test', fn)
+    em.on('test', fn)
+    em.on('test', fn)
+    expect(warnSpy).not.toHaveBeenCalled()
+    // call on 4th
+    em.on('test', fn)
+    expect(warnSpy).toHaveBeenCalledTimes(1)
+    // do not call additional times
+    em.on('test', fn)
+    expect(warnSpy).toHaveBeenCalledTimes(1)
+  })
+
+  it('has no warning if listener limit is set to 0', () => {
+    const fn = jest.fn()
+    const warnSpy = jest.spyOn(console, 'warn')
+    const em = new Emitter({ maxListeners: 0 })
+    expect(em.maxListeners).toBe(0)
+    for (let i = 0; i++; i < 20) {
+      em.on('test', fn)
+    }
+    expect(warnSpy).not.toHaveBeenCalled()
+  })
 })

packages/core/src/emitter/index.ts → packages/generic-utils/src/emitter/emitter.ts

@@ -2,6 +2,13 @@ type EventName = string
 type EventFnArgs = any[]
 type EmitterContract = Record<EventName, EventFnArgs>
 
+export interface EmitterOptions {
+  /** How many event listeners for a particular event before emitting a warning (0 = disabled)
+   *  @default 10
+   **/
+  maxListeners?: number
+}
+
 /**
  * Event Emitter that takes the expected contract as a generic
  * @example
@@ -16,7 +23,32 @@ type EmitterContract = Record<EventName, EventFnArgs>
  * ```
  */
 export class Emitter<Contract extends EmitterContract = EmitterContract> {
+  maxListeners: number
+  constructor(options?: EmitterOptions) {
+    this.maxListeners = options?.maxListeners ?? 10
+  }
   private callbacks: Partial<Contract> = {}
+  private warned = false
+
+  private warnIfPossibleMemoryLeak<EventName extends keyof Contract>(
+    event: EventName
+  ) {
+    if (this.warned) {
+      return
+    }
+    if (
+      this.maxListeners &&
+      this.callbacks[event]!.length > this.maxListeners
+    ) {
+      console.warn(
+        `Event Emitter: Possible memory leak detected; ${String(
+          event
+        )} has exceeded ${this.maxListeners} listeners.`
+      )
+      this.warned = true
+    }
+  }
+
   on<EventName extends keyof Contract>(
     event: EventName,
     callback: (...args: Contract[EventName]) => void
@@ -25,6 +57,7 @@ export class Emitter<Contract extends EmitterContract = EmitterContract> {
       this.callbacks[event] = [callback] as Contract[EventName]
     } else {
       this.callbacks[event]!.push(callback)
+      this.warnIfPossibleMemoryLeak(event)
     }
     return this
   }

packages/generic-utils/src/emitter/index.ts

@@ -0,0 +1 @@
+export * from './emitter'

packages/generic-utils/src/index.ts

@@ -1 +1,2 @@
 export * from './create-deferred'
+export * from './emitter'

packages/node/src/app/emitter.ts

@@ -1,4 +1,5 @@
-import { CoreEmitterContract, Emitter } from '@segment/analytics-core'
+import type { CoreEmitterContract } from '@segment/analytics-core'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { Context } from './context'
 import type { AnalyticsSettings } from './settings'
 import { SegmentEvent } from './types'

packages/node/src/lib/abort.ts

@@ -1,7 +1,7 @@
 /**
  * use non-native event emitter for the benefit of non-node runtimes like CF workers.
  */
-import { Emitter } from '@segment/analytics-core'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { detectRuntime } from './env'
 
 /**

packages/node/src/plugins/segmentio/__tests__/methods.test.ts

@@ -3,7 +3,7 @@ import { createSuccess } from '../../../__tests__/test-helpers/factories'
 import { createConfiguredNodePlugin } from '../index'
 import { PublisherProps } from '../publisher'
 import { Context } from '../../../app/context'
-import { Emitter } from '@segment/analytics-core'
+import { Emitter } from '@segment/analytics-generic-utils'
 import {
   assertHTTPRequestOptions,
   httpClientOptionsBodyMatcher,

packages/node/src/plugins/segmentio/__tests__/publisher.test.ts

@@ -1,4 +1,4 @@
-import { Emitter } from '@segment/analytics-core'
+import { Emitter } from '@segment/analytics-generic-utils'
 import { range } from 'lodash'
 import { createConfiguredNodePlugin } from '..'
 import { Context } from '../../../app/context'

yarn.lock

@@ -3742,6 +3742,7 @@ __metadata:
   resolution: "@segment/analytics-core@workspace:packages/core"
   dependencies:
     "@lukeed/uuid": ^2.0.0
+    "@segment/analytics-generic-utils": 1.0.0
     dset: ^3.1.2
     tslib: ^2.4.1
   languageName: unknown
@@ -3761,6 +3762,7 @@ __metadata:
     "@lukeed/uuid": ^2.0.0
     "@segment/analytics-browser-actions-braze": ^1.3.0
     "@segment/analytics-core": 1.3.2
+    "@segment/analytics-generic-utils": 1.0.0
     "@segment/analytics.js-integration": ^3.3.3
     "@segment/analytics.js-integration-amplitude": ^3.3.3
     "@segment/analytics.js-video-plugins": ^0.2.1