Expose Stream APIs (#633)

* remove unsupported field in e2e

* expose objects, types and utils from core to support the streaming apis

* expose streaming apis in JS

* expose streaming apis in realtime-api

* js e2e tests for streamings

* changeset

* rename to just stream

* save last file

.changeset/angry-avocados-yell.md

@@ -0,0 +1,6 @@
+---
+'@signalwire/js': minor
+'@signalwire/realtime-api': minor
+---
+
+Expose `getStreams` and `startStream` on RoomSession.

.changeset/rude-cobras-perform.md

@@ -0,0 +1,5 @@
+---
+'@sw-internal/e2e-js': patch
+---
+
+Add e2e for the Stream APIs

.changeset/spotty-buttons-promise.md

@@ -0,0 +1,5 @@
+---
+'@signalwire/core': minor
+---
+
+Add methods, interfaces and utils to support the Stream APIs.

internal/e2e-js/tests/roomSessionStreaming.spec.ts

@@ -0,0 +1,172 @@
+import { test, expect } from '@playwright/test'
+import type { Video } from '@signalwire/js'
+import { createTestServer, createTestRoomSession } from '../utils'
+
+test.describe('RoomSession', () => {
+ let server: any = null
+
+ test.beforeAll(async () => {
+ server = await createTestServer()
+ await server.start()
+ })
+
+ test.afterAll(async () => {
+ await server.close()
+ })
+
+ test('should handle Stream events and methods', async ({ context }) => {
+ const pageOne = await context.newPage()
+ const pageTwo = await context.newPage()
+ const pageThree = await context.newPage()
+
+ pageOne.on('console', (log) => console.log('[pageOne]', log))
+ pageTwo.on('console', (log) => console.log('[pageTwo]', log))
+ pageThree.on('console', (log) => console.log('[pageThree]', log))
+
+ await Promise.all([
+ pageOne.goto(server.url),
+ pageTwo.goto(server.url),
+ pageThree.goto(server.url),
+ ])
+
+ const connectionSettings = {
+ vrt: {
+ room_name: 'another',
+ user_name: 'e2e_test',
+ auto_create_room: true,
+ permissions: ['room.stream'],
+ },
+ initialEvents: ['stream.started', 'stream.ended'],
+ }
+
+ await Promise.all([
+ createTestRoomSession(pageOne, connectionSettings),
+ createTestRoomSession(pageTwo, connectionSettings),
+ createTestRoomSession(pageThree, connectionSettings),
+ ])
+
+ // --------------- Joining from the 2nd tab and resolve on 'stream.started' ---------------
+ const pageTwoStreamPromise = pageTwo.evaluate(() => {
+ return new Promise((resolve) => {
+ // @ts-expect-error
+ const roomObj: Video.RoomSession = window._roomObj
+ roomObj.on('stream.started', (stream: any) => resolve(stream))
+ roomObj.join()
+ })
+ })
+
+ // --------------- Joining from the 1st tab and resolve on 'room.joined' ---------------
+ await pageOne.evaluate(() => {
+ return new Promise((resolve) => {
+ // @ts-expect-error
+ const roomObj = window._roomObj
+ roomObj.on('room.joined', resolve)
+ roomObj.join()
+ })
+ })
+
+ // Checks that the video is visible on pageOne
+ await pageOne.waitForSelector('div[id^="sw-sdk-"] > video', {
+ timeout: 5000,
+ })
+
+ // --------------- Start stream from 1st room ---------------
+ await pageOne.evaluate(
+ async ({ STREAMING_URL }) => {
+ // @ts-expect-error
+ const roomObj: Video.RoomSession = window._roomObj
+
+ const streamStarted = new Promise((resolve, reject) => {
+ roomObj.on('stream.started', (params) => {
+ if (params.state === 'streaming') {
+ resolve(true)
+ } else {
+ reject(new Error('[stream.started] state is not "stream"'))
+ }
+ })
+ })
+
+ await roomObj.startStream({
+ url: STREAMING_URL!,
+ })
+
+ return streamStarted
+ },
+ { STREAMING_URL: process.env.STREAMING_URL }
+ )
+
+ // Checks that the video is visible on pageTwo
+ await pageTwo.waitForSelector('div[id^="sw-sdk-"] > video', {
+ timeout: 5000,
+ })
+
+ // --------------- Joining from the 3rd tab and get the active streams ---------------
+ const { streamsOnJoined, streamsOnGet, streamOnEnd }: any =
+ await pageThree.evaluate(() => {
+ return new Promise((resolve) => {
+ // @ts-expect-error
+ const roomObj: Video.RoomSession = window._roomObj
+
+ roomObj.on('room.joined', async (params) => {
+ const result = await roomObj.getStreams()
+
+ const streamOnEnd = await Promise.all(
+ result.streams.map((stream: any) => {
+ const streamEnded = new Promise((resolve) => {
+ roomObj.on('stream.ended', (params) => {
+ if (params.id === stream.id) {
+ resolve(params)
+ }
+ })
+ })
+
+ stream.stop().then(() => {
+ console.log(`Stream ${stream.id} stopped!`)
+ })
+
+ return streamEnded
+ })
+ )
+
+ resolve({
+ streamsOnJoined: params.room_session.streams,
+ streamsOnGet: result.streams,
+ streamOnEnd,
+ })
+ })
+
+ roomObj.join()
+ })
+ })
+
+ expect(streamsOnJoined.length).toEqual(streamsOnGet.length)
+ expect(streamsOnGet.length).toEqual(streamOnEnd.length)
+ ;[streamsOnJoined, streamsOnGet, streamsOnGet].forEach((streams: any[]) => {
+ streams.forEach((stream) => {
+ // Since functions can't be serialized back to this
+ // thread (from the previous step) we just check that
+ // the property is there.
+ expect('stop' in stream).toBeTruthy()
+ expect(stream.id).toBeDefined()
+ expect(stream.roomSessionId).toBeDefined()
+ expect(stream.state).toBeDefined()
+ expect(stream.url).toEqual(process.env.STREAMING_URL)
+ })
+ })
+
+ // --------------- Make sure pageTwo got the `stream.started` event ---------------
+ await pageTwoStreamPromise
+
+ await new Promise((r) => setTimeout(r, 1000))
+
+ // --------------- Leaving the rooms ---------------
+ await Promise.all([
+ // @ts-expect-error
+ pageOne.evaluate(() => window._roomObj.leave()),
+ // @ts-expect-error
+ pageTwo.evaluate(() => window._roomObj.leave()),
+ // @ts-expect-error
+ pageThree.evaluate(() => window._roomObj.leave()),
+ ])
+ })
+})

internal/e2e-js/utils.ts

@@ -66,7 +66,6 @@ export const createTestRoomSession = async (
  audio: true,
  video: true,
  logLevel: 'trace',
- _hijack: true,
  debug: {
  logWsTraffic: true,
  },

packages/core/src/index.ts

@@ -85,7 +85,11 @@ export * as Rooms from './rooms'
 export * as Chat from './chat'
 export * as PubSub from './pubSub'
 export * as MemberPosition from './memberPosition'
-export type { RoomSessionRecording, RoomSessionPlayback } from './rooms'
+export type {
+ RoomSessionRecording,
+ RoomSessionPlayback,
+ RoomSessionStream,
+} from './rooms'
 export const selectors = {
  ...sessionSelectors,
 }

packages/core/src/redux/features/shared/namespace.ts

@@ -10,6 +10,7 @@ import type {
  VideoRoomAudienceCountEvent,
  VoiceCallEvent,
  InternalVideoRoomAudienceCountEvent,
+ VideoStreamEvent,
 } from '../../../types'
 import { getLogger } from '../../../utils'
 import type { MapToPubSubShape, PubSubAction } from '../../interfaces'
@@ -59,6 +60,12 @@ const isVideoPlaybackEvent = (
  return action.type.startsWith('video.playback.')
 }
 
+const isVideoStreamEvent = (
+ action: PubSubAction
+): action is MapToPubSubShape<VideoStreamEvent> => {
+ return action.type.startsWith('video.stream.')
+}
+
 const isChatEvent = (
  action: PubSubAction
 ): action is MapToPubSubShape<ChatEvent> => {
@@ -79,6 +86,7 @@ export const findNamespaceInPayload = (action: PubSubAction): string => {
  isVideoLayoutEvent(action) ||
  isVideoRecordingEvent(action) ||
  isVideoPlaybackEvent(action) ||
+ isVideoStreamEvent(action) ||
  isVideoRoomAudienceCountEvent(action)
  ) {
  return action.payload.room_session_id

packages/core/src/rooms/RoomSessionStream.test.ts

@@ -0,0 +1,42 @@
+import { configureJestStore } from '../testUtils'
+import { EventEmitter } from '../utils/EventEmitter'
+import {
+ createRoomSessionStreamObject,
+ RoomSessionStream,
+ RoomSessionStreamEventsHandlerMapping,
+} from './RoomSessionStream'
+
+describe('RoomSessionStream', () => {
+ describe('createRoomSessionStreamObject', () => {
+ let instance: RoomSessionStream
+ beforeEach(() => {
+ instance = createRoomSessionStreamObject({
+ store: configureJestStore(),
+ emitter: new EventEmitter<RoomSessionStreamEventsHandlerMapping>(),
+ })
+ // @ts-expect-error
+ instance.execute = jest.fn()
+ })
+
+ it('should control an active stream', async () => {
+ // Mock properties
+ instance.id = 'c22d7223-5a01-49fe-8da0-46bec8e75e32'
+ instance.roomSessionId = 'room-session-id'
+
+ const baseExecuteParams = {
+ method: '',
+ params: {
+ room_session_id: 'room-session-id',
+ stream_id: 'c22d7223-5a01-49fe-8da0-46bec8e75e32',
+ },
+ }
+
+ await instance.stop()
+ // @ts-expect-error
+ expect(instance.execute).toHaveBeenLastCalledWith({
+ ...baseExecuteParams,
+ method: 'video.stream.stop',
+ })
+ })
+ })
+})

packages/core/src/rooms/RoomSessionStream.ts

@@ -0,0 +1,52 @@
+import { connect } from '../redux'
+import { BaseComponent } from '../BaseComponent'
+import { BaseComponentOptions } from '../utils/interfaces'
+import { OnlyFunctionProperties } from '../types'
+import type {
+ VideoStreamContract,
+ VideoStreamEventNames,
+} from '../types/videoStream'
+
+/**
+ * Represents a specific Stream of a room session.
+ */
+export interface RoomSessionStream extends VideoStreamContract {}
+
+export type RoomSessionStreamEventsHandlerMapping = Record<
+ VideoStreamEventNames,
+ (stream: RoomSessionStream) => void
+>
+
+export class RoomSessionStreamAPI
+ extends BaseComponent<RoomSessionStreamEventsHandlerMapping>
+ implements OnlyFunctionProperties<RoomSessionStream>
+{
+ async stop() {
+ await this.execute({
+ method: 'video.stream.stop',
+ params: {
+ room_session_id: this.getStateProperty('roomSessionId'),
+ stream_id: this.getStateProperty('id'),
+ },
+ })
+ }
+}
+
+export const createRoomSessionStreamObject = (
+ params: BaseComponentOptions<RoomSessionStreamEventsHandlerMapping>
+): RoomSessionStream => {
+ const stream = connect<
+ RoomSessionStreamEventsHandlerMapping,
+ RoomSessionStreamAPI,
+ RoomSessionStream
+ >({
+ store: params.store,
+ Component: RoomSessionStreamAPI,
+ componentListeners: {
+ errors: 'onError',
+ responses: 'onSuccess',
+ },
+ })(params)
+
+ return stream
+}

packages/core/src/rooms/index.ts

@@ -11,3 +11,4 @@ export interface BaseRoomInterface<
 export * from './methods'
 export * from './RoomSessionRecording'
 export * from './RoomSessionPlayback'
+export * from './RoomSessionStream'