✨ [RUM-10144] apply context defined just after init to events generated during init

packages/core/src/index.ts

@@ -56,7 +56,7 @@ export {
 } from './domain/telemetry'
 export { monitored, monitor, callMonitored, setDebugMode, monitorError } from './tools/monitor'
 export type { Subscription } from './tools/observable'
-export { Observable } from './tools/observable'
+export { Observable, BufferedObservable } from './tools/observable'
 export type { SessionManager } from './domain/session/sessionManager'
 export { startSessionManager, stopSessionManager } from './domain/session/sessionManager'
 export {

packages/core/src/tools/boundedBuffer.ts

@@ -2,12 +2,18 @@ import { removeItem } from './utils/arrayUtils'
 
 const BUFFER_LIMIT = 500
 
+/**
+ * @deprecated Use `BufferedObservable` instead.
+ */
 export interface BoundedBuffer<T = void> {
   add: (callback: (arg: T) => void) => void
   remove: (callback: (arg: T) => void) => void
   drain: (arg: T) => void
 }
 
+/**
+ * @deprecated Use `BufferedObservable` instead.
+ */
 export function createBoundedBuffer<T = void>(): BoundedBuffer<T> {
   const buffer: Array<(arg: T) => void> = []
 

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

@@ -1,4 +1,4 @@
-import { mergeObservables, Observable } from './observable'
+import { BufferedObservable, mergeObservables, Observable } from './observable'
 
 describe('observable', () => {
   let observable: Observable<void>
@@ -119,3 +119,144 @@ describe('mergeObservables', () => {
     expect(subscriber).not.toHaveBeenCalled()
   })
 })
+
+describe('BufferedObservable', () => {
+  it('invokes the observer with buffered data', async () => {
+    const observable = new BufferedObservable<string>(100)
+    observable.notify('first')
+    observable.notify('second')
+
+    const observer = jasmine.createSpy('observer')
+    observable.subscribe(observer)
+
+    await nextMicroTask()
+
+    expect(observer).toHaveBeenCalledTimes(2)
+  })
+
+  it('invokes the observer asynchronously', async () => {
+    const observable = new BufferedObservable<string>(100)
+    observable.notify('first')
+
+    const observer = jasmine.createSpy('observer')
+    observable.subscribe(observer)
+
+    expect(observer).not.toHaveBeenCalled()
+
+    await nextMicroTask()
+
+    expect(observer).toHaveBeenCalledWith('first')
+  })
+
+  it('invokes the observer when new data is notified after subscription', async () => {
+    const observable = new BufferedObservable<string>(100)
+
+    const observer = jasmine.createSpy('observer')
+    observable.subscribe(observer)
+
+    observable.notify('first')
+
+    await nextMicroTask()
+
+    observable.notify('second')
+
+    expect(observer).toHaveBeenCalledTimes(2)
+    expect(observer).toHaveBeenCalledWith('first')
+    expect(observer).toHaveBeenCalledWith('second')
+  })
+
+  it('drops data when the buffer is full', async () => {
+    const observable = new BufferedObservable<string>(2)
+    observable.notify('first') // This should be dropped
+    observable.notify('second')
+    observable.notify('third')
+
+    const observer = jasmine.createSpy('observer')
+    observable.subscribe(observer)
+
+    await nextMicroTask()
+
+    expect(observer).toHaveBeenCalledTimes(2)
+    expect(observer).toHaveBeenCalledWith('second')
+    expect(observer).toHaveBeenCalledWith('third')
+  })
+
+  it('allows to unsubscribe from the observer, the middle of buffered data', async () => {
+    const observable = new BufferedObservable<string>(100)
+    observable.notify('first')
+    observable.notify('second')
+
+    const observer = jasmine.createSpy('observer').and.callFake(() => {
+      subscription.unsubscribe()
+    })
+    const subscription = observable.subscribe(observer)
+
+    await nextMicroTask()
+
+    expect(observer).toHaveBeenCalledTimes(1)
+  })
+
+  it('allows to unsubscribe before the buffered data', async () => {
+    const observable = new BufferedObservable<string>(100)
+    observable.notify('first')
+
+    const observer = jasmine.createSpy('observer')
+    const subscription = observable.subscribe(observer)
+
+    subscription.unsubscribe()
+
+    await nextMicroTask()
+
+    expect(observer).not.toHaveBeenCalled()
+  })
+
+  it('allows to unsubscribe after the buffered data', async () => {
+    const observable = new BufferedObservable<string>(100)
+
+    const observer = jasmine.createSpy('observer')
+    const subscription = observable.subscribe(observer)
+
+    await nextMicroTask()
+
+    subscription.unsubscribe()
+
+    observable.notify('first')
+
+    expect(observer).not.toHaveBeenCalled()
+  })
+
+  it('calling unbuffer() removes buffered data', async () => {
+    const observable = new BufferedObservable<string>(2)
+    observable.notify('first')
+    observable.notify('second')
+
+    observable.unbuffer()
+    await nextMicroTask()
+
+    const observer = jasmine.createSpy('observer')
+    observable.subscribe(observer)
+    await nextMicroTask()
+
+    expect(observer).not.toHaveBeenCalled()
+  })
+
+  it('when calling unbuffer() right after subscription, buffered data should still be notified', async () => {
+    const observable = new BufferedObservable<string>(2)
+    observable.notify('first')
+    observable.notify('second')
+
+    const observer = jasmine.createSpy('observer')
+    observable.subscribe(observer)
+
+    observable.unbuffer()
+    await nextMicroTask()
+
+    expect(observer).toHaveBeenCalledTimes(2)
+    expect(observer).toHaveBeenCalledWith('first')
+    expect(observer).toHaveBeenCalledWith('second')
+  })
+})
+
+function nextMicroTask() {
+  return Promise.resolve()
+}

packages/core/src/tools/observable.ts

@@ -1,32 +1,42 @@
+import { monitorError } from './monitor'
+
 export interface Subscription {
   unsubscribe: () => void
 }
 
+type Observer<T> = (data: T) => void
+
 // eslint-disable-next-line no-restricted-syntax
 export class Observable<T> {
-  private observers: Array<(data: T) => void> = []
+  protected observers: Array<Observer<T>> = []
   private onLastUnsubscribe?: () => void
 
   constructor(private onFirstSubscribe?: (observable: Observable<T>) => (() => void) | void) {}
 
-  subscribe(f: (data: T) => void): Subscription {
-    this.observers.push(f)
-    if (this.observers.length === 1 && this.onFirstSubscribe) {
-      this.onLastUnsubscribe = this.onFirstSubscribe(this) || undefined
-    }
+  subscribe(observer: Observer<T>): Subscription {
+    this.addObserver(observer)
     return {
-      unsubscribe: () => {
-        this.observers = this.observers.filter((other) => f !== other)
-        if (!this.observers.length && this.onLastUnsubscribe) {
-          this.onLastUnsubscribe()
-        }
-      },
+      unsubscribe: () => this.removeObserver(observer),
     }
   }
 
   notify(data: T) {
     this.observers.forEach((observer) => observer(data))
   }
+
+  protected addObserver(observer: Observer<T>) {
+    this.observers.push(observer)
+    if (this.observers.length === 1 && this.onFirstSubscribe) {
+      this.onLastUnsubscribe = this.onFirstSubscribe(this) || undefined
+    }
+  }
+
+  protected removeObserver(observer: Observer<T>) {
+    this.observers = this.observers.filter((other) => observer !== other)
+    if (!this.observers.length && this.onLastUnsubscribe) {
+      this.onLastUnsubscribe()
+    }
+  }
 }
 
 export function mergeObservables<T>(...observables: Array<Observable<T>>) {
@@ -37,3 +47,62 @@ export function mergeObservables<T>(...observables: Array<Observable<T>>) {
     return () => subscriptions.forEach((subscription) => subscription.unsubscribe())
   })
 }
+
+// eslint-disable-next-line no-restricted-syntax
+export class BufferedObservable<T> extends Observable<T> {
+  private buffer: T[] = []
+
+  constructor(private maxBufferSize: number) {
+    // no onFirstSubscribe as it makes less sense with buffered data
+    super()
+  }
+
+  notify(data: T) {
+    this.buffer.push(data)
+    if (this.buffer.length > this.maxBufferSize) {
+      this.buffer.shift()
+    }
+    super.notify(data)
+  }
+
+  subscribe(observer: Observer<T>): Subscription {
+    let closed = false
+
+    const subscription = {
+      unsubscribe: () => {
+        closed = true
+        this.removeObserver(observer)
+      },
+    }
+
+    enqueueMicroTask(() => {
+      for (const data of this.buffer) {
+        if (closed) {
+          return
+        }
+        observer(data)
+      }
+
+      if (!closed) {
+        this.addObserver(observer)
+      }
+    })
+
+    return subscription
+  }
+
+  /**
+   * Drop buffered data and don't buffer future data. This is to avoid leaking memory when it's not
+   * needed anymore. This is not be required in most cases, but still useful to clarify our intent
+   * and lowering our memory impact.
+   */
+  unbuffer() {
+    enqueueMicroTask(() => {
+      this.maxBufferSize = this.buffer.length = 0
+    })
+  }
+}
+
+function enqueueMicroTask(callback: () => void) {
+  Promise.resolve().then(callback).catch(monitorError)
+}