✨ stream plugin

Author: mormubis

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

@@ -12,7 +12,7 @@ import {
 } from '@datadog/browser-core'
 import type { RumEventDomainContext } from '../domainContext.types'
 import { RumEventType } from '../rawRumEvent.types'
-import type { RumEvent } from '../rumEvent.types'
+import type { RumEvent, RumViewEvent, RumVitalEvent } from '../rumEvent.types'
 import type { LifeCycle } from './lifeCycle'
 import { LifeCycleEventType } from './lifeCycle'
 import type { RumConfiguration } from './configuration'
@@ -133,11 +133,28 @@ export function startRumAssembly(
         }
 
         if (rawRumEvent.type === 'stream') {
-          // @ts-expect-error TBF
-          serverRumEvent.type = 'view'
-        }
+          const streamEvent = {
+            ...(serverRumEvent as RumViewEvent),
+            view: {
+              ...serverRumEvent.view,
+              id: serverRumEvent.stream?.id,
+              action: {
+                count: 0,
+              },
+              error: {
+                count: 0,
+              },
+              resource: {
+                count: 0,
+              },
+            },
+            type: 'view',
+          }
 
-        lifeCycle.notify(LifeCycleEventType.RUM_EVENT_COLLECTED, serverRumEvent)
+          lifeCycle.notify(LifeCycleEventType.RUM_EVENT_COLLECTED, streamEvent as RumEvent & Context)
+        } else {
+          lifeCycle.notify(LifeCycleEventType.RUM_EVENT_COLLECTED, serverRumEvent)
+        }
       }
     }
   )

packages/rum-core/src/domain/stream/index.ts

@@ -13,7 +13,6 @@ export function createStreamPlugin(): { plugin: RumPlugin; createStream: () => R
 
   const api: API = {
     get addEvent() {
-      console.log('>>>', 'getter addEvent', addEvent)
       return addEvent ?? store
     },
   }

packages/rum-core/src/domain/stream/metric.ts

@@ -1,19 +1,41 @@
-// eslint-disable-next-line no-restricted-syntax
-abstract class Metric {
-  constructor(private name: string) {}
+export function createWeightAverageMetric() {
+  let lastUpdate = 0
+  let weight = 0
+  let value: number | undefined
 
-  update(value: number, weight?: number): void {
-    // This method is a placeholder for metric updates.
+  return {
+    get value() {
+      console.log('>>>', value)
+      return value
+    },
+    update(newLastUpdate: number, newValue: number) {
+      if (newLastUpdate <= lastUpdate) {
+        return
+      }
+
+      const newWeight = newLastUpdate - lastUpdate
+      value = ((value ?? 0) * weight + newValue * newWeight) / (weight + newWeight)
+      weight += newWeight
+      lastUpdate = newLastUpdate
+    },
   }
 }
 
-// eslint-disable-next-line no-restricted-syntax
-export class WeightAverageMetric extends Metric {
-  private average: number = 0
-  private weight: number = 0
+export function createLastMetric() {
+  let lastUpdate = 0
+  let value: number | undefined
+
+  return {
+    get value() {
+      return value
+    },
+    update(newLastUpdate: number, newValue: number) {
+      if (newLastUpdate <= lastUpdate) {
+        return
+      }
 
-  update(value: number, weight: number): void {
-    this.average = (this.average * this.weight + value * weight) / (this.weight + weight)
-    this.weight += weight
+      value = newValue
+      lastUpdate = newLastUpdate
+    },
   }
 }

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

@@ -1,36 +1,70 @@
 import { clocksNow, generateUUID } from '@datadog/browser-core'
 import type { StartRumResult } from '../..'
-import type { WeightAverageMetric } from './metric'
+import { createLastMetric, createWeightAverageMetric } from './metric'
 import { createTimer } from './timer'
 
 export interface API {
   addEvent: AddEvent
 }
 
-export type AddEvent = StartRumResult['addEvent']
-
 interface Meta {
   duration?: number
   format?: string
   resolution?: string
 }
 
-interface Metrics {
-  bitrate: WeightAverageMetric
-  fps: WeightAverageMetric
-  timestamp: number
-  watchTime: number
-}
+export type AddEvent = StartRumResult['addEvent']
 
 type Transition = 'end' | 'error' | 'pause' | 'play' | 'preload' | 'quit' | 'rebuff' | 'seek' | 'start'
 
 export function createStream(api: API) {
   const id = generateUUID()
   const origin = clocksNow()
   const timer = createTimer()
+  const meta: Meta = {}
+  const metrics = {
+    bitrate: createWeightAverageMetric(),
+    fps: createWeightAverageMetric(),
+    timestamp: createLastMetric(),
+    watchTime: createLastMetric(),
+  }
+
+  function sendStreamEvent() {
+    const now = clocksNow()
+
+    api.addEvent(
+      now.relative,
+      {
+        date: now.timeStamp,
+        type: 'stream',
+        stream: {
+          id,
+          bitrate: metrics.bitrate.value,
+          duration: meta.duration,
+          format: meta.format,
+          fps: metrics.fps.value,
+          resolution: meta.resolution,
+          timestamp: metrics.timestamp.value,
+          watch_time: metrics.watchTime.value,
+        },
+      },
+      {}
+    )
+  }
 
   return {
-    interaction(id: string): void {},
+    interaction(id: string): void {
+      console.log('>>>', 'interaction', id)
+    },
+    set<K extends keyof Meta>(key: K, value: Meta[K]): void {
+      if (meta[key] !== undefined) {
+        return
+      }
+
+      meta[key] = value
+
+      sendStreamEvent()
+    },
     transition(state: Transition, context: any): void {
       if (state === 'play') {
         timer.start()
@@ -53,20 +87,10 @@ export function createStream(api: API) {
         context
       )
     },
-    update(key: string, value: any): void {
-      const now = clocksNow()
+    update(key: keyof typeof metrics, value: number): void {
+      metrics[key].update(timer.value, value)
 
-      api.addEvent(
-        now.relative,
-        {
-          date: now.timeStamp,
-          type: 'stream',
-          stream: {
-            id: generateUUID(),
-          },
-        },
-        {}
-      )
+      sendStreamEvent()
     },
   }
 }

packages/rum-core/src/domain/stream/timer.ts

@@ -2,21 +2,33 @@ import type { RelativeTime } from '@datadog/browser-core'
 import { relativeNow } from '@datadog/browser-core'
 
 export function createTimer() {
-  let start: RelativeTime
-  let count = 0
+  let counter = 0
+  let start: RelativeTime = 0 as RelativeTime
+  let stopped = true
 
   return {
-    start(): number {
-      start = relativeNow()
+    get value() {
+      if (stopped) {
+        return counter
+      }
 
-      return count
+      return relativeNow() - start
     },
-    stop(): number {
-      const duration = relativeNow() - start
+    start(): void {
+      if (!stopped) {
+        return
+      }
 
-      count += duration
+      start = relativeNow()
+      stopped = false
+    },
+    stop(): void {
+      if (stopped) {
+        return
+      }
 
-      return count
+      counter = relativeNow() - start
+      stopped = true
     },
   }
 }