wip

Author: wkozyra95

tools/src/bin/generate_from_types.rs

@@ -30,6 +30,7 @@ enum ApiTypes {
     RegisterWebRenderer(smelter_api::WebRendererSpec),
     RegisterShader(smelter_api::ShaderSpec),
     UpdateOutput(Box<routes::update_output::UpdateOutputRequest>),
+    UpdateInput(routes::update_input::UpdateInputRequest),
 }
 
 pub fn generate_json_schema(check_flag: bool) {

ts/create-smelter-app/package.json

@@ -14,7 +14,7 @@
   ],
   "main": "dist/index.js",
   "scripts": {
-    "lint": "eslint src",
+    "lint": "eslint ./src",
     "typecheck": "tsc --noEmit",
     "watch": "tsc --watch --preserveWatchOutput",
     "build": "tsc",

ts/examples/node-examples/src/utils.ts

@@ -1,5 +1,5 @@
 import path from 'path';
-import fs, { mkdirp, pathExists, writeFile } from 'fs-extra';
+import fs, { mkdirp, pathExists } from 'fs-extra';
 import type { ChildProcess } from 'child_process';
 import { spawn as nodeSpawn } from 'child_process';
 import { promisify } from 'util';
@@ -8,28 +8,6 @@ import fetch from 'node-fetch';
 
 const pipeline = promisify(Stream.pipeline);
 
-const TMP_SDP_DIR = '/tmp/smelter-examples';
-
-export async function ffplayStartRtpServerAsync(
-  ip: string,
-  video_port: number,
-  audio_port: number | undefined = undefined
-): Promise<{ spawn_promise: SpawnPromise }> {
-  await mkdirp(TMP_SDP_DIR);
-  let sdpFilePath;
-  if (audio_port === undefined) {
-    sdpFilePath = path.join(TMP_SDP_DIR, `video_input_${video_port}.sdp`);
-    await writeVideoSdpFile(ip, video_port, sdpFilePath);
-  } else {
-    sdpFilePath = path.join(TMP_SDP_DIR, `video_audio_input_${video_port}_${audio_port}.sdp`);
-    await writeVideoAudioSdpFile(ip, video_port, audio_port, sdpFilePath);
-  }
-
-  const promise = spawn('ffplay', ['-protocol_whitelist', 'file,rtp,udp', sdpFilePath]);
-  await sleep(2000);
-  return { spawn_promise: promise };
-}
-
 export async function ffplayStartRtmpServerAsync(
   port: number
 ): Promise<{ spawn_promise: SpawnPromise }> {
@@ -41,37 +19,6 @@ export async function ffplayStartRtmpServerAsync(
   return { spawn_promise: promise };
 }
 
-export async function gstReceiveTcpStream(
-  ip: string,
-  port: number
-): Promise<{ spawn_promise: SpawnPromise }> {
-  const tcpReceiver = `tcpclientsrc host=${ip} port=${port} ! "application/x-rtp-stream" ! rtpstreamdepay ! queue ! demux.`;
-  const videoPipe =
-    'demux.src_96 ! "application/x-rtp,media=video,clock-rate=90000,encoding-name=H264" ! queue ! rtph264depay ! decodebin ! videoconvert ! autovideosink';
-  const audioPipe =
-    'demux.src_97 ! "application/x-rtp,media=audio,clock-rate=48000,encoding-name=OPUS" ! queue ! rtpopusdepay ! decodebin ! audioconvert ! autoaudiosink ';
-  const gstCmd = `gst-launch-1.0 -v rtpptdemux name=demux ${tcpReceiver} ${videoPipe} ${audioPipe}`;
-
-  const promise = spawn('bash', ['-c', gstCmd]);
-  return { spawn_promise: promise };
-}
-
-export function ffmpegSendVideoFromMp4(port: number, mp4Path: string): SpawnPromise {
-  return spawn('ffmpeg', [
-    '-stream_loop',
-    '-1',
-    '-re',
-    '-i',
-    mp4Path,
-    '-an',
-    '-c:v',
-    'libx264',
-    '-f',
-    'rtp',
-    `rtp://127.0.0.1:${port}?rtcpport=${port}`,
-  ]);
-}
-
 interface SpawnPromise extends Promise<void> {
   child: ChildProcess;
 }
@@ -96,46 +43,6 @@ function spawn(command: string, args: string[]): SpawnPromise {
   }) as SpawnPromise;
 }
 
