♻️ [RUMF-1015] add callback capabilities to observable (#1055)

* ♻️  add callback capabilities to observable

* ♻️ use new capabilities for domMutationObservable

* 👌 tweak API

* 👌 simplify implementation

packages/core/src/tools/observable.spec.ts

@@ -0,0 +1,70 @@
+import { Observable } from '@datadog/browser-core'
+
+describe('observable', () => {
+  let observable: Observable<void>
+  let subscriber: jasmine.Spy<jasmine.Func>
+
+  beforeEach(() => {
+    observable = new Observable()
+    subscriber = jasmine.createSpy('sub')
+  })
+
+  it('should allow to subscribe and be notified', () => {
+    observable.subscribe(subscriber)
+    expect(subscriber).not.toHaveBeenCalled()
+
+    observable.notify()
+    expect(subscriber).toHaveBeenCalledTimes(1)
+
+    observable.notify()
+    expect(subscriber).toHaveBeenCalledTimes(2)
+  })
+
+  it('should notify multiple clients', () => {
+    const otherSubscriber = jasmine.createSpy('sub2')
+    observable.subscribe(subscriber)
+    observable.subscribe(otherSubscriber)
+
+    observable.notify()
+
+    expect(subscriber).toHaveBeenCalled()
+    expect(otherSubscriber).toHaveBeenCalled()
+  })
+
+  it('should allow to unsubscribe', () => {
+    const subscription = observable.subscribe(subscriber)
+
+    subscription.unsubscribe()
+    observable.notify()
+
+    expect(subscriber).not.toHaveBeenCalled()
+  })
+
+  it('should execute onFirstSubscribe callback', () => {
+    const onFirstSubscribe = jasmine.createSpy('callback')
+    const otherSubscriber = jasmine.createSpy('sub2')
+    observable = new Observable(onFirstSubscribe)
+    expect(onFirstSubscribe).not.toHaveBeenCalled()
+
+    observable.subscribe(subscriber)
+    expect(onFirstSubscribe).toHaveBeenCalledTimes(1)
+
+    observable.subscribe(otherSubscriber)
+    expect(onFirstSubscribe).toHaveBeenCalledTimes(1)
+  })
+
+  it('should execute onLastUnsubscribe callback', () => {
+    const onLastUnsubscribe = jasmine.createSpy('callback')
+    const otherSubscriber = jasmine.createSpy('sub2')
+    observable = new Observable(() => onLastUnsubscribe)
+    const subscription = observable.subscribe(subscriber)
+    const otherSubscription = observable.subscribe(otherSubscriber)
+    expect(onLastUnsubscribe).not.toHaveBeenCalled()
+
+    subscription.unsubscribe()
+    expect(onLastUnsubscribe).not.toHaveBeenCalled()
+
+    otherSubscription.unsubscribe()
+    expect(onLastUnsubscribe).toHaveBeenCalled()
+  })
+})

packages/core/src/tools/observable.ts

@@ -4,12 +4,21 @@ export interface Subscription {
 
 export class Observable<T> {
   private observers: Array<(data: T) => void> = []
+  private onLastUnsubscribe?: () => void
+
+  constructor(private onFirstSubscribe?: () => (() => void) | void) {}
 
   subscribe(f: (data: T) => void): Subscription {
+    if (!this.observers.length && this.onFirstSubscribe) {
+      this.onLastUnsubscribe = this.onFirstSubscribe() || undefined
+    }
     this.observers.push(f)
     return {
       unsubscribe: () => {
         this.observers = this.observers.filter((other) => f !== other)
+        if (!this.observers.length && this.onLastUnsubscribe) {
+          this.onLastUnsubscribe()
+        }
       },
     }
   }

packages/rum-core/src/boot/startRum.spec.ts

@@ -1,9 +1,8 @@
-import { RelativeTime, Configuration } from '@datadog/browser-core'
+import { RelativeTime, Configuration, Observable } from '@datadog/browser-core'
 import { RumSession } from '@datadog/browser-rum-core'
 import { createRumSessionMock, RumSessionMock } from '../../test/mockRumSession'
 import { isIE } from '../../../core/test/specHelper'
 import { noopRecorderApi, setup, TestSetupBuilder } from '../../test/specHelper'
-import { DOMMutationObservable } from '../browser/domMutationObservable'
 import { RumPerformanceNavigationTiming } from '../browser/performanceCollection'
 
 import { LifeCycle, LifeCycleEventType } from '../domain/lifeCycle'
@@ -26,7 +25,7 @@ function startRum(
   configuration: Configuration,
   session: RumSession,
   location: Location,
-  domMutationObservable: DOMMutationObservable
+  domMutationObservable: Observable<void>
 ) {
   const { stop: rumEventCollectionStop, foregroundContexts } = startRumEventCollection(
     applicationId,

packages/rum-core/src/browser/domMutationObservable.ts

@@ -1,57 +1,23 @@
-import { monitor, Subscription } from '@datadog/browser-core'
+import { monitor, Observable } from '@datadog/browser-core'
 
-export interface DOMMutationObservable {
-  subscribe(callback: () => void): Subscription
-}
-
-export function createDOMMutationObservable(): DOMMutationObservable {
-  let callbacks: Array<() => void> = []
+export function createDOMMutationObservable() {
   const MutationObserver = getMutationObserverConstructor()
-  const observer = MutationObserver ? new MutationObserver(monitor(notify)) : undefined
-
-  function notify() {
-    callbacks.forEach((callback) => callback())
-  }
 
-  function startDOMObservation() {
-    if (!observer) {
+  const observable: Observable<void> = new Observable<void>(() => {
+    if (!MutationObserver) {
       return
     }
-
+    const observer = new MutationObserver(monitor(() => observable.notify()))
     observer.observe(document, {
       attributes: true,
       characterData: true,
       childList: true,
       subtree: true,
     })
-  }
-
-  function stopDOMObservation() {
-    if (!observer) {
-      return
-    }
-
-    observer.disconnect()
-  }
+    return () => observer.disconnect()
+  })
 
-  return {
-    subscribe: (callback) => {
-      if (!callbacks.length) {
-        startDOMObservation()
-      }
-
-      callbacks.push(callback)
-      return {
-        unsubscribe: () => {
-          callbacks = callbacks.filter((other) => callback !== other)
-
-          if (!callbacks.length) {
-            stopDOMObservation()
-          }
-        },
-      }
-    },
-  }
+  return observable
 }
 
 type MutationObserverConstructor = new (callback: MutationCallback) => MutationObserver
@@ -66,7 +32,7 @@ export function getMutationObserverConstructor(): MutationObserverConstructor |
   let constructor: MutationObserverConstructor | undefined
   const browserWindow: BrowserWindow = window
 
-  // Angular uses Zone.js to provide a context persisting accross async tasks.  Zone.js replaces the
+  // Angular uses Zone.js to provide a context persisting across async tasks.  Zone.js replaces the
   // global MutationObserver constructor with a patched version to support the context propagation.
   // There is an ongoing issue[1][2] with this setup when using a MutationObserver within a Angular
   // component: on some occasions, the callback is being called in an infinite loop, causing the

packages/rum-core/src/domain/rumEventsCollection/action/actionCollection.ts

@@ -1,13 +1,12 @@
-import { combine, Configuration, toServerDuration, generateUUID } from '@datadog/browser-core'
+import { combine, Configuration, toServerDuration, generateUUID, Observable } from '@datadog/browser-core'
 import { ActionType, CommonContext, RumEventType, RawRumActionEvent } from '../../../rawRumEvent.types'
 import { LifeCycle, LifeCycleEventType, RawRumEventCollectedData } from '../../lifeCycle'
-import { DOMMutationObservable } from '../../../browser/domMutationObservable'
 import { ForegroundContexts } from '../../foregroundContexts'
 import { AutoAction, CustomAction, trackActions } from './trackActions'
 
 export function startActionCollection(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   configuration: Configuration,
   foregroundContexts: ForegroundContexts
 ) {

packages/rum-core/src/domain/rumEventsCollection/action/trackActions.ts

@@ -10,12 +10,12 @@ import {
   TimeStamp,
   Configuration,
   ONE_SECOND,
+  Observable,
 } from '@datadog/browser-core'
 import { ActionType } from '../../../rawRumEvent.types'
 import { LifeCycle, LifeCycleEventType } from '../../lifeCycle'
 import { EventCounts, trackEventCounts } from '../../trackEventCounts'
 import { waitIdlePageActivity } from '../../trackPageActivities'
-import { DOMMutationObservable } from '../../../browser/domMutationObservable'
 import { getActionNameFromElement } from './getActionNameFromElement'
 
 type AutoActionType = ActionType.CLICK
@@ -53,7 +53,7 @@ export const AUTO_ACTION_MAX_DURATION = 10 * ONE_SECOND
 
 export function trackActions(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   { actionNameAttribute }: Configuration
 ) {
   const action = startActionManagement(lifeCycle, domMutationObservable)
@@ -88,7 +88,7 @@ export function trackActions(
   }
 }
 
-function startActionManagement(lifeCycle: LifeCycle, domMutationObservable: DOMMutationObservable) {
+function startActionManagement(lifeCycle: LifeCycle, domMutationObservable: Observable<void>) {
   let currentAction: PendingAutoAction | undefined
   let currentIdlePageActivitySubscription: { stop: () => void }
 

packages/rum-core/src/domain/rumEventsCollection/view/trackViewMetrics.ts

@@ -1,10 +1,18 @@
-import { Duration, noop, elapsed, round, timeStampNow, RelativeTime, ONE_SECOND } from '@datadog/browser-core'
+import {
+  Duration,
+  noop,
+  elapsed,
+  round,
+  timeStampNow,
+  RelativeTime,
+  ONE_SECOND,
+  Observable,
+} from '@datadog/browser-core'
 import { RumLayoutShiftTiming, supportPerformanceTimingEvent } from '../../../browser/performanceCollection'
 import { ViewLoadingType } from '../../../rawRumEvent.types'
 import { LifeCycle, LifeCycleEventType } from '../../lifeCycle'
 import { EventCounts, trackEventCounts } from '../../trackEventCounts'
 import { waitIdlePageActivity } from '../../trackPageActivities'
-import { DOMMutationObservable } from '../../../browser/domMutationObservable'
 
 export interface ViewMetrics {
   eventCounts: EventCounts
@@ -14,7 +22,7 @@ export interface ViewMetrics {
 
 export function trackViewMetrics(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   scheduleViewUpdate: () => void,
   loadingType: ViewLoadingType
 ) {
@@ -96,7 +104,7 @@ function trackLoadingTime(loadType: ViewLoadingType, callback: (loadingTime: Dur
 
 function trackActivityLoadingTime(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   callback: (loadingTimeValue: Duration | undefined) => void
 ) {
   const startTime = timeStampNow()

packages/rum-core/src/domain/rumEventsCollection/view/trackViews.ts

@@ -11,8 +11,8 @@ import {
   timeStampNow,
   TimeStamp,
   display,
+  Observable,
 } from '@datadog/browser-core'
-import { DOMMutationObservable } from '../../../browser/domMutationObservable'
 import { ViewLoadingType, ViewCustomTimings } from '../../../rawRumEvent.types'
 
 import { LifeCycle, LifeCycleEventType } from '../../lifeCycle'
@@ -56,7 +56,7 @@ export const SESSION_KEEP_ALIVE_INTERVAL = 5 * ONE_MINUTE
 export function trackViews(
   location: Location,
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   areViewsTrackedAutomatically: boolean,
   initialViewName?: string
 ) {
@@ -168,7 +168,7 @@ export function trackViews(
 
 function newView(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   initialLocation: Location,
   loadingType: ViewLoadingType,
   referrer: string,

packages/rum-core/src/domain/rumEventsCollection/view/viewCollection.ts

@@ -5,19 +5,19 @@ import {
   ServerDuration,
   toServerDuration,
   Configuration,
+  Observable,
 } from '@datadog/browser-core'
 import { RecorderApi } from '../../../boot/rumPublicApi'
 import { RawRumViewEvent, RumEventType } from '../../../rawRumEvent.types'
 import { LifeCycle, LifeCycleEventType, RawRumEventCollectedData } from '../../lifeCycle'
-import { DOMMutationObservable } from '../../../browser/domMutationObservable'
 import { ForegroundContexts } from '../../foregroundContexts'
 import { trackViews, ViewEvent } from './trackViews'
 
 export function startViewCollection(
   lifeCycle: LifeCycle,
   configuration: Configuration,
   location: Location,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   foregroundContexts: ForegroundContexts,
   recorderApi: RecorderApi,
   initialViewName?: string

packages/rum-core/src/domain/trackPageActivities.ts

@@ -1,5 +1,4 @@
 import { monitor, Observable, Subscription, TimeStamp, timeStampNow } from '@datadog/browser-core'
-import { DOMMutationObservable } from '../browser/domMutationObservable'
 import { LifeCycle, LifeCycleEventType } from './lifeCycle'
 
 // Delay to wait for a page activity to validate the tracking process
@@ -15,7 +14,7 @@ export type CompletionCallbackParameters = { hadActivity: true; endTime: TimeSta
 
 export function waitIdlePageActivity(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable,
+  domMutationObservable: Observable<void>,
   completionCallback: (params: CompletionCallbackParameters) => void,
   maxDuration?: number
 ) {
@@ -65,7 +64,7 @@ export function waitIdlePageActivity(
 // process is still alive after maxDuration, it has been validated.
 export function trackPageActivities(
   lifeCycle: LifeCycle,
-  domMutationObservable: DOMMutationObservable
+  domMutationObservable: Observable<void>
 ): { observable: Observable<PageActivityEvent>; stop: () => void } {
   const observable = new Observable<PageActivityEvent>()
   const subscriptions: Subscription[] = []