-async function writeVideoAudioSdpFile(
-  ip: string,
-  video_port: number,
-  audio_port: number,
-  destination: string
-): Promise<void> {
-  await writeFile(
-    destination,
-    `
-v=0
-o=- 0 0 IN IP4 ${ip}
-s=No Name
-c=IN IP4 ${ip}
-m=video ${video_port} RTP/AVP 96
-a=rtpmap:96 H264/90000
-a=fmtp:96 packetization-mode=1
-a=rtcp-mux
-m=audio ${audio_port} RTP/AVP 97
-a=rtpmap:97 opus/48000/2
-a=rtcp-mux
-`
-  );
-}
-
-async function writeVideoSdpFile(ip: string, port: number, destination: string): Promise<void> {
-  await writeFile(
-    destination,
-    `
-v=0
-o=- 0 0 IN IP4 ${ip}
-s=No Name
-c=IN IP4 ${ip}
-m=video ${port} RTP/AVP 96
-a=rtpmap:96 H264/90000
-a=fmtp:96 packetization-mode=1
-a=rtcp-mux
-`
-  );
-}
-
 export async function sleep(timeoutMs: number): Promise<void> {
   await new Promise<void>(res => {
     setTimeout(() => {

ts/examples/vite-example/src/smelter-examples/MultipleOutputs.tsx

@@ -60,7 +60,10 @@ function MultipleOutputs() {
   );
 }
 
-function SceneTile(props: { state?: 'ready' | 'playing' | 'finished'; inputId: string }) {
+function SceneTile(props: {
+  state?: 'ready' | 'playing' | 'paused' | 'finished';
+  inputId: string;
+}) {
   if (props.state === 'playing') {
     return (
       <View>

ts/examples/vite-example/src/smelter-examples/playground/Scene.tsx

@@ -48,8 +48,8 @@ function volumeForId(store: OutputStore, id: string): number {
 
 type InputTileProps = {
   inputId: string;
-  videoState?: 'ready' | 'playing' | 'finished';
-  audioState?: 'ready' | 'playing' | 'finished';
+  videoState?: 'ready' | 'playing' | 'paused' | 'finished';
+  audioState?: 'ready' | 'playing' | 'paused' | 'finished';
   offsetMs?: number | null;
   videoDurationMs?: number;
   audioDurationMs?: number;

ts/package.json

@@ -5,7 +5,7 @@
   "type": "module",
   "scripts": {
     "lint": "pnpm -r --no-bail run lint",
-    "lint:fix": "pnpm -r run lint -- --fix",
+    "lint:fix": "pnpm -r run lint --fix",
     "build": "pnpm -r run build",
     "build:web-wasm": "pnpm -C smelter-browser-render run build-wasm && pnpm -C smelter run build && pnpm -C smelter-core run build && pnpm -C smelter-browser-render run build && pnpm -C smelter-web-wasm run build",
     "build:web-client": "pnpm -C smelter run build && pnpm -C smelter-core run build && pnpm -C smelter-web-client run build",

ts/smelter-core/src/api.ts

@@ -21,10 +21,15 @@ export type MultipartRequest = {
 };
 
 export type RegisterInputResponse = {
-  endpoint_route?: string;
-  bearer_token?: string;
+  // Common options (but MP4 only for now)
+
   video_duration_ms?: number;
   audio_duration_ms?: number;
+
+  // WHIP specific
+
+  endpoint_route?: string;
+  bearer_token?: string;
 };
 
 export type RegisterOutputResponse = {
@@ -79,6 +84,17 @@ export class ApiClient {
     });
   }
 
+  public async updateInput(
+    inputId: InputRef,
+    request: Api.UpdateInputRequest
+  ): Promise<RegisterInputResponse> {
+    return this.serverManager.sendRequest({
+      method: 'POST',
+      route: `/api/input/${encodeURIComponent(inputRefIntoRawId(inputId))}/update`,
+      body: request,
+    });
+  }
+
   public async unregisterInput(
     inputId: InputRef,
     body: { schedule_time_ms?: number }

ts/smelter-core/src/event.ts

@@ -15,6 +15,8 @@ export function parseEvent(event: any, logger: Logger): SmelterEvent | null {
       SmelterEventType.AUDIO_INPUT_DELIVERED,
       SmelterEventType.VIDEO_INPUT_PLAYING,
       SmelterEventType.AUDIO_INPUT_PLAYING,
+      SmelterEventType.VIDEO_INPUT_PAUSED,
+      SmelterEventType.AUDIO_INPUT_PAUSED,
       SmelterEventType.VIDEO_INPUT_EOS,
       SmelterEventType.AUDIO_INPUT_EOS,
     ].includes(event.type)

ts/smelter-core/src/index.ts

@@ -8,3 +8,4 @@ export { OfflineSmelter } from './offline/compositor';
 export { SmelterManager, SetupInstanceOptions } from './smelterManager';
 export { Logger, LoggerLevel } from './logger';
 export { StateGuard } from './utils';
+export { InputHandle, WhipInputHandle, Mp4InputHandle } from './inputHandle';

ts/smelter-core/src/inputHandle.ts

@@ -0,0 +1,61 @@
+import type { ApiClient, RegisterInputResponse } from './api';
+import type { InputRef, RegisterInput } from './api/input';
+
+export function newInputHandle(
+  inputRef: InputRef,
+  api: ApiClient,
+  response: RegisterInputResponse,
+  kind: RegisterInput['type']
+) {
+  if (kind == 'whip_server') {
+    return new WhipInputHandle(inputRef, api, response);
+  } else if (kind == 'mp4') {
+    return new Mp4InputHandle(inputRef, api, response);
+  } else {
+    return new InputHandle(inputRef, api, response);
+  }
+}
+
+export class InputHandle {
+  protected inputRef: InputRef;
+  protected api: ApiClient;
+  protected registerResponse: RegisterInputResponse;
+
+  constructor(inputRef: InputRef, api: ApiClient, response: RegisterInputResponse) {
+    this.inputRef = inputRef;
+    this.api = api;
+    this.registerResponse = response;
+  }
+
+  public get videoDurationMs(): number | undefined {
+    return this.registerResponse.video_duration_ms;
+  }
+
+  public get audioDurationMs(): number | undefined {
+    return this.registerResponse.audio_duration_ms;
+  }
+
+  public async pause(): Promise<void> {
+    await this.api.updateInput(this.inputRef, { pause: true });
+  }
+
+  public async resume(): Promise<void> {
+    await this.api.updateInput(this.inputRef, { pause: false });
+  }
+}
+
+export class Mp4InputHandle extends InputHandle {
+  public async seek(seekMs: number): Promise<void> {
+    await this.api.updateInput(this.inputRef, { seek_ms: seekMs });
+  }
+}
+
+export class WhipInputHandle extends InputHandle {
+  public get endpointRoute(): string | undefined {
+    return this.registerResponse.endpoint_route;
+  }
+
+  public get bearerToken(): string | undefined {
+    return this.registerResponse.bearer_token;
+  }
+}