diff --git a/README.md b/README.md index e66a96d..8225230 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,8 @@ Add this entry to your OpenCode configuration. Open **Add tab > Device** in a session. First use downloads the pinned device helpers from npm. -Android SDK tools and Node must already be installed. Android video uses the device's H.264 encoder through scrcpy. +Android SDK tools and Node must already be installed. Both platforms stream H.264 through a shared viewer. +Android uses the device's encoder through scrcpy; iOS uses the simulator helper's native encoder. ## Use diff --git a/dist/gui.cjs b/dist/gui.cjs index ee878fd..cdc23ac 100644 --- a/dist/gui.cjs +++ b/dist/gui.cjs @@ -1,23 +1,23 @@ -"use strict";Object.defineProperties(exports,{__esModule:{value:!0},[Symbol.toStringTag]:{value:"Module"}});const X=require("solid-js/web"),re=require("solid-js"),Et=require("@opencode/plugin/gui"),Ma=require("@opencode/schema/rpc"),f=require("effect"),Ra=8*1024*1024,Ke=f.Schema.optional,Le=f.Schema.String.check(f.Schema.isMaxLength(2048)),zr=f.Schema.String.check(f.Schema.isMaxLength(1e5)),Lt=f.Schema.Int.check(f.Schema.isGreaterThanOrEqualTo(0)),Hr=f.Schema.Int.check(f.Schema.isBetween({minimum:0,maximum:1e6})),Je=f.Schema.Literals(["ios","android"]).annotate({identifier:"Device.Platform"}),ja=f.Schema.Literals(["shutdown","booting","booted","unavailable"]).annotate({identifier:"Device.DeviceState"}),Ua=f.Schema.Literals(["available","owned-by-this-host","owned-by-other","unknown"]).annotate({identifier:"Device.Ownership"}),Kr=f.Schema.Literals(["apps","launch","install","snapshot","press","fill","swipe","key","screenshot","stream"]).annotate({identifier:"Device.Capability"}),$a=f.Schema.Literals(["home","back","recents","lock","volumeUp","volumeDown"]).annotate({identifier:"Device.NavigationKey"}),In=Lt.annotate({identifier:"Device.Generation"}),Dn=f.Schema.String.check(f.Schema.isPattern(/^dh_[a-f0-9-]{36}$/u)).pipe(f.Schema.brand("Device.HandleToken")).annotate({identifier:"Device.HandleToken"}),Oe=f.Schema.Struct({token:Dn,platform:Je,deviceId:Le,generation:In}).annotate({identifier:"Device.DeviceHandle"}),ni=f.Schema.String.check(f.Schema.isPattern(/^@?e[1-9][0-9]*$/u)).pipe(f.Schema.brand("Device.ElementRef")).annotate({identifier:"Device.ElementRef"}),Jr=f.Schema.Struct({token:Dn,generation:In,revision:Lt,ref:ni}).annotate({identifier:"Device.SnapshotReference"}),Fa=f.Schema.Struct({ref:Ke(ni),role:Le,name:Ke(zr),value:Ke(zr),enabled:Ke(f.Schema.Boolean),focused:Ke(f.Schema.Boolean)}).annotate({identifier:"Device.AccessibilityNode"}),Wa=f.Schema.Struct({token:Dn,generation:In,revision:Lt,interactiveOnly:f.Schema.Boolean,complete:f.Schema.Boolean,nodes:f.Schema.Array(Fa)}).annotate({identifier:"Device.Snapshot"}),xn=f.Schema.Struct({x:Hr,y:Hr}).annotate({identifier:"Device.LogicalPoint"}),Va=f.Schema.Struct({id:Le,name:Le,platform:Je,osVersion:Ke(Le),state:ja,ownership:Ua}).annotate({identifier:"Device.DeviceInfo"}),Ga=f.Schema.Struct({supported:f.Schema.Array(Kr),unsupported:f.Schema.Array(f.Schema.Struct({capability:Kr,reason:Le}))}).annotate({identifier:"Device.Capabilities"}),qa=f.Schema.Struct({id:Le,name:Ke(Le)}).annotate({identifier:"Device.AppInfo"}),za=f.Schema.Struct({app:Le,state:f.Schema.Literals(["launched","foreground","unknown"])}).annotate({identifier:"Device.LaunchResult"}),Ha=f.Schema.Struct({installed:f.Schema.Boolean,app:Ke(Le)}).annotate({identifier:"Device.InstallResult"}),Ka=f.Schema.Struct({data:f.Schema.Uint8ArrayFromBase64.check(f.Schema.isMaxLength(Ra)),width:Lt,height:Lt,deviceId:Le,platform:Je}).annotate({identifier:"Device.Screenshot"}),Ja=f.Schema.Literals(["mjpeg","h264"]).annotate({identifier:"Device.StreamProtocol"}),Xa=f.Schema.Struct({platform:Je,deviceId:Le,baseUrl:Le,token:Le,protocol:Ja,loopback:f.Schema.Literal(!0)}).annotate({identifier:"Device.StreamDescriptor"}),Ya=f.Schema.Literals(["missing-prerequisite","permission-denied","device-busy","detached","stale-reference","unsupported","not-found","invalid-artifact","timeout","helper-failed","internal"]).annotate({identifier:"Device.DeviceErrorCode"}),Tn=f.Schema.Struct({code:Ya,message:Le,mayHaveExecuted:f.Schema.Boolean}).annotate({identifier:"Device.DeviceError"}),st=f.Schema.optional,je=f.Schema.String.check(f.Schema.isMaxLength(2048)),ri=je.check(f.Schema.isMinLength(1)),Ln={sessionID:ri,connectionID:f.Schema.String},Xr={...Ln,requestID:f.Schema.String},Yr=f.Schema.Number.check(f.Schema.isInt(),f.Schema.isBetween({minimum:1,maximum:1e6})),Qr=f.Schema.optional(f.Schema.Struct({width:Yr,height:Yr})),ct=f.Schema.Struct({sessionID:ri,serverID:je,location:f.Schema.optionalKey(f.Schema.Struct({directory:je}))}),zt=f.Schema.Struct({handle:Oe,selection:f.Schema.optionalKey(f.Schema.Struct({platform:Je,deviceId:je})),capabilities:Ga,agentAccess:f.Schema.Boolean}).annotate({identifier:"DeviceBridge.Attachment"}),ii=f.Schema.Union([f.Schema.Struct({type:f.Schema.Literal("list"),platform:st(Je)}),f.Schema.Struct({type:f.Schema.Literal("state")}),f.Schema.Struct({type:f.Schema.Literal("open"),platform:Je,deviceId:je,boot:st(f.Schema.Boolean)}),f.Schema.Struct({type:f.Schema.Literal("apps"),handle:Oe}),f.Schema.Struct({type:f.Schema.Literal("launch"),handle:Oe,app:je,deepLink:st(je),relaunch:st(f.Schema.Boolean)}),f.Schema.Struct({type:f.Schema.Literal("install"),handle:Oe,artifactPath:je}),f.Schema.Struct({type:f.Schema.Literal("snapshot"),handle:Oe,interactiveOnly:st(f.Schema.Boolean)}),f.Schema.Struct({type:f.Schema.Literal("press"),handle:Oe,reference:Jr}),f.Schema.Struct({type:f.Schema.Literal("tap"),handle:Oe,point:xn,viewport:Qr}),f.Schema.Struct({type:f.Schema.Literal("fill"),handle:Oe,reference:Jr,text:je}),f.Schema.Struct({type:f.Schema.Literal("swipe"),handle:Oe,from:xn,to:xn,viewport:Qr}),f.Schema.Struct({type:f.Schema.Literal("key"),handle:Oe,key:$a}),f.Schema.Struct({type:f.Schema.Literal("screenshot"),handle:Oe}),f.Schema.Struct({type:f.Schema.Literal("close"),handle:Oe,shutdown:st(f.Schema.Boolean)})]).pipe(f.Schema.toTaggedUnion("type")).annotate({identifier:"DeviceBridge.Operation"}),Qa=f.Schema.Union([f.Schema.Struct({type:f.Schema.Literal("devices"),devices:f.Schema.Array(Va)}),f.Schema.Struct({type:f.Schema.Literal("state"),state:f.Schema.NullOr(zt)}),f.Schema.Struct({type:f.Schema.Literal("attachment"),attachment:zt}),f.Schema.Struct({type:f.Schema.Literal("apps"),apps:f.Schema.Array(qa)}),f.Schema.Struct({type:f.Schema.Literal("launch"),result:za}),f.Schema.Struct({type:f.Schema.Literal("install"),result:Ha}),f.Schema.Struct({type:f.Schema.Literal("snapshot"),snapshot:Wa}),f.Schema.Struct({type:f.Schema.Literal("screenshot"),screenshot:Ka}),f.Schema.Struct({type:f.Schema.Literal("done")})]).pipe(f.Schema.toTaggedUnion("type")).annotate({identifier:"DeviceBridge.Value"}),qt=f.Schema.Union([f.Schema.Struct({type:f.Schema.Literal("success"),value:Qa}),f.Schema.Struct({type:f.Schema.Literal("failure"),error:Tn})]).pipe(f.Schema.toTaggedUnion("type")).annotate({identifier:"DeviceBridge.Outcome"}),Za=f.Schema.Union([f.Schema.Struct({type:f.Schema.Literal("success"),stream:Xa}),f.Schema.Struct({type:f.Schema.Literal("failure"),error:Tn})]).pipe(f.Schema.toTaggedUnion("type")).annotate({identifier:"DeviceBridge.StreamOutcome"}),es=f.Schema.Union([f.Schema.Struct({type:f.Schema.Literal("attached"),connectionID:f.Schema.String,version:f.Schema.Literal(1)}),f.Schema.Struct({type:f.Schema.Literal("command"),connectionID:f.Schema.String,requestID:f.Schema.String}),f.Schema.Struct({type:f.Schema.Literal("cancel"),connectionID:f.Schema.String,requestID:f.Schema.String})]).pipe(f.Schema.toTaggedUnion("type")).annotate({identifier:"DeviceBridge.Control"}),Me=f.Schema.toStandardSchemaV1,Gt={unavailable:Me(Tn)};Ma.Rpc.define({id:"device.control",methods:{attach:{input:Me(f.Schema.Struct({...Ln,version:f.Schema.Literal(1)})),output:Me(f.Schema.Literals(["closed","replaced"])),errors:Gt},state:{input:Me(f.Schema.Struct({...Ln,state:f.Schema.NullOr(zt)})),output:Me(f.Schema.Void),errors:Gt},command:{input:Me(f.Schema.Struct(Xr)),output:Me(ii),errors:Gt},result:{input:Me(f.Schema.Struct({...Xr,outcome:qt})),output:Me(f.Schema.Void),errors:Gt}},events:{control:{schema:Me(es)}}});const Zr=f.Schema.String.check(f.Schema.isMaxLength(2048)),ts=Et.Ipc.define({id:"demul.device.bridge.v2",state:f.Schema.Array(f.Schema.Struct({scope:ct,state:f.Schema.NullOr(zt)})),methods:{connect:{input:ct},disconnect:{input:ct},list:{input:f.Schema.Struct({platform:f.Schema.optional(Je),serverID:Zr,sessionID:f.Schema.optionalKey(Zr)}),output:qt},access:{input:f.Schema.Struct({scope:ct,enabled:f.Schema.Boolean}),output:qt},manual:{input:f.Schema.Struct({scope:ct,operation:ii}),output:qt},stream:{input:f.Schema.Struct({scope:ct,handle:Oe}),output:Za}}});function ei(h){let I;return{async run(E,A){I?.abort();const H=new AbortController;I=H,h.pending(!0);try{const n=await E(H.signal);I===H&&A(n)}catch(n){I===H&&!H.signal.aborted&&h.error(n)}finally{I===H&&(I=void 0,h.pending(!1))}},cancel(){I?.abort(),I=void 0,h.pending(!1)}}}function ns(h,I,E,A){if(!(!h||h!==I||E!=="playing"||!Number.isSafeInteger(A.width)||!Number.isSafeInteger(A.height)||A.width<=0||A.height<=0))return{token:h,width:A.width,height:A.height}}function rs(h,I){return!!I&&h.token===I.token&&h.width===I.width&&h.height===I.height}const is=1397050709,os={1:16,2:24},oi=16*1024*1024;function as(h){let I=!1,E;const A=()=>{try{return h.stream.protocol==="h264"?ms(h):gs(h)}catch(ee){return h.onState("paused"),h.onError(ee instanceof Error?ee.message:String(ee)),()=>{}}},H=()=>{const ee=I&&document.visibilityState==="visible";if(ee&&!E){E=A();return}ee||!E||(E(),E=void 0,h.onState(I?"paused":"idle"))},n=()=>H();return document.addEventListener("visibilitychange",n),{active(ee){I=ee,H()},destroy(){I=!1,document.removeEventListener("visibilitychange",n),E?.(),E=void 0,h.canvas.width=0,h.canvas.height=0,h.onState("idle")}}}function ss(h){const I=new Uint8Array(h);if(I.byteLength<4)return;const E=new DataView(h);return E.getUint32(0)===is?cs(I,E):ai(I,performance.now()*1e3,!1)}function cs(h,I){if(h.byteLength<16)return;const E=os[I.getUint8(4)];if(!E||h.byteLength<=E)return;const A=I.getBigUint64(8);if(!(A>BigInt(Number.MAX_SAFE_INTEGER)))return ai(h.subarray(E),Number(A),(I.getUint8(5)&1)===1,E===24?Number(I.getBigUint64(16))/1e3:void 0)}function ai(h,I,E,A){const H=li(h,0);if(H===void 0||H>=h.byteLength)return;const n=ys(h);return{bytes:h,key:E||n.key,codec:n.codec,timestamp:I,serverTimestamp:A}}function ls(h,I,E){if(![h.left,h.top,h.width,h.height,I.width,I.height,E.x,E.y].every(Number.isFinite)||h.width<=0||h.height<=0||!ci(I))return;const A=Math.min(h.width/I.width,h.height/I.height),H=I.width*A,n=I.height*A,ee=E.x-h.left-(h.width-H)/2,q=E.y-h.top-(h.height-n)/2;if(!(ee<0||q<0||ee>=H||q>=n))return{x:Math.min(I.width-1,Math.floor(ee/A)),y:Math.min(I.height-1,Math.floor(q/A))}}function ds(h){if(!h||/[\r\n]/.test(h))return;const I=h.split(";");if(I.shift()?.trim().toLowerCase()!=="multipart/x-mixed-replace")return;const E=I.map(n=>n.trim()).find(n=>n.toLowerCase().startsWith("boundary="))?.slice(9).trim();if(!E)return;const A=E.startsWith('"')&&E.endsWith('"')?E.slice(1,-1):E,H=A.startsWith("--")?A.slice(2):A;if(H)return H}function hs(h,I){const E=new TextEncoder().encode(`--${I}`),A=new Uint8Array([13,10,13,10]),H=[];let n=0,ee=0;for(;nh.byteLength)return{frames:H,rest:h.slice(q)};if(!us(h,K)){n=K;continue}const ae=ti(h,A,K+2);if(ae<0)return{frames:H,rest:h.slice(q)};const J=new TextDecoder("ascii").decode(h.subarray(K+2,ae)),ue=fs(J),me=ae+A.byteLength;if(!ue){n=ae+A.byteLength;continue}const fe=me+ue;if(fe>h.byteLength)return{frames:H,rest:h.slice(q)};const Ie=h.subarray(me,fe);ps(Ie)&&H.push(Ie),ee=fe,n=fe}return{frames:H,rest:h.slice(ee)}}function us(h,I){return h[I]===13&&h[I+1]===10}function fs(h){const I=h.match(/(?:^|\r\n)Content-Length:\s*(\d+)(?:\r\n|$)/i)?.[1],E=h.match(/(?:^|\r\n)Content-Type:\s*([^\r\n]+)(?:\r\n|$)/i)?.[1].trim().toLowerCase(),A=I?Number(I):0;if(Number.isSafeInteger(A)&&A>0&&A<=oi&&E==="image/jpeg")return A}function ps(h){return h.byteLength>=4&&h[0]===255&&h[1]===216&&h.at(-2)===255&&h.at(-1)===217}function ms(h){if(typeof VideoDecoder!="function"||typeof EncodedVideoChunk!="function")return h.onState("paused"),h.onError("H.264 device video requires WebCodecs support in this browser"),()=>{};const I=new URL("/video?frame-meta=1",new URL(h.stream.baseUrl).origin);I.protocol=I.protocol==="https:"?"wss:":"ws:";const E=new WebSocket(I,`opencode-device.${h.stream.token}`);let A,H,n=0,ee=!0,q=!1,K=!1,ae;const J=()=>{clearTimeout(ae),ae=setTimeout(()=>fe("Android video did not produce a screen within 15 seconds. Refresh devices to retry."),15e3)};E.binaryType="arraybuffer",h.onState("connecting");const ue=()=>{if(E.readyState===WebSocket.OPEN)try{E.send(JSON.stringify({type:"reset-video",ack:!1}))}catch{}},me=(ne=!1)=>{n++;const de=A,ye=K;A=void 0,H=void 0,ee=!0,K=!1,de&&de.state!=="closed"&&de.close(),ne&&ye&&!q&&(J(),h.onState("connecting")),ne&&ue()},fe=ne=>{q||(q=!0,clearTimeout(ae),E.close(),me(),h.onError(ne.replaceAll(h.stream.token,"[redacted]").slice(0,512)),h.onState("paused"))},Ie=(ne,de)=>{if(!(q||A!==de||de.state!=="configured"))try{de.decode(new EncodedVideoChunk({type:ne.key?"key":"delta",timestamp:ne.timestamp,data:ne.bytes})),ee=!1}catch(ye){fe(`H.264 decode failed: ${ye instanceof Error?ye.message:String(ye)}`)}},Xe=(ne,de)=>{me(),H=de;const ye=n,Ue={codec:de,optimizeForLatency:!0,hardwareAcceleration:"prefer-hardware"};VideoDecoder.isConfigSupported(Ue).then(ke=>{if(q||ye!==n)return;if(!ke.supported){fe(`H.264 codec ${de} is not supported by this browser`);return}const be=new VideoDecoder({output(_e){if(q||A!==be){_e.close();return}try{si(h,_e,_e.displayWidth,_e.displayHeight),K||(K=!0,clearTimeout(ae),h.onState("playing"))}catch($e){fe(`Device video draw failed: ${$e instanceof Error?$e.message:String($e)}`)}finally{_e.close()}},error(_e){q||A!==be||fe(`H.264 decoder failed: ${_e.message}`)}});try{be.configure(ke.config??Ue)}catch(_e){be.close(),fe(`H.264 decoder configuration failed: ${_e instanceof Error?_e.message:String(_e)}`);return}A=be,Ie(ne,be)}).catch(ke=>{q||ye!==n||fe(`H.264 decoder setup failed: ${ke instanceof Error?ke.message:String(ke)}`)})},Be=ne=>{if(A&&A.decodeQueueSize>2)return me(!0);if(ee&&!ne.key)return;const de=ne.codec??H;if(!de)return ue();if(A&&de===H)return Ie(ne,A);Xe(ne,de)};return J(),E.addEventListener("open",ue),E.addEventListener("message",ne=>{if(q)return;if(typeof ne.data=="string"){const ye=Ss(ne.data);ye?.type==="error"&&fe(`Android video failed: ${ye.message}`),ye?.type==="reset"&&me(!0);return}if(!(ne.data instanceof ArrayBuffer))return;const de=ss(ne.data);de&&Be(de)}),E.addEventListener("error",()=>fe("Device video connection failed")),E.addEventListener("close",ne=>{fe(`Device video connection closed (code ${ne.code}). Refresh devices to retry.`)}),()=>{q=!0,clearTimeout(ae),E.close(),me()}}function gs(h){const I=new AbortController;let E,A=!1,H=!1;const n=async ee=>{const q=await createImageBitmap(new Blob([ee],{type:"image/jpeg"}));try{if(A)return;si(h,q,q.width,q.height),H||(H=!0,h.onState("playing"))}finally{q.close()}};return h.onState("connecting"),fetch(new URL("/video",new URL(h.stream.baseUrl).origin),{method:"GET",headers:{Authorization:`Bearer ${h.stream.token}`},signal:I.signal}).then(async ee=>{if(!ee.ok||!ee.body)throw new Error(`Device video returned ${ee.status}`);const q=ds(ee.headers.get("content-type"));if(!q)throw new Error("Device video returned an invalid MJPEG content type");E=ee.body.getReader();let K=new Uint8Array;for(;!A;){const ae=await E.read();if(ae.done)break;K=bs(K,ae.value);const J=hs(K,q);K=J.rest;const ue=J.frames.at(-1);if(ue&&await n(ue),K.byteLength>oi)throw new Error("Device MJPEG stream exceeded 16 MiB without a complete frame")}if(!A)throw new Error("Device video stream ended. Refresh devices to retry.")}).catch(ee=>{A||(h.onState("paused"),h.onError(ee instanceof Error?ee.message:String(ee)))}).finally(async()=>{I.abort(),await E?.cancel().catch(()=>{}),E?.releaseLock(),E=void 0}),()=>{A=!0,I.abort(),E?.cancel().catch(()=>{})}}function si(h,I,E,A){if(!Number.isSafeInteger(E)||!Number.isSafeInteger(A)||E<=0||A<=0)throw new Error("Device video produced invalid screen dimensions");(h.canvas.width!==E||h.canvas.height!==A)&&(h.canvas.width=E,h.canvas.height=A);const H=h.canvas.getContext("2d",{alpha:!1});if(!H)throw new Error("Device video canvas is unavailable");H.drawImage(I,0,0,E,A)}function Ss(h){try{const I=JSON.parse(h);if(!I||typeof I!="object")return;if("ok"in I&&I.ok===!1)return{type:"error",message:"error"in I&&typeof I.error=="string"?I.error:"the capture session failed"};if(vs(I))return{type:"reset"}}catch{return}}function vs(h){return"type"in h&&h.type==="video-session"&&"size"in h&&h.size!==null&&typeof h.size=="object"&&ci(h.size)}function ci(h){return!("width"in h)||!("height"in h)?!1:typeof h.width=="number"&&typeof h.height=="number"&&Number.isSafeInteger(h.width)&&Number.isSafeInteger(h.height)&&h.width>0&&h.height>0}function ys(h){let I=!1;for(let E=0;EI.toString(16).padStart(2,"0")).join("")}`}function bs(h,I){const E=new Uint8Array(h.byteLength+I.byteLength);return E.set(h),E.set(I,h.byteLength),E}function ti(h,I,E){for(let A=E;A<=h.byteLength-I.byteLength;A++){let H=!0;for(let n=0;n"),xs=X.template(" + + + + + + void outcome(() => bridge.access({ scope: scope(), enabled }))} + /> + + + ); + return (
@@ -427,65 +479,7 @@ export default Plugin.define({
- -
-
- - {(platform) => ( - - )} - -
- - - - - - - void outcome(() => bridge.access({ scope: scope(), enabled })) - } - /> - -
+
); }; diff --git a/src/h264.test.ts b/src/h264.test.ts new file mode 100644 index 0000000..314b8a5 --- /dev/null +++ b/src/h264.test.ts @@ -0,0 +1,97 @@ +import { expect, test } from "bun:test"; +import { H264Configuration, parseFramePacket } from "./h264.js"; +import { sps, pps, idr, delta, keyframe, metadataFrame } from "../test/video.js"; + +test.each([sps, pps, new Uint8Array([...sps, ...pps])])( + "parameter sets are not keyframes or pictures", + (data) => { + expect(parseFramePacket(data)).toMatchObject({ key: false, picture: false }); + }, +); + +test("an SPS before a delta does not turn the delta into a keyframe", () => { + expect(parseFramePacket(new Uint8Array([...sps, ...delta]))).toMatchObject({ + key: false, + picture: true, + codec: "avc1.640029", + }); +}); + +test("an IDR is detected independently of SPS ordering and start-code length", () => { + for (const data of [ + idr, + new Uint8Array([...sps, ...pps, ...idr]), + new Uint8Array([...idr, ...sps]), + ]) + expect(parseFramePacket(data)).toMatchObject({ key: true, picture: true }); +}); + +test("codec detection reads an SPS across mixed start-code boundaries", () => { + expect(parseFramePacket(new Uint8Array([0, ...pps, ...sps.subarray(1), ...idr]))).toMatchObject({ + codec: "avc1.640029", + }); +}); + +test("a metadata key flag never turns configuration-only data into a picture", () => { + const data = metadataFrame(1, 1, 0n, sps); + expect(parseFramePacket(data)).toMatchObject({ key: false, picture: false }); +}); + +test("typed-array views exclude bytes outside their byte offset and length", () => { + const pooled = new Uint8Array(100); + pooled.set(new Uint8Array([...sps, ...idr]), 11); + const view = pooled.subarray(11, 11 + sps.byteLength + idr.byteLength); + expect(parseFramePacket(view)).toMatchObject({ bytes: view, key: true, codec: "avc1.640029" }); +}); + +test("separate parameter sets are copied and prepended to an IDR, never to a delta", () => { + const configuration = new H264Configuration(); + const mutable = sps.slice(); + configuration.update(parseFramePacket(mutable)!); + configuration.update(parseFramePacket(pps)!); + mutable.fill(0); + expect(configuration.codec).toBe("avc1.640029"); + expect(configuration.bytes(parseFramePacket(idr)!)).toEqual( + new Uint8Array([...sps, 0, 0, 0, 1, 0x68, 0xeb, ...idr]), + ); + const packet = parseFramePacket(delta)!; + expect(configuration.bytes(packet)).toBe(packet.bytes); + const complete = parseFramePacket(new Uint8Array([...sps, ...pps, ...idr]))!; + expect(configuration.bytes(complete)).toBe(complete.bytes); + configuration.clear(); + expect(configuration.codec).toBeUndefined(); + const bare = parseFramePacket(idr)!; + expect(configuration.bytes(bare)).toBe(bare.bytes); +}); + +test("raw Annex-B packets use a local timestamp", () => { + const packet = parseFramePacket(keyframe.buffer)!; + expect(packet.bytes).toEqual(keyframe); + expect(packet.key).toBe(true); + expect(packet.timestamp).toBeGreaterThan(0); +}); + +test.each([1, 2])("SEMU v%i preserves timestamps and excludes its metadata header", (version) => { + const data = metadataFrame(version, 0, 123_456n, keyframe, 1_735_689_600_123_000n); + expect(parseFramePacket(data.buffer)).toEqual({ + bytes: keyframe, + key: true, + picture: true, + codec: "avc1.640029", + sps: sps.subarray(4), + pps: undefined, + timestamp: 123_456, + serverTimestamp: version === 2 ? 1_735_689_600_123 : undefined, + }); +}); + +test.each([ + new Uint8Array(), + new Uint8Array([0, 0, 0, 1]), + metadataFrame(3, 0, 1n, keyframe), + metadataFrame(2, 0, 1n, new Uint8Array()), + metadataFrame(1, 0, BigInt(Number.MAX_SAFE_INTEGER) + 1n, keyframe), + new Uint8Array([0x53, 0x45, 0x4d, 0x55, 1]), +])("rejects malformed framing", (data) => { + expect(parseFramePacket(data.buffer)).toBeUndefined(); +}); diff --git a/src/h264.ts b/src/h264.ts new file mode 100644 index 0000000..19dabc6 --- /dev/null +++ b/src/h264.ts @@ -0,0 +1,126 @@ +export interface FramePacket { + readonly bytes: Uint8Array; + readonly key: boolean; + readonly picture: boolean; + readonly codec?: string; + readonly sps?: Uint8Array; + readonly pps?: Uint8Array; + readonly timestamp: number; + readonly serverTimestamp?: number; +} + +const MAGIC = 0x53454d55; +const FRAME_HEADERS: Readonly> = { 1: 16, 2: 24 }; +const START_CODE = new Uint8Array([0, 0, 0, 1]); + +// Accept typed-array views as well as browser ArrayBuffers without copying pooled Node buffers. +export function parseFramePacket(input: ArrayBuffer | Uint8Array): FramePacket | undefined { + const data = input instanceof Uint8Array ? input : new Uint8Array(input); + if (data.byteLength < 4) return; + const view = new DataView(data.buffer, data.byteOffset, data.byteLength); + if (view.getUint32(0) === MAGIC) return metadataFrame(data, view); + return videoFrame(data, performance.now() * 1_000, false); +} + +function metadataFrame(data: Uint8Array, view: DataView): FramePacket | undefined { + if (data.byteLength < 16) return; + const header = FRAME_HEADERS[view.getUint8(4)]; + if (!header || data.byteLength <= header) return; + const pts = view.getBigUint64(8); + if (pts > BigInt(Number.MAX_SAFE_INTEGER)) return; + return videoFrame( + data.subarray(header), + Number(pts), + (view.getUint8(5) & 1) === 1, + header === 24 ? Number(view.getBigUint64(16)) / 1_000 : undefined, + ); +} + +function videoFrame( + bytes: Uint8Array, + timestamp: number, + key: boolean, + serverTimestamp?: number, +): FramePacket | undefined { + const start = nalStart(bytes, 0); + if (start === undefined || start >= bytes.byteLength) return; + const info = annexBInfo(bytes); + return { bytes, ...info, key: info.key || (info.picture && key), timestamp, serverTimestamp }; +} + +function annexBInfo(data: Uint8Array) { + let key = false; + let picture = false; + let sps: Uint8Array | undefined; + let pps: Uint8Array | undefined; + for (const nal of nals(data)) { + const type = nal[0] & 31; + if (type === 5) key = true; + if (type >= 1 && type <= 5) picture = true; + if (type === 7) sps = nal; + if (type === 8) pps = nal; + } + return { key, picture, codec: sps && spsCodec(sps), sps, pps }; +} + +// Yield NAL bodies, excluding either form of Annex-B start code. +function* nals(data: Uint8Array): Generator { + let start = nalStart(data, 0); + if (start === undefined) return; + for (let index = start; index < data.byteLength - 2; index++) { + const next = nalStart(data, index); + if (next === undefined) continue; + if (index > start) yield data.subarray(start, index); + start = next; + index = next - 1; + } + if (start < data.byteLength) yield data.subarray(start); +} + +function nalStart(data: Uint8Array, index: number): number | undefined { + if (data[index] !== 0 || data[index + 1] !== 0) return; + if (data[index + 2] === 1) return index + 3; + if (data[index + 2] === 0 && data[index + 3] === 1) return index + 4; +} + +function spsCodec(sps: Uint8Array): string | undefined { + if (sps.byteLength < 4) return; + return `avc1.${[sps[1], sps[2], sps[3]].map((byte) => byte.toString(16).padStart(2, "0")).join("")}`; +} + +export class H264Configuration { + codec: string | undefined; + private sps: Uint8Array | undefined; + private pps: Uint8Array | undefined; + + update(frame: FramePacket): void { + if (frame.sps) { + this.sps = frame.sps.slice(); + this.codec = frame.codec; + } + if (frame.pps) this.pps = frame.pps.slice(); + } + + // Android may send parameter sets separately. Supply them with the IDR, never as a fake keyframe. + bytes(frame: FramePacket): Uint8Array { + if (!frame.key) return frame.bytes; + const parts: Uint8Array[] = []; + if (!frame.sps && this.sps) parts.push(START_CODE, this.sps); + if (!frame.pps && this.pps) parts.push(START_CODE, this.pps); + if (!parts.length) return frame.bytes; + parts.push(frame.bytes); + const result = new Uint8Array(parts.reduce((size, part) => size + part.byteLength, 0)); + let offset = 0; + for (const part of parts) { + result.set(part, offset); + offset += part.byteLength; + } + return result; + } + + clear(): void { + this.codec = undefined; + this.sps = undefined; + this.pps = undefined; + } +} diff --git a/src/main.test.ts b/src/main.test.ts index c6369ed..fd1a0c4 100644 --- a/src/main.test.ts +++ b/src/main.test.ts @@ -83,7 +83,7 @@ function fixture() { deviceId: handle.deviceId, baseUrl: "http://127.0.0.1:3400", token: streamToken, - protocol: "mjpeg", + protocol: "h264", loopback: true, }; }, diff --git a/src/main.ts b/src/main.ts index daa427e..737c76b 100644 --- a/src/main.ts +++ b/src/main.ts @@ -25,7 +25,7 @@ interface Connection { requests: Map; stopEvents: () => void; closed: boolean; - proxy?: { token: string; pending: Promise }; + proxy?: { token: string; sourceToken: string; pending: Promise }; } export function createMainSetup( @@ -316,11 +316,16 @@ export function createMainSetup( return deviceStreamOutcome(async () => { const raw = await getController().stream(caller(connection), input.handle); requireStream(connection, input, ipcCaller.window, ipcCaller.signal); - if (connection.proxy && connection.proxy.token !== input.handle.token) + if ( + connection.proxy && + (connection.proxy.token !== input.handle.token || + connection.proxy.sourceToken !== raw.token) + ) await stopProxy(connection); if (!connection.proxy) { const proxy = { token: input.handle.token, + sourceToken: raw.token, pending: startStreamProxy(raw), }; connection.proxy = proxy; diff --git a/src/native/errors.ts b/src/native/errors.ts index ba94019..d051fc8 100644 --- a/src/native/errors.ts +++ b/src/native/errors.ts @@ -31,3 +31,14 @@ export function deviceError( export function isDeviceFailure(value: unknown): value is DeviceFailure { return value instanceof DeviceFailure; } + +export function cancellationError(signal: AbortSignal, mayHaveExecuted = false): DeviceFailure { + const timedOut = signal.reason instanceof Error && signal.reason.name === "TimeoutError"; + return deviceError( + timedOut ? "timeout" : "detached", + timedOut + ? "The device operation exceeded its deadline." + : "The device operation was cancelled.", + mayHaveExecuted, + ); +} diff --git a/src/native/hub.ts b/src/native/hub.ts index 57dc2e4..8f3f26d 100644 --- a/src/native/hub.ts +++ b/src/native/hub.ts @@ -1,5 +1,5 @@ import { Device } from "../rpc.js"; -import { deviceError } from "./errors.js"; +import { cancellationError, deviceError, isDeviceFailure } from "./errors.js"; import { spawnProcess, stopProcess, type LongRunningProcess } from "./process.js"; import { Option, Schema } from "effect"; import { parseJson } from "./json.js"; @@ -9,6 +9,23 @@ export async function sendGesture( gesture: | { type: "tap"; x: number; y: number } | { type: "swipe"; x1: number; y1: number; x2: number; y2: number; durationMs: number }, + signal?: AbortSignal, +): Promise { + signal?.throwIfAborted(); + const requestSignal = AbortSignal.any([AbortSignal.timeout(10_000), ...(signal ? [signal] : [])]); + try { + await postGesture(stream, gesture, requestSignal); + } catch (error) { + if (isDeviceFailure(error)) throw error; + if (requestSignal.aborted) throw cancellationError(requestSignal, true); + throw deviceError("helper-failed", "The device gesture connection failed.", true); + } +} + +async function postGesture( + stream: Device.StreamDescriptor, + gesture: Parameters[1], + signal: AbortSignal, ): Promise { const url = new URL(`/vendor/serve-emu/api/${gesture.type}`, stream.baseUrl); url.searchParams.set("device", stream.deviceId); @@ -17,7 +34,7 @@ export async function sendGesture( headers: { Authorization: `Bearer ${stream.token}`, "Content-Type": "application/json" }, body: JSON.stringify(gesture), redirect: "error", - signal: AbortSignal.timeout(10_000), + signal, }); const result = parseJson( Schema.Struct({ ok: Schema.Boolean, error: Schema.optionalKey(Schema.String) }), @@ -46,6 +63,8 @@ export interface HubOptions { stopKillMs?: number; env?: Record; signal?: AbortSignal; + // Preserve a stop that timed out before startup completed, so the owning runtime can retry it. + onCleanupFailure?: (stop: () => Promise) => void; } // The hub's ready endpoint: a loopback origin and the session token minted by `--require-token`. @@ -57,11 +76,6 @@ export interface HubReady { // The loopback host the hub binds to. The stream never leaves the machine. export const LOOPBACK_HOST = "127.0.0.1"; -// Android's serve-emu always streams H.264 and iOS's serve-sim streams MJPEG, whatever `--transport` says. -function streamProtocolFor(platform: Device.Platform): Device.StreamProtocol { - return platform === "ios" ? "mjpeg" : "h264"; -} - export function hubArgs(options: HubOptions): string[] { return [ options.hubCli, @@ -69,7 +83,7 @@ export function hubArgs(options: HubOptions): string[] { LOOPBACK_HOST, "--require-token", "--transport", - streamProtocolFor(options.platform), + "h264", "--platform", options.platform, "--single-device", @@ -98,7 +112,7 @@ export function streamDescriptor( deviceId, baseUrl: ready.baseUrl, token: ready.token, - protocol: streamProtocolFor(platform), + protocol: "h264", loopback: true, }; } @@ -115,18 +129,14 @@ export function startHub(options: HubOptions, spawn = spawnProcess): Promise((resolve, reject) => { let settled = false; - let ready: HubReady | null = null; const process = spawn({ command: options.externalNode, args: hubArgs(options), env: options.env, onStdoutLine: (line) => { - if (ready) return; + if (settled) return; const parsed = parseReady(line); - if (parsed) { - ready = parsed; - finishReady(parsed); - } + if (parsed) finishReady(parsed); }, }); const timer = setTimeout(() => { @@ -142,7 +152,13 @@ export function startHub(options: HubOptions, spawn = spawnProcess): Promise finish(() => { - void stop().then(() => reject(error), reject); + void stop().then( + () => reject(error), + (cleanupError) => { + options.onCleanupFailure?.(stop); + reject(cleanupError); + }, + ); }); const abort = () => fail(options.signal?.reason); options.signal?.addEventListener("abort", abort, { once: true }); diff --git a/src/native/native.test.ts b/src/native/native.test.ts index 8793fe1..bbac0a4 100644 --- a/src/native/native.test.ts +++ b/src/native/native.test.ts @@ -223,6 +223,27 @@ test("native commands leave the event loop responsive and cancellation kills the ).rejects.toBe(reason); }); +test("failed process stops are shared while pending but can be retried after the deadline", async () => { + const exited = Promise.withResolvers<{ code: number | null; signal: NodeJS.Signals | null }>(); + const signals: NodeJS.Signals[] = []; + const child: LongRunningProcess = { + pid: undefined, + exited: exited.promise, + stop: (signal = "SIGTERM") => { + signals.push(signal); + }, + }; + const first = stopProcess(child, 1, 1); + expect(stopProcess(child, 1, 1)).toBe(first); + await expect(first).rejects.toMatchObject({ code: "timeout" }); + const retry = stopProcess(child, 1, 1); + expect(retry).not.toBe(first); + exited.resolve({ code: 0, signal: null }); + await retry; + expect(signals).toEqual(["SIGTERM", "SIGKILL", "SIGTERM"]); + expect(stopProcess(child)).toBe(retry); +}); + test("cancelling Android startup terminates only the emulator it spawned", async () => { const abort = new AbortController(); const exited = Promise.withResolvers<{ code: number | null; signal: NodeJS.Signals | null }>(); @@ -750,6 +771,78 @@ test("hub startup rollback awaits bounded forced termination", async () => { expect(signals).toEqual(["SIGTERM", "SIGKILL"]); }); +test("a startup cleanup timeout retains a retryable stop rather than losing the helper", async () => { + const exited = Promise.withResolvers<{ code: number | null; signal: NodeJS.Signals | null }>(); + let retained: (() => Promise) | undefined; + const signals: NodeJS.Signals[] = []; + await expect( + startHub( + { + externalNode: "fixture-node", + hubCli: "fixture-hub", + platform: "android", + deviceId: "fixture-device", + readyTimeoutMs: 1, + stopGraceMs: 1, + stopKillMs: 1, + onCleanupFailure: (stop) => { + retained = stop; + }, + }, + () => ({ + pid: undefined, + exited: exited.promise, + stop: (signal = "SIGTERM") => { + signals.push(signal); + }, + }), + ), + ).rejects.toMatchObject({ code: "timeout" }); + expect(retained).toBeDefined(); + const retry = retained!(); + exited.resolve({ code: 0, signal: null }); + await retry; + expect(signals).toEqual(["SIGTERM", "SIGKILL", "SIGTERM"]); +}); + +test.each(["AbortError", "TimeoutError"])( + "Android gesture cancellation during response capture preserves execution uncertainty: %s", + async (name) => { + const received = Promise.withResolvers(); + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch() { + received.resolve(); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("{")); + }, + }), + ); + }, + }); + const stream = streamDescriptor( + { baseUrl: server.url.origin, token: "fixture-secret" }, + "android", + "phone", + ); + try { + const abort = new AbortController(); + const pending = sendGesture(stream, { type: "tap", x: 0.5, y: 0.5 }, abort.signal); + const outcome = pending.catch((error: unknown) => error); + await received.promise; + abort.abort(new DOMException("cancelled", name)); + expect(await outcome).toMatchObject({ + data: { code: name === "TimeoutError" ? "timeout" : "detached", mayHaveExecuted: true }, + }); + } finally { + await server.stop(true); + } + }, +); + test("unobserved process exit releases resources and reports a bounded timeout", async () => { const signals: NodeJS.Signals[] = []; let released = false; @@ -825,7 +918,7 @@ test("an explicit unsupported Node runtime is not replaced by automatic discover }); }); -test("hub arg builder pins host, token, the platform's served transport, platform, and single device", () => { +test("hub arg builder pins host, token, H.264, platform, and single device", () => { const android = hubArgs({ externalNode: "node", hubCli: "/h/cli.js", @@ -852,8 +945,9 @@ test("hub arg builder pins host, token, the platform's served transport, platfor platform: "ios", deviceId: "UDID", }); - expect(ios).toContain("mjpeg"); - expect(ios).not.toContain("h264"); + expect(ios).toContain("h264"); + expect(ios).not.toContain("mjpeg"); + expect(ios).not.toContain("scrcpy"); }); test("parseReady extracts a loopback origin and token", () => { diff --git a/src/native/process.ts b/src/native/process.ts index 2e9fe82..e0deaf1 100644 --- a/src/native/process.ts +++ b/src/native/process.ts @@ -132,6 +132,9 @@ export function stopProcess( }; const result = stop(); stopping.set(process, result); + void result.catch(() => { + if (stopping.get(process) === result) stopping.delete(process); + }); return result; } diff --git a/src/native/serve-sim.test.ts b/src/native/serve-sim.test.ts index b25f63d..30246f5 100644 --- a/src/native/serve-sim.test.ts +++ b/src/native/serve-sim.test.ts @@ -1,6 +1,6 @@ import { afterEach, expect, test } from "bun:test"; import { createServer, type Server } from "node:http"; -import type { AddressInfo } from "node:net"; +import { createServer as createTcpServer, type AddressInfo, type Socket } from "node:net"; import { WebSocketServer, type RawData } from "ws"; import type { Device } from "../rpc.js"; import { @@ -26,7 +26,7 @@ const descriptor = (baseUrl: string): Device.StreamDescriptor => ({ deviceId: "UDID-1", baseUrl, token: HUB_TOKEN, - protocol: "mjpeg", + protocol: "h264", loopback: true, }); @@ -126,8 +126,6 @@ test("sendTouches authenticates the device-scoped HID socket and sends binary me const baseUrl = await listen(server); const received: Array<{ data: Buffer; binary: boolean }> = []; const closed = Promise.withResolvers(); - let lastMessageAt = 0; - let closedAt = 0; let offeredProtocols: string[] = []; let requestUrl: string | undefined; let selectedProtocol: string | undefined; @@ -145,16 +143,14 @@ test("sendTouches authenticates the device-scoped HID socket and sends binary me selectedProtocol = socket.protocol; socket.on("message", (data, binary) => { received.push({ data: toBuffer(data), binary }); - lastMessageAt = performance.now(); - }); - socket.on("close", () => { - closedAt = performance.now(); - closed.resolve(); }); + socket.on("close", () => closed.resolve()); }); const events = swipeEvents({ x: 0.25, y: 0.75 }, { x: 0.75, y: 0.25 }, 4); + const startedAt = performance.now(); await sendTouches(descriptor(baseUrl), events, 1, 1_000); + const elapsed = performance.now() - startedAt; await closed.promise; const url = new URL(requestUrl!, baseUrl); @@ -163,7 +159,7 @@ test("sendTouches authenticates the device-scoped HID socket and sends binary me expect(offeredProtocols).toEqual([`serve-sim.token.${HUB_TOKEN}`]); expect(selectedProtocol).toBe(`serve-sim.token.${HUB_TOKEN}`); expect(received).toEqual(events.map((event) => ({ data: encodeTouch(event), binary: true }))); - expect(closedAt - lastMessageAt).toBeGreaterThanOrEqual(490); + expect(elapsed).toBeGreaterThanOrEqual(490); }); test.each(buttonCases)( @@ -173,8 +169,6 @@ test.each(buttonCases)( const baseUrl = await listen(server); const received: Array<{ data: Buffer; binary: boolean }> = []; const closed = Promise.withResolvers(); - let lastMessageAt = 0; - let closedAt = 0; let requestUrl: string | undefined; let selectedProtocol: string | undefined; const wss = track( @@ -189,15 +183,13 @@ test.each(buttonCases)( selectedProtocol = socket.protocol; socket.on("message", (data, binary) => { received.push({ data: toBuffer(data), binary }); - lastMessageAt = performance.now(); - }); - socket.on("close", () => { - closedAt = performance.now(); - closed.resolve(); }); + socket.on("close", () => closed.resolve()); }); + const startedAt = performance.now(); await pressButton(descriptor(baseUrl), button); + const elapsed = performance.now() - startedAt; await closed.promise; expect(requestUrl).toBe("/vendor/serve-sim/helper/ws?device=UDID-1"); @@ -205,7 +197,7 @@ test.each(buttonCases)( expect(received).toEqual([ { data: Buffer.concat([Buffer.from([4]), Buffer.from(payload, "utf8")]), binary: true }, ]); - expect(closedAt - lastMessageAt).toBeGreaterThanOrEqual(490); + expect(elapsed).toBeGreaterThanOrEqual(490); }, ); @@ -252,6 +244,62 @@ test("sendTouches reports a refused upgrade without claiming input ran or leakin }); }); +test("cancelling an opening HID socket closes it promptly without claiming input was sent", async () => { + const connected = Promise.withResolvers(); + const sockets = new Set(); + const server = createTcpServer((socket) => { + sockets.add(socket); + socket.on("data", () => connected.resolve()); + socket.on("end", () => socket.end()); + socket.on("close", () => sockets.delete(socket)); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + teardown.push( + () => + new Promise((resolve) => { + for (const socket of sockets) socket.destroy(); + server.close(() => resolve()); + }), + ); + const baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + const abort = new AbortController(); + const operation = pressButton(descriptor(baseUrl), "home", abort.signal); + const outcome = operation.catch((error: unknown) => error); + await connected.promise; + abort.abort(); + expect(await outcome).toMatchObject({ data: { code: "detached", mayHaveExecuted: false } }); +}); + +test("cancelling a running gesture stops further HID input and preserves dispatch uncertainty", async () => { + const server = createServer(); + const baseUrl = await listen(server); + const received: Buffer[] = []; + const first = Promise.withResolvers(); + const closed = Promise.withResolvers(); + const wss = track(new WebSocketServer({ server })); + wss.on("connection", (socket) => { + socket.on("message", (data) => { + received.push(toBuffer(data)); + first.resolve(); + }); + socket.on("close", () => closed.resolve()); + }); + const abort = new AbortController(); + const operation = sendTouches( + descriptor(baseUrl), + swipeEvents({ x: 0, y: 0 }, { x: 1, y: 1 }), + 1_000, + 1_000, + abort.signal, + ); + const outcome = operation.catch((error: unknown) => error); + await first.promise; + abort.abort(); + expect(await outcome).toMatchObject({ data: { code: "detached", mayHaveExecuted: true } }); + await closed.promise; + expect(received).toHaveLength(1); +}); + test("waitForScreen polls the authenticated config route until capture becomes ready", async () => { const requests: Array<{ url?: string; auth?: string }> = []; const server = createServer((request, response) => { diff --git a/src/native/serve-sim.ts b/src/native/serve-sim.ts index bca27fd..d8deac2 100644 --- a/src/native/serve-sim.ts +++ b/src/native/serve-sim.ts @@ -1,7 +1,7 @@ import { setTimeout as delay } from "node:timers/promises"; import { WebSocket } from "ws"; import type { Device } from "../rpc.js"; -import { deviceError } from "./errors.js"; +import { cancellationError, deviceError } from "./errors.js"; // The hub's vendored serve-sim routes for an iOS Simulator. Manual input goes over its HID WebSocket: each message is one // tag byte followed by UTF-8 JSON, and touch coordinates are normalized to the 0..1 screen surface, not pixels. @@ -78,20 +78,35 @@ export function swipeEvents(from: ScreenPoint, to: ScreenPoint, steps = SWIPE_ST return [{ type: "begin", ...from }, ...moves, { type: "end", ...to }]; } -export function tap(stream: Device.StreamDescriptor, point: ScreenPoint): Promise { - return sendTouches(stream, tapEvents(point), TAP_HOLD_MS); +export function tap( + stream: Device.StreamDescriptor, + point: ScreenPoint, + signal?: AbortSignal, +): Promise { + return sendTouches(stream, tapEvents(point), TAP_HOLD_MS, undefined, signal); } export function swipe( stream: Device.StreamDescriptor, from: ScreenPoint, to: ScreenPoint, + signal?: AbortSignal, ): Promise { - return sendTouches(stream, swipeEvents(from, to), SWIPE_DURATION_MS / SWIPE_STEPS); + return sendTouches( + stream, + swipeEvents(from, to), + SWIPE_DURATION_MS / SWIPE_STEPS, + undefined, + signal, + ); } -export function pressButton(stream: Device.StreamDescriptor, button: HidButton): Promise { - return sendHid(stream, [encodeButton(button)], 0, "button press"); +export function pressButton( + stream: Device.StreamDescriptor, + button: HidButton, + signal?: AbortSignal, +): Promise { + return sendHid(stream, [encodeButton(button)], 0, "button press", undefined, signal); } // Send touch events over one short-lived HID socket, spacing them so the simulator sees a real gesture. @@ -100,8 +115,9 @@ export function sendTouches( events: readonly TouchEvent[], intervalMs: number, openTimeoutMs?: number, + signal?: AbortSignal, ): Promise { - return sendHid(stream, events.map(encodeTouch), intervalMs, "touch input", openTimeoutMs); + return sendHid(stream, events.map(encodeTouch), intervalMs, "touch input", openTimeoutMs, signal); } async function sendHid( @@ -110,35 +126,62 @@ async function sendHid( intervalMs: number, action: string, openTimeoutMs = 10_000, + signal?: AbortSignal, ): Promise { + signal?.throwIfAborted(); const url = helperUrl(stream, "ws"); url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; const socket = new WebSocket(url, [`serve-sim.token.${stream.token}`]); // Errors surface through `opened` and `send`; this keeps a late socket error from becoming an uncaught event. socket.on("error", () => undefined); let sent = 0; + const abort = () => socket.terminate(); + signal?.addEventListener("abort", abort, { once: true }); try { - await opened(socket, openTimeoutMs); - for (const message of messages) { - if (sent > 0) await delay(intervalMs); - await send(socket, message); - sent++; - } - await delay(HID_SETTLE_MS); + await opened(socket, openTimeoutMs, signal); + await sendMessages(socket, messages, intervalMs, () => sent++, signal); + await delay(HID_SETTLE_MS, undefined, { signal }); } catch (error) { - const detail = - error instanceof Error ? error.message.replaceAll(stream.token, "[redacted]") : ""; - throw deviceError( - "helper-failed", - `The simulator did not accept the ${action}${detail ? `: ${detail}` : "."}`, - sent > 0, - ); + throw hidFailure(error, stream.token, action, sent, signal); } finally { + signal?.removeEventListener("abort", abort); if (socket.readyState === WebSocket.OPEN) socket.close(1000); else socket.terminate(); } } +async function sendMessages( + socket: WebSocket, + messages: readonly Buffer[], + intervalMs: number, + onSend: () => void, + signal?: AbortSignal, +): Promise { + for (const [index, message] of messages.entries()) { + signal?.throwIfAborted(); + if (index > 0) await delay(intervalMs, undefined, { signal }); + if (socket.readyState !== WebSocket.OPEN) throw new Error("the HID socket closed"); + onSend(); + await send(socket, message); + } +} + +function hidFailure( + error: unknown, + token: string, + action: string, + sent: number, + signal?: AbortSignal, +) { + if (signal?.aborted) return cancellationError(signal, sent > 0); + const detail = error instanceof Error ? error.message.replaceAll(token, "[redacted]") : ""; + return deviceError( + "helper-failed", + `The simulator did not accept the ${action}${detail ? `: ${detail}` : "."}`, + sent > 0, + ); +} + // The hub reports ready before native capture starts; poll the screen config until the simulator display is served. export async function waitForScreen( stream: Device.StreamDescriptor, @@ -149,17 +192,9 @@ export async function waitForScreen( const url = helperUrl(stream, "config"); while (Date.now() < deadline) { signal?.throwIfAborted(); - const response = await fetch(url, { - headers: { Authorization: `Bearer ${stream.token}` }, - redirect: "error", - signal: AbortSignal.any([ - AbortSignal.timeout(Math.max(1, Math.min(5_000, deadline - Date.now()))), - ...(signal ? [signal] : []), - ]), - }).catch(() => null); - await response?.body?.cancel().catch(() => undefined); + const available = await screenAvailable(url, stream.token, deadline - Date.now(), signal); signal?.throwIfAborted(); - if (response?.ok) return; + if (available) return; await delay(Math.min(intervalMs, Math.max(0, deadline - Date.now())), undefined, { signal }); } throw deviceError( @@ -168,7 +203,25 @@ export async function waitForScreen( ); } -function opened(socket: WebSocket, timeoutMs: number): Promise { +async function screenAvailable( + url: URL, + token: string, + remainingMs: number, + signal?: AbortSignal, +): Promise { + const response = await fetch(url, { + headers: { Authorization: `Bearer ${token}` }, + redirect: "error", + signal: AbortSignal.any([ + AbortSignal.timeout(Math.max(1, Math.min(5_000, remainingMs))), + ...(signal ? [signal] : []), + ]), + }).catch(() => null); + await response?.body?.cancel().catch(() => undefined); + return response?.ok === true; +} + +function opened(socket: WebSocket, timeoutMs: number, signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { const timer = setTimeout( () => fail(new Error("the HID socket did not open in time")), @@ -178,7 +231,9 @@ function opened(socket: WebSocket, timeoutMs: number): Promise { clearTimeout(timer); socket.off("open", succeed); socket.off("error", fail); + socket.off("close", closed); socket.off("unexpected-response", rejected); + signal?.removeEventListener("abort", cancelled); }; const succeed = () => { cleanup(); @@ -190,15 +245,22 @@ function opened(socket: WebSocket, timeoutMs: number): Promise { }; const rejected = (_request: unknown, response: { statusCode?: number }) => fail(new Error(`the HID socket was refused (HTTP ${response.statusCode ?? "unknown"})`)); + const closed = () => fail(new Error("the HID socket closed before it opened")); + const cancelled = () => { + cleanup(); + reject(signal?.reason); + }; socket.once("open", succeed); socket.once("error", fail); + socket.once("close", closed); socket.once("unexpected-response", rejected); + signal?.addEventListener("abort", cancelled, { once: true }); + if (signal?.aborted) cancelled(); }); } function send(socket: WebSocket, data: Buffer): Promise { return new Promise((resolve, reject) => { - if (socket.readyState !== WebSocket.OPEN) return reject(new Error("the HID socket closed")); socket.send(data, { binary: true }, (error) => (error ? reject(error) : resolve())); }); } diff --git a/src/rpc.ts b/src/rpc.ts index 3d1a51a..a6aa521 100644 --- a/src/rpc.ts +++ b/src/rpc.ts @@ -176,8 +176,8 @@ export const Screenshot = Schema.Struct({ platform: Platform, }).annotate({ identifier: "Device.Screenshot" }); -// The stream wire protocol each backend speaks. iOS and Android differ; the viewer decodes the advertised one. -export const StreamProtocol = Schema.Literals(["mjpeg", "h264"]).annotate({ +// Both platforms expose Annex-B H.264 through the video-only WebSocket proxy. +export const StreamProtocol = Schema.Literal("h264").annotate({ identifier: "Device.StreamProtocol", }); export type StreamProtocol = typeof StreamProtocol.Type; diff --git a/src/stream-avcc.test.ts b/src/stream-avcc.test.ts new file mode 100644 index 0000000..173c079 --- /dev/null +++ b/src/stream-avcc.test.ts @@ -0,0 +1,70 @@ +import { expect, test } from "bun:test"; +import { EventEmitter } from "node:events"; +import { createServer, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import { WebSocket } from "ws"; +import { bridgeAvcc } from "./stream-avcc.js"; +import { avccDescription, avccSample, idr, delta } from "../test/video.js"; + +test("iOS backpressure restarts capture once after draining and releases every subscription", async () => { + const frames: Buffer[] = []; + const responses: ServerResponse[] = []; + const closed: Promise[] = []; + const first = Promise.withResolvers(); + const recovered = Promise.withResolvers(); + const blocked = Promise.withResolvers(); + let buffered = 0; + class Viewer extends EventEmitter { + readyState: number = WebSocket.OPEN; + get bufferedAmount() { + if (buffered > 0) blocked.resolve(); + return buffered; + } + send(bytes: Buffer) { + frames.push(bytes); + if (frames.length === 1) first.resolve(); + else recovered.resolve(); + } + close() { + this.readyState = WebSocket.CLOSED; + this.emit("close"); + } + } + const viewer = new Viewer(); + const server = createServer((request, response) => { + expect(request.headers.authorization).toBe("Bearer hub-secret"); + responses.push(response); + closed.push(new Promise((resolve) => request.socket.once("close", () => resolve()))); + response.writeHead(200).flushHeaders(); + response.write(Buffer.concat([avccDescription(), avccSample(2, [idr.subarray(4)])])); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const bridge = bridgeAvcc( + { + url: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, + authorization: "Bearer hub-secret", + }, + viewer as unknown as WebSocket, + 100, + ); + try { + await first.promise; + buffered = 95; + responses[0]!.write(avccSample(3, [delta.subarray(3)])); + await blocked.promise; + buffered = 0; + responses[0]!.write(avccSample(3, [delta.subarray(3)])); + await recovered.promise; + expect(responses).toHaveLength(2); + expect(frames).toHaveLength(2); + expect(frames[1]).toEqual(frames[0]); + } finally { + viewer.close(); + await bridge.done; + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + await Promise.all(closed); + } + expect(viewer.listenerCount("close")).toBe(0); + expect(viewer.listenerCount("error")).toBe(0); +}); diff --git a/src/stream-avcc.ts b/src/stream-avcc.ts new file mode 100644 index 0000000..014031e --- /dev/null +++ b/src/stream-avcc.ts @@ -0,0 +1,116 @@ +import { WebSocket } from "ws"; +import { get as httpGet, type IncomingMessage } from "node:http"; +import { get as httpsGet } from "node:https"; +import { AvccStream, type AvccFrame } from "./avcc.js"; +import { VideoFlow } from "./video-flow.js"; + +export interface AvccSource { + readonly url: string; + readonly authorization: string; +} + +export function bridgeAvcc(source: AvccSource, downstream: WebSocket, maxBufferedBytes: number) { + let current: AbortController | undefined; + let receivedFrame = false; + let restarting = false; + const abort = () => current?.abort(); + const reset = () => { + // Opening the AVCC subscription already requests a keyframe. Ignore resets until it has sent video. + if (!receivedFrame || restarting) return; + restarting = true; + abort(); + }; + downstream.on("close", abort); + downstream.on("error", abort); + + const run = async () => { + try { + while (downstream.readyState === WebSocket.OPEN) { + const controller = new AbortController(); + current = controller; + restarting = false; + receivedFrame = false; + try { + await pumpAvcc( + source, + controller.signal, + forwardFrames(downstream, maxBufferedBytes, reset, () => { + receivedFrame = true; + }), + ); + } catch (error) { + if (!controller.signal.aborted) throw error; + } finally { + controller.abort(); + } + if (!restarting) break; + } + } catch (error) { + if (downstream.readyState !== WebSocket.OPEN) return; + const token = source.authorization.slice("Bearer ".length); + const message = (error instanceof Error ? error.message : String(error)) + .replaceAll(token, "[redacted]") + .slice(0, 512); + downstream.send(JSON.stringify({ ok: false, error: message })); + downstream.close(1011); + } finally { + current = undefined; + downstream.off("close", abort); + downstream.off("error", abort); + } + }; + + return { + done: run(), + reset, + }; +} + +function forwardFrames( + downstream: WebSocket, + maxBufferedBytes: number, + recover: () => void, + onFrame: () => void, +) { + const flow = new VideoFlow(maxBufferedBytes, recover); + return (frame: AvccFrame) => { + if (downstream.readyState !== WebSocket.OPEN) return; + if (!flow.accept({ ...frame, picture: true }, downstream.bufferedAmount)) return; + downstream.send(frame.bytes, { binary: true }); + onFrame(); + }; +} + +async function pumpAvcc(source: AvccSource, signal: AbortSignal, send: (frame: AvccFrame) => void) { + const response = await openStream(source, signal); + const stream = new AvccStream(); + try { + if (response.statusCode !== 200) + throw new Error(`iOS H.264 stream returned HTTP ${response.statusCode}`); + for await (const chunk of response) { + if (signal.aborted) return; + for (const frame of stream.push(chunk)) { + if (signal.aborted) return; + send(frame); + } + } + if (signal.aborted) return; + stream.finish(); + throw new Error("iOS H.264 stream ended. Refresh devices to retry."); + } finally { + response.destroy(); + } +} + +// Node HTTP closes the capture connection on abort in both Node and Bun, and never follows redirects with credentials. +function openStream(source: AvccSource, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const url = new URL(source.url); + const get = url.protocol === "https:" ? httpsGet : httpGet; + const request = get(url, { headers: { authorization: source.authorization }, signal }, resolve); + request.once("error", reject); + request.once("close", () => + reject(new Error("iOS H.264 connection closed before stream headers")), + ); + }); +} diff --git a/src/stream-proxy.test.ts b/src/stream-proxy.test.ts index 15258fa..81111a3 100644 --- a/src/stream-proxy.test.ts +++ b/src/stream-proxy.test.ts @@ -1,9 +1,12 @@ import { afterEach, describe, expect, test } from "bun:test"; import { createServer, type Server } from "node:http"; +import { on } from "node:events"; import type { AddressInfo } from "node:net"; import { WebSocket, WebSocketServer, type RawData } from "ws"; import type { Device } from "./rpc.js"; import { startStreamProxy, type StreamProxy } from "./stream-proxy.js"; +import { parseFramePacket } from "./h264.js"; +import { avccDescription, avccRecord, avccSample, idr, delta } from "../test/video.js"; const HUB_TOKEN = "hub-secret-token"; @@ -23,10 +26,59 @@ afterEach(async () => { for (const close of teardown.splice(0).reverse()) await close(); }); +test("malformed HTTP and upgrade targets return 400 without crashing the Node host", async () => { + const built = await Bun.build({ + entrypoints: ["src/stream-proxy.ts"], + target: "node", + format: "cjs", + packages: "external", + }); + if (!built.success) throw new Error(built.logs.join("\n")); + const bundle = await built.outputs[0]!.text(); + const child = Bun.spawn(["node"], { stdin: "pipe", stdout: "pipe", stderr: "pipe" }); + child.stdin.write(` + const assert = require("node:assert/strict"); + const { connect } = require("node:net"); + const bundle = { exports: {} }; + new Function("require", "module", "exports", ${JSON.stringify(bundle)})(require, bundle, bundle.exports); + (async () => { + const proxy = await bundle.exports.startStreamProxy({ platform: "android", deviceId: "fixture", baseUrl: "http://127.0.0.1:1", token: "fixture", protocol: "h264", loopback: true }); + const origin = new URL(proxy.descriptor.baseUrl); + try { + for (const target of ["http://[", "//[", "http://%"] ) { + for (const upgrade of [false, true]) { + const response = await new Promise((resolve, reject) => { + const socket = connect(Number(origin.port), origin.hostname); + let result = ""; + socket.setTimeout(2000, () => socket.destroy(new Error("request timed out"))); + socket.on("error", reject); + socket.on("data", chunk => result += chunk.toString()); + socket.on("close", () => resolve(result)); + socket.on("connect", () => socket.write("GET " + target + " HTTP/1.1\\r\\nHost: localhost\\r\\nConnection: " + (upgrade ? "Upgrade\\r\\nUpgrade: websocket\\r\\nSec-WebSocket-Version: 13\\r\\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" : "close") + "\\r\\n\\r\\n")); + }); + assert.equal(response.startsWith("HTTP/1.1 400"), true); + } + } + assert.equal((await fetch(proxy.descriptor.baseUrl + "/video")).status, 426); + } finally { + await proxy.stop(); + } + console.log("malformed targets rejected; proxy still responsive"); + })().catch(error => { console.error(error); process.exitCode = 1; }); + `); + child.stdin.end(); + const [code, stdout, stderr] = await Promise.all([ + child.exited, + new Response(child.stdout).text(), + new Response(child.stderr).text(), + ]); + expect(stderr).toBe(""); + expect(code).toBe(0); + expect(stdout).toContain("proxy still responsive"); +}, 10_000); + function track } | { close(): void }>(resource: T): T { - teardown.push(() => - "stop" in resource ? resource.stop() : new Promise((r) => (resource.close(), r())), - ); + teardown.push(() => ("stop" in resource ? resource.stop() : resource.close())); return resource; } @@ -40,103 +92,131 @@ function proxyVideoUrl(proxy: StreamProxy, query = ""): string { return `${proxy.descriptor.baseUrl}/video${query}`; } -describe("MJPEG proxy (iOS)", () => { +function viewer( + proxy: StreamProxy, + query = "", + protocols = [`opencode-device.${proxy.descriptor.token}`], +) { + return track(new WebSocket(proxyVideoUrl(proxy, query), protocols)); +} + +describe("H.264 proxy (iOS)", () => { const simDescriptor = (baseUrl: string) => - descriptor({ platform: "ios", protocol: "mjpeg", deviceId: "UDID-1", baseUrl }); + descriptor({ platform: "ios", deviceId: "UDID-1", baseUrl }); - // A stand-in for the hub's serve-sim route: it enforces the fixed path, query, and Bearer token, then streams MJPEG. - function fakeSimHub(): Promise<{ - baseUrl: string; - lastRequest: () => { url?: string; auth?: string }; - }> { - let url: string | undefined; - let auth: string | undefined; + // A seed JPEG, AVC decoder description, keyframe, and delta in serve-sim's length-prefixed HTTP format. + const avcc = Buffer.concat([ + avccRecord(4, Buffer.from([0xff, 0xd8, 0xff, 0xd9])), + avccDescription(), + avccSample(2, [idr.subarray(4)]), + avccSample(3, [delta.subarray(3)]), + ]); + function fakeSimHub( + options: { status?: number; payload?: Buffer; end?: boolean; headers?: boolean } = {}, + ) { + const requests: Array<{ url?: string; auth?: string; method?: string }> = []; + const closed = Promise.withResolvers(); + const requested = Promise.withResolvers(); const server = track( createServer((request, response) => { - url = request.url; - auth = request.headers.authorization; + requests.push({ + url: request.url, + auth: request.headers.authorization, + method: request.method, + }); + requested.resolve(); + request.socket.on("close", () => closed.resolve()); if ( - request.url !== "/vendor/serve-sim/helper/stream.mjpeg?device=UDID-1" || - auth !== `Bearer ${HUB_TOKEN}` + request.url !== "/vendor/serve-sim/helper/stream.avcc?device=UDID-1" || + request.headers.authorization !== `Bearer ${HUB_TOKEN}` ) { response.writeHead(401).end(); return; } - response.writeHead(200, { "content-type": "multipart/x-mixed-replace; boundary=frame" }); - response.write("--frame\r\nContent-Type: image/jpeg\r\n\r\n"); - response.write(Buffer.from([0xff, 0xd8, 0xff, 0xe0, 1, 2, 3])); - response.write("\r\n--frame\r\nContent-Type: image/jpeg\r\n\r\n"); - response.write(Buffer.from([0xff, 0xd8, 0xff, 0xe0, 4, 5, 6])); + if (options.headers === false) return; + response.writeHead(options.status ?? 200, { "content-type": "application/octet-stream" }); + response.flushHeaders(); + response.write(options.payload ?? avcc); + if (options.end || options.status) response.end(); }), ); return listen(server).then((port) => ({ baseUrl: `http://127.0.0.1:${port}`, - lastRequest: () => ({ url, auth }), + requests, + closed: closed.promise, + requested: requested.promise, })); } - test("streams the hub body only to a correctly authenticated viewer and hides the hub token", async () => { + test("normalizes native H.264 for the shared authenticated viewer and skips JPEG seeds", async () => { const hub = await fakeSimHub(); const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); expect(proxy.descriptor.token).not.toBe(HUB_TOKEN); expect(proxy.descriptor.baseUrl.startsWith("http://127.0.0.1:")).toBe(true); expect(proxy.descriptor.loopback).toBe(true); - - const bytes = await readMjpeg(proxyVideoUrl(proxy), `Bearer ${proxy.descriptor.token}`); - expect(bytes.includes(Buffer.from([0xff, 0xd8, 0xff, 0xe0, 1, 2, 3]))).toBe(true); - expect(bytes.includes(Buffer.from([0xff, 0xd8, 0xff, 0xe0, 4, 5, 6]))).toBe(true); - expect(hub.lastRequest()).toEqual({ - url: "/vendor/serve-sim/helper/stream.mjpeg?device=UDID-1", - auth: `Bearer ${HUB_TOKEN}`, + expect(proxy.descriptor.protocol).toBe("h264"); + const client = viewer(proxy, "?frame-meta=1"); + const messages = await collect(client, 2); + expect(messages).toEqual([ + Buffer.from([ + 0, 0, 0, 1, 0x67, 0x64, 0, 0x29, 0, 0, 0, 1, 0x68, 0xeb, 0, 0, 0, 1, 0x65, 0x88, + ]), + Buffer.from([0, 0, 0, 1, 0x41, 0x99]), + ]); + expect(parseFramePacket(messages[0] as Buffer)).toMatchObject({ + key: true, + codec: "avc1.640029", }); + expect(hub.requests).toEqual([ + { + url: "/vendor/serve-sim/helper/stream.avcc?device=UDID-1", + auth: `Bearer ${HUB_TOKEN}`, + method: "GET", + }, + ]); + client.close(); + await hub.closed; }); - test("allows Authorization preflight only on the fixed video route", async () => { + test("only a validated video reset restarts the capture subscription", async () => { const hub = await fakeSimHub(); const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); - const response = await fetch(proxyVideoUrl(proxy), { - method: "OPTIONS", - headers: { - origin: "http://localhost:3000", - "access-control-request-method": "GET", - "access-control-request-headers": "authorization", - }, - }); - expect(response.status).toBe(204); - expect(response.headers.get("access-control-allow-origin")).toBe("*"); - expect(response.headers.get("access-control-allow-headers")?.toLowerCase()).toContain( - "authorization", - ); - expect(response.headers.get("access-control-allow-methods")).toBe("GET"); - expect(await response.text()).toBe(""); - expect(hub.lastRequest()).toEqual({ url: undefined, auth: undefined }); - expect((await fetch(`${proxy.descriptor.baseUrl}/secret`, { method: "OPTIONS" })).status).toBe( - 404, - ); + const client = viewer(proxy); + await collect(client, 2); + const next = collect(client, 2); + client.send(Buffer.from([9, 9, 9])); + client.send("steer the device"); + client.send(JSON.stringify({ type: "tap", x: 10, y: 20 })); + client.send(JSON.stringify({ type: "reset-video", ack: true, device: "must-not-forward" })); + await next; + expect(hub.requests).toHaveLength(2); + expect(hub.requests[1]).toEqual(hub.requests[0]); + client.close(); + await hub.closed; }); - test("rejects a missing, wrong, or malformed viewer request", async () => { + test("rejects unauthenticated sockets without opening an upstream stream", async () => { const hub = await fakeSimHub(); const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); - const token = proxy.descriptor.token; + expect(await opens(new WebSocket(proxyVideoUrl(proxy)))).toBe(false); + expect(await opens(new WebSocket(proxyVideoUrl(proxy), ["opencode-device.wrong"]))).toBe(false); + expect( + await opens( + new WebSocket(proxyVideoUrl(proxy, "?device=other"), [ + `opencode-device.${proxy.descriptor.token}`, + ]), + ), + ).toBe(false); + expect(hub.requests).toEqual([]); + }); - for (const authorization of [ - undefined, - "Bearer nope", - `Bearer ${token}x`, - "Bearer ", - `Basic ${token}`, - ]) { - expect( - ( - await fetch(proxyVideoUrl(proxy), { - headers: authorization ? { authorization } : undefined, - }) - ).status, - ).toBe(401); - } + test("HTTP video and input routes are unavailable", async () => { + const hub = await fakeSimHub(); + const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); for (const [method, route, httpStatus] of [ + ["GET", "/video", 426], + ["OPTIONS", "/video", 405], ["POST", "/video", 405], ["GET", "/", 404], ["GET", "/secret", 404], @@ -149,20 +229,66 @@ describe("MJPEG proxy (iOS)", () => { ( await fetch(`${proxy.descriptor.baseUrl}${route}`, { method, - headers: { authorization: `Bearer ${token}` }, + headers: { authorization: `Bearer ${proxy.descriptor.token}` }, }) ).status, ).toBe(httpStatus); } + expect(hub.requests).toEqual([]); }); - test("stop closes the listener so the proxy no longer accepts connections", async () => { + test.each([ + { options: { status: 401 }, error: "HTTP 401" }, + { options: { payload: Buffer.from([0, 128, 0, 1]) }, error: "record size" }, + { options: { payload: Buffer.from([0, 0]), end: true }, error: "Truncated" }, + { options: { end: true }, error: "stream ended" }, + ])("reports upstream failures and closes the viewer: $error", async ({ options, error }) => { + const hub = await fakeSimHub(options); + const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); + const client = viewer(proxy); + const messages: Array = []; + client.on("message", (data, binary) => + messages.push(binary ? toBuffer(data) : toBuffer(data).toString()), + ); + await new Promise((resolve, reject) => { + client.on("error", reject); + client.on("close", () => resolve()); + }); + const message = JSON.parse(messages.at(-1) as string); + expect(message.ok).toBe(false); + expect(message.error).toContain(error); + expect(message.error).not.toContain(HUB_TOKEN); + }); + + test.each([false, true])( + "stop aborts a pending upstream stream and closes the listener, headers=%s", + async (headers) => { + const hub = await fakeSimHub({ payload: Buffer.alloc(0), headers }); + const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); + const url = proxyVideoUrl(proxy); + const client = viewer(proxy); + const closed = new Promise((resolve) => client.on("close", () => resolve())); + await hub.requested; + await proxy.stop(); + await closed; + await hub.closed; + await expect(fetch(url)).rejects.toThrow(); + }, + ); + + test("viewer closure cancels its stream without stopping another viewer", async () => { const hub = await fakeSimHub(); - const proxy = await startStreamProxy(simDescriptor(hub.baseUrl)); - const url = proxyVideoUrl(proxy); - const token = proxy.descriptor.token; - await proxy.stop(); - await expect(fetch(url, { headers: { authorization: `Bearer ${token}` } })).rejects.toThrow(); + const proxy = track(await startStreamProxy(simDescriptor(hub.baseUrl))); + const first = viewer(proxy); + const second = viewer(proxy); + await Promise.all([collect(first, 2), collect(second, 2)]); + first.close(); + await hub.closed; + const next = collect(second, 2); + second.send(JSON.stringify({ type: "reset-video" })); + await next; + expect(second.readyState).toBe(WebSocket.OPEN); + second.close(); }); }); @@ -177,12 +303,11 @@ describe("H.264 proxy (Android)", () => { // A stand-in for the hub's serve-emu socket: it enforces the serve-emu.token subprotocol, records inbound frames // (to prove the proxy never relays a client message upstream), and sends one H.264 binary frame plus a text header. - function fakeEmuHub(): Promise { + function fakeEmuHub(frames = [Buffer.from([0, 0, 0, 1, 0x67, 0x42])]): Promise { let lastUrl: string | undefined; const received: RawData[] = []; const clientMessage = Promise.withResolvers(); - let resolveClosed: () => void = () => {}; - const clientClosed = new Promise((resolve) => (resolveClosed = resolve)); + const clientClosed = Promise.withResolvers(); const server = createServer(); const wss = new WebSocketServer({ server, @@ -195,9 +320,9 @@ describe("H.264 proxy (Android)", () => { received.push(data); clientMessage.resolve(data); }); - socket.on("close", resolveClosed); + socket.on("close", () => clientClosed.resolve()); socket.send(JSON.stringify({ type: "video-session", size: { width: 1080, height: 2400 } })); - socket.send(Buffer.from([0x00, 0x00, 0x00, 0x01, 0x67, 0x42]), { binary: true }); + for (const frame of frames) socket.send(frame, { binary: true }); socket.send( JSON.stringify({ type: "diagnostic", message: `fixture diagnostic ${HUB_TOKEN}` }), ); @@ -215,7 +340,7 @@ describe("H.264 proxy (Android)", () => { lastUrl: () => lastUrl, received: () => received, clientMessage: clientMessage.promise, - clientClosed, + clientClosed: clientClosed.promise, })); } @@ -223,7 +348,7 @@ describe("H.264 proxy (Android)", () => { const hub = await fakeEmuHub(); const proxy = track(await startStreamProxy(descriptor({ baseUrl: hub.baseUrl }))); - const client = new WebSocket(proxyVideoUrl(proxy, "?frame-meta=1"), [ + const client = viewer(proxy, "?frame-meta=1", [ "other", `opencode-device.${proxy.descriptor.token}`, ]); @@ -247,9 +372,7 @@ describe("H.264 proxy (Android)", () => { test("never relays device input or arbitrary client messages upstream", async () => { const hub = await fakeEmuHub(); const proxy = track(await startStreamProxy(descriptor({ baseUrl: hub.baseUrl }))); - const client = new WebSocket(proxyVideoUrl(proxy), [ - `opencode-device.${proxy.descriptor.token}`, - ]); + const client = viewer(proxy); await collect(client, 1); client.send(Buffer.from([9, 9, 9])); client.send("steer the device"); @@ -259,12 +382,24 @@ describe("H.264 proxy (Android)", () => { expect(hub.received()).toEqual([]); }); + test("forwards separate configuration but drops initial dependent pictures until an IDR", async () => { + const sps = Buffer.from([0, 0, 0, 1, 0x67, 0x64, 0, 0x29]); + const skipped = Buffer.from([0, 0, 1, 0x41, 0x11]); + const idr = Buffer.from([0, 0, 0, 1, 0x65, 0x88]); + const delta = Buffer.from([0, 0, 1, 0x41, 0x22]); + const hub = await fakeEmuHub([sps, skipped, idr, delta]); + const proxy = track(await startStreamProxy(descriptor({ baseUrl: hub.baseUrl }))); + const client = viewer(proxy); + const messages = await collect(client, 5); + expect(messages.filter((message) => Buffer.isBuffer(message))).toEqual([sps, idr, delta]); + client.close(); + await hub.clientClosed; + }); + test("allows only a fixed video reset and never forwards arbitrary request fields", async () => { const hub = await fakeEmuHub(); const proxy = track(await startStreamProxy(descriptor({ baseUrl: hub.baseUrl }))); - const client = new WebSocket(proxyVideoUrl(proxy), [ - `opencode-device.${proxy.descriptor.token}`, - ]); + const client = viewer(proxy); await collect(client, 1); client.send(JSON.stringify({ type: "tap", x: 10, y: 20 })); client.send(JSON.stringify({ type: "reset-video", ack: true, secret: "must-not-forward" })); @@ -297,38 +432,14 @@ describe("H.264 proxy (Android)", () => { }); }); -async function readMjpeg(url: string, authorization: string): Promise { - const controller = new AbortController(); - const response = await fetch(url, { headers: { authorization }, signal: controller.signal }); - expect(response.status).toBe(200); - expect(response.headers.get("access-control-allow-origin")).toBe("*"); - const reader = response.body!.getReader(); - const chunks: Uint8Array[] = []; - let total = 0; - while (total < 14) { - const next = await reader.read(); - if (next.done) break; - if (next.value) { - chunks.push(next.value); - total += next.value.byteLength; - } +async function collect(socket: WebSocket, count: number): Promise> { + const messages: Array = []; + for await (const [data, isBinary] of on(socket, "message", { close: ["close"] })) { + const bytes = toBuffer(data); + messages.push(isBinary ? bytes : bytes.toString("utf8")); + if (messages.length === count) return messages; } - controller.abort(); - await reader.cancel().catch(() => {}); - return Buffer.concat(chunks); -} - -function collect(socket: WebSocket, count: number): Promise> { - return new Promise((resolve, reject) => { - const messages: Array = []; - socket.on("message", (data: RawData, isBinary: boolean) => { - const buffer = toBuffer(data); - messages.push(isBinary ? buffer : buffer.toString("utf8")); - if (messages.length >= count) resolve(messages); - }); - socket.on("error", reject); - socket.on("close", () => reject(new Error("socket closed before enough messages"))); - }); + throw new Error("socket closed before enough messages"); } function opens(socket: WebSocket): Promise { diff --git a/src/stream-proxy.ts b/src/stream-proxy.ts index 75a48bc..7bee31a 100644 --- a/src/stream-proxy.ts +++ b/src/stream-proxy.ts @@ -1,12 +1,14 @@ import { randomBytes, timingSafeEqual } from "node:crypto"; -import { once } from "node:events"; -import { createServer, type Server, type ServerResponse } from "node:http"; +import { createServer, type Server } from "node:http"; import type { Duplex } from "node:stream"; import { WebSocket, WebSocketServer, type RawData } from "ws"; import { Schema } from "effect"; import type { Device } from "./rpc.js"; +import { bridgeAvcc, type AvccSource } from "./stream-avcc.js"; +import { parseFramePacket } from "./h264.js"; +import { VideoFlow } from "./video-flow.js"; -// Give the viewer a video-only loopback token, never the hub credential. Relay H.264 or MJPEG and fixed reset messages. +// Give the viewer a video-only loopback token, never the hub credential. Both platforms expose Annex-B H.264. const LOOPBACK_HOST = "127.0.0.1"; const VIDEO_PATH = "/video"; @@ -23,40 +25,38 @@ export interface StreamProxy { type UpstreamTarget = | { readonly kind: "ws"; readonly url: string; readonly subprotocol: string } - | { readonly kind: "mjpeg"; readonly url: string; readonly authorization: string }; + | ({ readonly kind: "avcc" } & AvccSource); export function startStreamProxy(raw: Device.StreamDescriptor): Promise { const token = randomBytes(24).toString("hex"); const upstream = buildUpstream(raw); - const responses = new Set(); - const controllers = new Set(); const sockets = new Set(); - const relay = - upstream.kind === "ws" ? new WebSocketServer({ noServer: true, maxPayload: 4_096 }) : undefined; - if (relay && upstream.kind === "ws") { - const hub = upstream; - relay.on("connection", (downstream: WebSocket) => bridgeWebSocket(hub, downstream, sockets)); - } + const tasks = new Set>(); + let stopping: Promise | undefined; + const relay = new WebSocketServer({ noServer: true, maxPayload: 4_096 }); + relay.on("connection", (downstream: WebSocket) => { + if (stopping) return downstream.terminate(); + if (upstream.kind === "ws") return bridgeWebSocket(upstream, downstream, sockets); + sockets.add(downstream); + downstream.on("close", () => sockets.delete(downstream)); + downstream.on("error", () => downstream.terminate()); + const bridge = bridgeAvcc(upstream, downstream, MAX_BUFFERED_BYTES); + downstream.on("message", (data: RawData, isBinary: boolean) => { + if (isVideoReset(data, isBinary)) bridge.reset(); + }); + const done = bridge.done.catch(() => downstream.terminate()); + tasks.add(done); + void done.then(() => tasks.delete(done)); + }); const server = createServer((request, response) => { - if (upstream.kind === "mjpeg" && request.method === "OPTIONS") - return preflight(response, videoRequestStatus("GET", request.url)); const status = videoRequestStatus(request.method, request.url); - if (status) return reject(response, status); - if (upstream.kind !== "mjpeg") return reject(response, 426); - // The viewer fetches from the renderer's origin; the Bearer token, not the origin, gates access. - response.setHeader("access-control-allow-origin", "*"); - if (!authorizeBearer(request.headers.authorization, token)) return unauthorized(response); - responses.add(response); - response.on("close", () => responses.delete(response)); - pumpMjpeg(upstream, response, controllers).catch( - () => response.destroyed || response.destroy(), - ); + response.writeHead(status ?? 426).end(); }); server.on("upgrade", (request, socket, head) => { - if (!relay || videoRequestStatus(request.method, request.url)) - return destroySocket(socket, 400); + if (stopping) return destroySocket(socket, 400); + if (videoRequestStatus(request.method, request.url)) return destroySocket(socket, 400); if (!authorizeSubprotocols(request.headers["sec-websocket-protocol"], token)) return destroySocket(socket, 401); relay.handleUpgrade(request, socket, head, (downstream) => { @@ -64,14 +64,14 @@ export function startStreamProxy(raw: Device.StreamDescriptor): Promise((resolve, reject2) => { - server.once("error", reject2); + return new Promise((resolve, reject) => { + server.once("error", reject); server.listen(0, LOOPBACK_HOST, () => { - server.removeListener("error", reject2); + server.removeListener("error", reject); const address = server.address(); if (!address || typeof address === "string") { server.close(); - reject2(new Error("The stream proxy did not bind a TCP port.")); + reject(new Error("The stream proxy did not bind a TCP port.")); return; } resolve({ @@ -83,7 +83,7 @@ export function startStreamProxy(raw: Device.StreamDescriptor): Promise stop(server, relay, responses, controllers, sockets), + stop: () => (stopping ??= stop(server, relay, sockets, tasks)), }); }); }); @@ -91,7 +91,7 @@ export function startStreamProxy(raw: Device.StreamDescriptor): Promise safeEqual(protocol, expected)); + const offered = Array.isArray(header) ? header.join(",") : (header ?? ""); + return offered.split(",").some((protocol) => safeEqual(protocol.trim(), expected)); } -// Pipe the hub's MJPEG body to the viewer with one-chunk backpressure, abortable when the viewer disconnects. -async function pumpMjpeg( - upstream: Extract, - response: ServerResponse, - controllers: Set, -) { - const controller = new AbortController(); - controllers.add(controller); - response.on("close", () => controller.abort()); - let reader: ReadableStreamDefaultReader | undefined; - try { - const fetched = await fetch(upstream.url, { - method: "GET", - headers: { authorization: upstream.authorization }, - redirect: "error", - signal: controller.signal, - }); - if (!fetched.ok || !fetched.body) return reject(response, 502); - response.writeHead(200, { - "content-type": - fetched.headers.get("content-type") ?? "multipart/x-mixed-replace; boundary=frame", - "cache-control": "no-store", - }); - reader = fetched.body.getReader(); - while (!controller.signal.aborted) { - const chunk = await reader.read(); - if (chunk.done) break; - if (chunk.value.byteLength && !response.write(chunk.value)) - await once(response, "drain", { signal: controller.signal }); - } - } finally { - controllers.delete(controller); - controller.abort(); - await reader?.cancel().catch(() => {}); - if (!response.destroyed) response.end(); - } -} +const decodeReset = Schema.decodeUnknownOption( + Schema.fromJsonString(Schema.Struct({ type: Schema.Literal("reset-video") })), +); -const VideoReset = Schema.Struct({ type: Schema.Literal("reset-video") }); +function isVideoReset(data: RawData, isBinary: boolean): boolean { + return !isBinary && decodeReset(text(data))._tag === "Some"; +} // Forward video; a validated reset request becomes a fixed proxy-owned recovery message. function bridgeWebSocket( @@ -176,6 +143,12 @@ function bridgeWebSocket( }); const hubToken = upstream.subprotocol.slice("serve-emu.token.".length); let resetRequested = false; + const requestReset = () => { + if (hub.readyState === WebSocket.CONNECTING) resetRequested = true; + if (hub.readyState === WebSocket.OPEN) + hub.send(JSON.stringify({ type: "reset-video", ack: false })); + }; + const flow = new VideoFlow(MAX_BUFFERED_BYTES, requestReset); sockets.add(downstream); sockets.add(hub); const closeBoth = () => { @@ -186,7 +159,11 @@ function bridgeWebSocket( }; hub.on("message", (data: RawData, isBinary: boolean) => { if (downstream.readyState !== WebSocket.OPEN) return; - if (isBinary && downstream.bufferedAmount > MAX_BUFFERED_BYTES) return; + if (isBinary) { + const bytes = buffer(data); + const frame = parseFramePacket(bytes); + if (!frame || !flow.accept({ ...frame, bytes }, downstream.bufferedAmount)) return; + } downstream.send(isBinary ? data : text(data).replaceAll(hubToken, "[redacted]"), { binary: isBinary, }); @@ -194,40 +171,37 @@ function bridgeWebSocket( hub.on("close", closeBoth); hub.on("error", closeBoth); hub.on("open", () => { - if (resetRequested) hub.send(JSON.stringify({ type: "reset-video", ack: false })); + if (resetRequested) requestReset(); resetRequested = false; }); downstream.on("message", (data: RawData, isBinary: boolean) => { - if (isBinary) return; - const parsed = Schema.decodeUnknownOption(Schema.fromJsonString(VideoReset))(text(data)); - if (parsed._tag === "None") return; - if (hub.readyState === WebSocket.CONNECTING) { - resetRequested = true; - return; - } - if (hub.readyState === WebSocket.OPEN) - hub.send(JSON.stringify({ type: "reset-video", ack: false })); + if (!isVideoReset(data, isBinary)) return; + flow.reset(); + requestReset(); }); downstream.on("close", closeBoth); downstream.on("error", closeBoth); } function text(data: RawData): string { - return ( - Array.isArray(data) ? Buffer.concat(data) : Buffer.isBuffer(data) ? data : Buffer.from(data) - ).toString("utf8"); + return buffer(data).toString("utf8"); +} + +function buffer(data: RawData): Buffer { + return Array.isArray(data) + ? Buffer.concat(data) + : Buffer.isBuffer(data) + ? data + : Buffer.from(data); } // Close the listener and every connection and fetch the proxy owns, then resolve once the listener has shut down. async function stop( server: Server, - relay: WebSocketServer | undefined, - responses: Set, - controllers: Set, + relay: WebSocketServer, sockets: Set, + tasks: Set>, ) { - for (const controller of controllers) controller.abort(); - for (const response of responses) if (!response.destroyed) response.destroy(); await Promise.all( [...sockets].map( (socket) => @@ -238,47 +212,20 @@ async function stop( }), ), ); - if (relay) await new Promise((resolve) => relay.close(() => resolve())); + await Promise.all(tasks); + await new Promise((resolve) => relay.close(() => resolve())); server.closeAllConnections(); await new Promise((resolve) => server.close(() => resolve())); } -function subprotocols(header: string | string[] | undefined): string[] { - const value = Array.isArray(header) ? header.join(",") : (header ?? ""); - return value - .split(",") - .map((entry) => entry.trim()) - .filter(Boolean); -} - function safeEqual(left: string, right: string): boolean { const a = Buffer.from(left); const b = Buffer.from(right); return a.length === b.length && timingSafeEqual(a, b); } -function reject(response: ServerResponse, status: number) { - response.writeHead(status).end(); -} - -function preflight(response: ServerResponse, status: number | undefined) { - if (status) return reject(response, status); - response - .writeHead(204, { - "access-control-allow-origin": "*", - "access-control-allow-methods": "GET", - "access-control-allow-headers": "authorization", - "access-control-max-age": "600", - }) - .end(); -} - -function unauthorized(response: ServerResponse) { - response.writeHead(401, { "www-authenticate": "Bearer" }).end(); -} - function destroySocket(socket: Duplex, status: number) { const reason = status === 401 ? "Unauthorized" : "Bad Request"; - socket.write(`HTTP/1.1 ${status} ${reason}\r\nConnection: close\r\n\r\n`); - socket.destroy(); + socket.on("error", () => socket.destroy()); + socket.end(`HTTP/1.1 ${status} ${reason}\r\nConnection: close\r\n\r\n`, () => socket.destroy()); } diff --git a/src/video-flow.test.ts b/src/video-flow.test.ts new file mode 100644 index 0000000..66794b3 --- /dev/null +++ b/src/video-flow.test.ts @@ -0,0 +1,55 @@ +import { expect, test } from "bun:test"; +import { VideoFlow } from "./video-flow.js"; + +const key = { key: true, picture: true, bytes: new Uint8Array(10) }; +const delta = { key: false, picture: true, bytes: new Uint8Array(10) }; +const configuration = { key: false, picture: false, bytes: new Uint8Array(10) }; + +test("initial playback forwards configuration but waits for a keyframe before deltas", () => { + const flow = new VideoFlow(100, () => { + throw new Error("unexpected recovery"); + }); + expect(flow.accept(configuration, 0)).toBe(true); + expect(flow.accept(delta, 0)).toBe(false); + expect(flow.accept(key, 0)).toBe(true); + expect(flow.accept(delta, 0)).toBe(true); +}); + +test("backpressure includes the next packet and requests recovery only once after the buffer drains", () => { + let recoveries = 0; + const flow = new VideoFlow(100, () => recoveries++); + expect(flow.accept(key, 0)).toBe(true); + expect(flow.accept(delta, 95)).toBe(false); + expect(flow.accept(key, 100)).toBe(false); + expect(recoveries).toBe(0); + expect(flow.accept(delta, 0)).toBe(false); + expect(flow.accept(configuration, 0)).toBe(true); + expect(flow.accept(delta, 0)).toBe(false); + expect(recoveries).toBe(1); + expect(flow.accept(key, 0)).toBe(true); + expect(flow.accept(delta, 0)).toBe(true); + expect(flow.accept(delta, 100)).toBe(false); + expect(flow.accept(delta, 0)).toBe(false); + expect(recoveries).toBe(2); +}); + +test("a naturally arriving keyframe recovers without requesting an unnecessary reset", () => { + const flow = new VideoFlow(100, () => { + throw new Error("unexpected recovery"); + }); + flow.accept(key, 0); + flow.accept(delta, 100); + expect(flow.accept(key, 0)).toBe(true); + expect(flow.accept(delta, 0)).toBe(true); +}); + +test("an explicit viewer reset discards deltas without duplicating the reset request", () => { + const flow = new VideoFlow(100, () => { + throw new Error("unexpected recovery"); + }); + flow.accept(key, 0); + flow.reset(); + expect(flow.accept(delta, 0)).toBe(false); + expect(flow.accept(configuration, 0)).toBe(true); + expect(flow.accept(key, 0)).toBe(true); +}); diff --git a/src/video-flow.ts b/src/video-flow.ts new file mode 100644 index 0000000..cbeae25 --- /dev/null +++ b/src/video-flow.ts @@ -0,0 +1,36 @@ +// After dropping a dependent frame, forward configuration only until a new keyframe arrives. +export class VideoFlow { + private waitingKeyframe = true; + private needsRecovery = false; + + constructor( + private readonly maxBufferedBytes: number, + private readonly recover: () => void, + ) {} + + accept( + frame: { key: boolean; picture: boolean; bytes: Uint8Array }, + bufferedAmount: number, + ): boolean { + if (bufferedAmount + frame.bytes.byteLength > this.maxBufferedBytes) { + this.waitingKeyframe = true; + this.needsRecovery = true; + return false; + } + if (frame.key) { + this.waitingKeyframe = false; + this.needsRecovery = false; + return true; + } + if (this.needsRecovery) { + this.needsRecovery = false; + this.recover(); + } + return !frame.picture || !this.waitingKeyframe; + } + + reset(): void { + this.waitingKeyframe = true; + this.needsRecovery = false; + } +} diff --git a/src/viewer-decoder.ts b/src/viewer-decoder.ts new file mode 100644 index 0000000..89c43a9 --- /dev/null +++ b/src/viewer-decoder.ts @@ -0,0 +1,41 @@ +interface DecoderCallbacks { + current(): boolean; + output(frame: VideoFrame): void; + error(message: string): void; +} + +// Decoder outputs are always closed, including obsolete generations and failed canvas draws. +export async function createH264Decoder( + config: VideoDecoderConfig, + callbacks: DecoderCallbacks, +): Promise { + const support = await VideoDecoder.isConfigSupported(config); + if (!callbacks.current()) return; + if (!support.supported) + throw new Error(`H.264 codec ${config.codec} is not supported by this browser`); + const decoder = new VideoDecoder({ + output(frame) { + try { + if (callbacks.current()) callbacks.output(frame); + } catch (error) { + callbacks.error( + `Device video draw failed: ${error instanceof Error ? error.message : String(error)}`, + ); + } finally { + frame.close(); + } + }, + error(error) { + if (callbacks.current()) callbacks.error(`H.264 decoder failed: ${error.message}`); + }, + }); + try { + decoder.configure(support.config ?? config); + } catch (error) { + decoder.close(); + throw new Error( + `H.264 decoder configuration failed: ${error instanceof Error ? error.message : String(error)}`, + ); + } + return decoder; +} diff --git a/src/viewer.test.ts b/src/viewer.test.ts index 11fd761..4916645 100644 --- a/src/viewer.test.ts +++ b/src/viewer.test.ts @@ -1,32 +1,43 @@ -import { describe, expect, jest, test } from "bun:test"; -import { - annexBCodec, - contentPoint, - createViewer, - multipartBoundary, - parseFramePacket, - parseMjpegFrames, - sameScreenFrame, - screenFrame, -} from "./viewer.js"; +import { afterEach, describe, expect, jest, test } from "bun:test"; +import { contentPoint, createViewer, sameScreenFrame, screenFrame } from "./viewer.js"; +import { sps, pps, idr, delta, keyframe as annexB } from "../test/video.js"; -const annexB = new Uint8Array([0, 0, 0, 1, 0x67, 0x64, 0, 0x29, 0, 0, 1, 0x65, 0x88]); +const teardown: Array<() => void> = []; +afterEach(() => { + for (const cleanup of teardown.splice(0).reverse()) cleanup(); +}); + +async function settleDecoder(): Promise { + for (let step = 0; step < 4; step++) await Promise.resolve(); +} -function h264Fixture() { +function h264Fixture(platform: "android" | "ios" = "android") { jest.useFakeTimers(); const listeners = new Set(); const checks: Array>> = []; + const chunks: EncodedVideoChunkInit[] = []; + const decoders: Decoder[] = []; + const sockets: Socket[] = []; + const sent: string[] = []; + let outputEnabled = true; + let framesClosed = 0; class Socket extends EventTarget { static instance: Socket; static OPEN = 1; readyState = 1; closed = false; binaryType = ""; - constructor() { + constructor( + readonly url: URL, + readonly protocol: string, + ) { super(); Socket.instance = this; + sockets.push(this); + } + send(message: string) { + sent.push(message); } - send() {} close() { this.closed = true; } @@ -42,12 +53,26 @@ function h264Fixture() { } state = "unconfigured"; decodeQueueSize = 0; - constructor(private readonly callbacks: VideoDecoderInit) {} + constructor(private readonly callbacks: VideoDecoderInit) { + decoders.push(this); + } configure() { this.state = "configured"; } decode() { - this.callbacks.output({ displayWidth: 100, displayHeight: 200, close() {} } as VideoFrame); + if (outputEnabled) this.output(); + } + output() { + this.callbacks.output({ + displayWidth: 100, + displayHeight: 200, + close() { + framesClosed++; + }, + } as VideoFrame); + } + error() { + this.callbacks.error(new DOMException("obsolete decoder failure")); } close() { this.state = "closed"; @@ -61,7 +86,11 @@ function h264Fixture() { }, WebSocket: Socket, VideoDecoder: Decoder, - EncodedVideoChunk: class {}, + EncodedVideoChunk: class { + constructor(input: EncodedVideoChunkInit) { + chunks.push(input); + } + }, }; const originals = new Map( Object.keys(globals).map((key) => [key, Object.getOwnPropertyDescriptor(globalThis, key)]), @@ -78,7 +107,7 @@ function h264Fixture() { const viewer = createViewer({ canvas, stream: { - platform: "android", + platform, deviceId: "phone", baseUrl: "http://127.0.0.1:3400", token: "private-token", @@ -88,6 +117,14 @@ function h264Fixture() { onState: (state) => states.push(state), onError: (message) => errors.push(message), }); + teardown.push(() => { + viewer.destroy(); + jest.useRealTimers(); + for (const [key, original] of originals) { + if (original) Object.defineProperty(globalThis, key, original); + else Reflect.deleteProperty(globalThis, key); + } + }); viewer.active(true); return { viewer, @@ -96,91 +133,266 @@ function h264Fixture() { canvas, listeners, checks, + chunks, + decoders, + sent, + stall(value: boolean) { + outputEnabled = !value; + }, + framesClosed: () => framesClosed, socket: Socket.instance, - restore() { - viewer.destroy(); - jest.useRealTimers(); - for (const [key, original] of originals) { - if (original) Object.defineProperty(globalThis, key, original); - else Reflect.deleteProperty(globalThis, key); - } + sockets, + visibility(state: "visible" | "hidden") { + globals.document.visibilityState = state; + for (const listener of listeners) listener(new Event("visibilitychange")); }, }; } -test("unexpected Android video closure is visible instead of a silently blank panel", () => { - const item = h264Fixture(); - try { +test.each(["android", "ios"] as const)( + "%s uses the same authenticated H.264 viewer", + (platform) => { + const item = h264Fixture(platform); + expect(item.socket.url.toString()).toBe("ws://127.0.0.1:3400/video?frame-meta=1"); + expect(item.socket.protocol).toBe("opencode-device.private-token"); + }, +); + +describe("H.264 viewer", () => { + test.each([ + { + name: "separate configuration", + packets: [sps, pps, delta], + expected: new Uint8Array([...sps, 0, ...pps, ...idr]), + }, + { name: "SPS plus delta", packets: [new Uint8Array([...sps, ...delta])], expected: annexB }, + ])( + "$name waits for an IDR and includes cached parameters in its decoder data", + async ({ packets, expected }) => { + const item = h264Fixture(); + for (const data of packets) item.socket.emit("message", { data: data.slice().buffer }); + expect(item.checks).toHaveLength(0); + expect(item.chunks).toHaveLength(0); + item.socket.emit("message", { data: idr.slice().buffer }); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.chunks).toHaveLength(1); + expect(item.chunks[0]).toMatchObject({ type: "key", data: expected }); + expect(item.states.at(-1)).toBe("playing"); + }, + ); + + test("dropped packets during asynchronous setup request a new IDR instead of decoding dependent deltas", async () => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.slice().buffer }); + item.socket.emit("message", { data: delta.slice().buffer }); + item.socket.emit("message", { data: annexB.slice().buffer }); + expect(item.checks).toHaveLength(1); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.sent).toHaveLength(1); + expect(JSON.parse(item.sent[0]!)).toEqual({ type: "reset-video", ack: false }); + item.socket.emit("message", { data: delta.slice().buffer }); + expect(item.chunks).toHaveLength(1); + item.socket.emit("message", { data: idr.slice().buffer }); + expect(item.chunks).toHaveLength(2); + expect(item.checks).toHaveLength(1); + }); + + test("decoder backlog resets playback and rejects deltas until an IDR arrives", async () => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.slice().buffer }); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + const previous = item.decoders[0]!; + previous.decodeQueueSize = 3; + item.socket.emit("message", { data: delta.slice().buffer }); + expect(previous.state).toBe("closed"); + expect(item.states.at(-1)).toBe("connecting"); + const closed = item.framesClosed(); + previous.output(); + previous.error(); + expect(item.errors).toEqual([]); + expect(item.framesClosed()).toBe(closed + 1); + expect(item.states.at(-1)).toBe("connecting"); + item.socket.emit("message", { data: delta.slice().buffer }); + expect(item.chunks).toHaveLength(1); + item.socket.emit("message", { data: idr.slice().buffer }); + item.checks[1]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.states.at(-1)).toBe("playing"); + expect(item.chunks.at(-1)!.type).toBe("key"); + }); + + test("a running decoder stall requests recovery and then reports a bounded failure", async () => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.slice().buffer }); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + item.stall(true); + item.socket.emit("message", { data: delta.slice().buffer }); + jest.advanceTimersByTime(14_999); + item.socket.emit("message", { data: delta.slice().buffer }); + jest.advanceTimersByTime(1); + expect(item.states.at(-1)).toBe("connecting"); + expect(item.sent).toHaveLength(1); + expect(item.decoders[0]!.state).toBe("closed"); + jest.advanceTimersByTime(15_000); + expect(item.states.at(-1)).toBe("paused"); + expect(item.errors).toHaveLength(1); + expect(jest.getTimerCount()).toBe(0); + }); + + test("a static screen with no submitted video does not trigger a stall timeout", async () => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.slice().buffer }); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + jest.advanceTimersByTime(60_000); + expect(item.states.at(-1)).toBe("playing"); + expect(item.errors).toEqual([]); + expect(jest.getTimerCount()).toBe(0); + }); + + test("unexpected video closure is visible instead of a silently blank panel", () => { + const item = h264Fixture(); item.socket.emit("close", { code: 1006 }); expect(item.states).toEqual(["connecting", "paused"]); expect(item.errors[0]).toContain("connection closed (code 1006)"); expect(jest.getTimerCount()).toBe(0); - } finally { - item.restore(); - } -}); + }); -test("Android video startup has a deadline and stops failed connections", () => { - const item = h264Fixture(); - try { + test("video startup has a deadline and stops failed connections", () => { + const item = h264Fixture(); jest.advanceTimersByTime(15_000); expect(item.errors[0]).toContain("within 15 seconds"); expect(item.socket.closed).toBe(true); expect(item.states.at(-1)).toBe("paused"); expect(jest.getTimerCount()).toBe(0); - } finally { - item.restore(); - } -}); + }); -test("Android backend errors are shown once and redact the stream token", () => { - const item = h264Fixture(); - try { + test("backend errors are shown once and redact the stream token", () => { + const item = h264Fixture(); item.socket.emit("message", { data: JSON.stringify({ ok: false, error: "capture failed private-token" }), }); item.socket.emit("close", { code: 1006 }); - expect(item.errors).toEqual(["Android video failed: capture failed [redacted]"]); + expect(item.errors).toEqual(["Device video failed: capture failed [redacted]"]); expect(item.socket.closed).toBe(true); - } finally { - item.restore(); - } -}); + }); -test("Android ignores stale decoder checks, renders the current frame, and releases viewer resources", async () => { - const item = h264Fixture(); - try { - item.socket.emit("message", { data: annexB.slice().buffer }); - item.socket.emit("message", { - data: JSON.stringify({ type: "video-session", size: { width: 100, height: 200 } }), - }); - item.socket.emit("message", { data: annexB.slice().buffer }); - item.checks[0].reject(new Error("obsolete codec check")); - item.checks[1].resolve({ supported: true }); - await Promise.resolve(); - expect(item.states).toEqual(["connecting", "playing"]); - expect(item.canvas.width).toBe(100); - expect(item.canvas.height).toBe(200); + test.each([ + { trigger: "session", phase: "configuring" }, + { trigger: "session", phase: "playing" }, + { trigger: "codec", phase: "configuring" }, + { trigger: "codec", phase: "playing" }, + ])( + "$trigger replacement during $phase invalidates the previous decoder generation", + async ({ trigger, phase }) => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.buffer }); + if (phase === "playing") { + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + } + if (trigger === "session") { + item.socket.emit("message", { + data: JSON.stringify({ type: "video-session", size: { width: 100, height: 200 } }), + }); + item.socket.emit("message", { data: annexB.buffer }); + } else { + item.socket.emit("message", { + data: new Uint8Array([...sps.slice(0, 5), 0x4d, 0, 0x1f, ...idr]).buffer, + }); + } + expect(item.checks).toHaveLength(2); + expect(jest.getTimerCount()).toBe(1); + if (phase === "playing") { + expect(item.decoders[0]!.state).toBe("closed"); + item.decoders[0]!.error(); + } else if (trigger === "session") item.checks[0]!.reject(new Error("obsolete codec check")); + else item.checks[0]!.resolve({ supported: true }); + item.checks[1]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.decoders).toHaveLength(phase === "playing" ? 2 : 1); + expect(item.states).toEqual( + phase === "playing" + ? ["connecting", "playing", "connecting", "playing"] + : ["connecting", "playing"], + ); + expect(item.canvas).toMatchObject({ width: 100, height: 200 }); + expect(jest.getTimerCount()).toBe(0); + expect(item.errors).toEqual([]); + }, + ); + + test.each(["unsupported", "rejected"])( + "a %s codec check stops startup and reports one error", + async (failure) => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.buffer }); + if (failure === "unsupported") item.checks[0]!.resolve({ supported: false }); + else item.checks[0]!.reject(new Error("codec check failed")); + await settleDecoder(); + expect(item.errors).toHaveLength(1); + expect(item.states.at(-1)).toBe("paused"); + expect(item.socket.closed).toBe(true); + expect(item.decoders).toHaveLength(0); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + test("hiding the page releases capture and ignores a pending codec check", async () => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.buffer }); + item.visibility("hidden"); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.socket.closed).toBe(true); + expect(item.states.at(-1)).toBe("paused"); + expect(item.decoders).toHaveLength(0); expect(jest.getTimerCount()).toBe(0); - expect(item.errors).toEqual([]); - expect(item.listeners.size).toBe(1); - item.socket.emit("message", { - data: JSON.stringify({ type: "video-session", size: { width: 100, height: 200 } }), - }); - expect(jest.getTimerCount()).toBe(1); - item.socket.emit("message", { data: annexB.slice().buffer }); - item.checks[2].resolve({ supported: true }); - await Promise.resolve(); + item.visibility("visible"); + expect(item.states.at(-1)).toBe("connecting"); + }); + + test("a failed canvas draw closes the output frame and decoder", async () => { + const item = h264Fixture(); + item.canvas.getContext = () => null; + item.socket.emit("message", { data: annexB.buffer }); + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.framesClosed()).toBe(1); + expect(item.decoders[0]!.state).toBe("closed"); + expect(item.errors[0]).toContain("canvas is unavailable"); expect(jest.getTimerCount()).toBe(0); - item.viewer.destroy(); - expect(item.canvas.width).toBe(0); - expect(item.canvas.height).toBe(0); - expect(item.listeners.size).toBe(0); - item.socket.emit("close", { code: 1006 }); - expect(item.errors).toEqual([]); - } finally { - item.restore(); - } + }); + + test.each(["configuring", "playing"])( + "destroying a %s viewer is terminal and releases its resources", + async (phase) => { + const item = h264Fixture(); + item.socket.emit("message", { data: annexB.buffer }); + if (phase === "playing") { + item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + } + item.viewer.destroy(); + item.viewer.active(true); + item.visibility("visible"); + if (phase === "configuring") item.checks[0]!.resolve({ supported: true }); + await settleDecoder(); + expect(item.sockets).toHaveLength(1); + expect(item.decoders).toHaveLength(phase === "playing" ? 1 : 0); + if (phase === "playing") expect(item.decoders[0]!.state).toBe("closed"); + expect(item.states.at(-1)).toBe("idle"); + expect(item.listeners.size).toBe(0); + expect(item.canvas).toMatchObject({ width: 0, height: 0 }); + item.socket.emit("close", { code: 1006 }); + expect(item.errors).toEqual([]); + expect(jest.getTimerCount()).toBe(0); + }, + ); }); test("screen input requires a current rendered frame and rejects a changed gesture target", () => { @@ -199,95 +411,6 @@ test("screen input requires a current rendered frame and rejects a changed gestu expect(sameScreenFrame(first, first)).toBe(true); }); -describe("parseFramePacket", () => { - test("parses raw Annex-B and detects an IDR keyframe", () => { - const packet = parseFramePacket(annexB.slice().buffer); - expect(packet?.bytes).toEqual(annexB); - expect(packet?.key).toBe(true); - expect(packet?.timestamp).toBeGreaterThan(0); - }); - - test("parses SEMU v1 metadata", () => { - const data = framed(1, 1, 123_456n, annexB); - const packet = parseFramePacket(data.buffer); - expect(packet).toEqual({ - bytes: data.subarray(16), - key: true, - codec: "avc1.640029", - timestamp: 123_456, - serverTimestamp: undefined, - }); - }); - - test("parses the v2 server timestamp without treating it as dimensions", () => { - const data = framed(2, 0, 222n, annexB, 1_735_689_600_123_000n); - const packet = parseFramePacket(data.buffer); - expect(packet).toEqual({ - bytes: data.subarray(24), - key: true, - codec: "avc1.640029", - timestamp: 222, - serverTimestamp: 1_735_689_600_123, - }); - }); - - test.each([ - new Uint8Array(), - new Uint8Array([0, 0, 0, 1]), - framed(3, 0, 1n, annexB), - framed(2, 0, 1n, new Uint8Array()), - framed(1, 0, BigInt(Number.MAX_SAFE_INTEGER) + 1n, annexB), - new Uint8Array([0x53, 0x45, 0x4d, 0x55, 1]), - ])("rejects malformed framing", (data) => { - expect(parseFramePacket(data.buffer)).toBeUndefined(); - }); -}); - -test("annexBCodec reads an SPS across mixed start-code boundaries", () => { - expect( - annexBCodec( - new Uint8Array([0, 0, 0, 1, 0x68, 1, 2, 0, 0, 1, 0x67, 0x64, 0, 0x29, 0, 0, 0, 1, 0x65]), - ), - ).toBe("avc1.640029"); -}); - -describe("MJPEG multipart framing", () => { - test("reads the boundary and a Content-Length framed JPEG across chunks", () => { - expect(multipartBoundary('multipart/x-mixed-replace; boundary="frame"')).toBe("frame"); - const first = join( - new TextEncoder().encode("--frame\r\nContent-Type: image/jpeg\r\nContent-Length: 4\r\n\r\n"), - new Uint8Array([0xff, 0xd8]), - ); - const second = new Uint8Array([0xff, 0xd9, 13, 10]); - const partial = parseMjpegFrames(first, "frame"); - expect(partial.frames).toEqual([]); - const complete = parseMjpegFrames(join(partial.rest, second), "frame"); - expect(complete.frames).toEqual([new Uint8Array([0xff, 0xd8, 0xff, 0xd9])]); - }); - - test.each([ - null, - "image/jpeg", - "multipart/x-mixed-replace", - "multipart/x-mixed-replace; boundary=\r\ninvalid", - ])("rejects malformed content types", (contentType) => { - expect(multipartBoundary(contentType)).toBeUndefined(); - }); - - test("does not emit malformed parts", () => { - const malformed = new TextEncoder().encode( - "--frame\r\nContent-Type: text/plain\r\nContent-Length: 4\r\n\r\nnope\r\n", - ); - const validHeader = new TextEncoder().encode( - "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: 4\r\n\r\n", - ); - const data = join(join(malformed, validHeader), new Uint8Array([0xff, 0xd8, 0xff, 0xd9])); - expect(parseMjpegFrames(data, "frame").frames).toEqual([ - new Uint8Array([0xff, 0xd8, 0xff, 0xd9]), - ]); - }); -}); - describe("contentPoint", () => { const bounds = { left: 10, top: 20, width: 200, height: 200 }; const size = { width: 100, height: 50 }; @@ -311,29 +434,3 @@ describe("contentPoint", () => { expect(contentPoint(nextBounds, nextSize, { x: 100, y: 100 })).toBeUndefined(); }); }); - -function framed( - version: number, - flags: number, - pts: bigint, - payload: Uint8Array, - serverTimestamp = 0n, -) { - const header = version === 2 ? 24 : 16; - const data = new Uint8Array(header + payload.byteLength); - const view = new DataView(data.buffer); - view.setUint32(0, 0x53454d55); - view.setUint8(4, version); - view.setUint8(5, flags); - view.setBigUint64(8, pts); - if (version === 2) view.setBigUint64(16, serverTimestamp); - data.set(payload, header); - return data; -} - -function join(left: Uint8Array, right: Uint8Array) { - const data = new Uint8Array(left.byteLength + right.byteLength); - data.set(left); - data.set(right, left.byteLength); - return data; -} diff --git a/src/viewer.ts b/src/viewer.ts index 4bbbd2d..3f7737a 100644 --- a/src/viewer.ts +++ b/src/viewer.ts @@ -1,4 +1,6 @@ import type { Device } from "./rpc.js"; +import { H264Configuration, parseFramePacket, type FramePacket } from "./h264.js"; +import { createH264Decoder } from "./viewer-decoder.js"; export type ViewerState = "idle" | "connecting" | "playing" | "paused"; @@ -14,16 +16,7 @@ export function screenFrame( state: ViewerState, size: { width: number; height: number }, ): ScreenFrame | undefined { - if ( - !token || - token !== renderedToken || - state !== "playing" || - !Number.isSafeInteger(size.width) || - !Number.isSafeInteger(size.height) || - size.width <= 0 || - size.height <= 0 - ) - return; + if (!token || token !== renderedToken || state !== "playing" || !validVideoSize(size)) return; return { token, width: size.width, height: size.height }; } @@ -43,30 +36,29 @@ export interface ViewerOptions { readonly onError: (message: string) => void; } -export interface FramePacket { - readonly bytes: Uint8Array; - readonly key: boolean; - readonly codec?: string; - readonly timestamp: number; - readonly serverTimestamp?: number; +type Playback = { type: "idle" | "destroyed" } | { type: "running"; stop(): void }; +interface DecoderSetup { + type: "configuring"; + codec: string; + missed: boolean; } - -export interface MjpegFrames { - readonly frames: ReadonlyArray>; - readonly rest: Uint8Array; -} - -const MAGIC = 0x53454d55; -const FRAME_HEADERS: Readonly> = { 1: 16, 2: 24 }; -const MAX_JPEG_BUFFER = 16 * 1024 * 1024; +type H264State = + | { type: "waiting" | "stopped" } + | DecoderSetup + | { + type: "decoding" | "playing"; + setup: DecoderSetup; + decoder: VideoDecoder; + waitingKeyframe: boolean; + }; export function createViewer(options: ViewerOptions) { let active = false; - let stop: (() => void) | undefined; + let playback: Playback = { type: "idle" }; const start = () => { try { - return options.stream.protocol === "h264" ? playH264(options) : playMjpeg(options); + return playH264(options); } catch (error) { options.onState("paused"); options.onError(error instanceof Error ? error.message : String(error)); @@ -74,29 +66,31 @@ export function createViewer(options: ViewerOptions) { } }; const update = () => { + if (playback.type === "destroyed") return; const shouldPlay = active && document.visibilityState === "visible"; - if (shouldPlay && !stop) { - stop = start(); + if (shouldPlay && playback.type === "idle") { + playback = { type: "running", stop: start() }; return; } - if (shouldPlay || !stop) return; - stop(); - stop = undefined; + if (shouldPlay || playback.type !== "running") return; + playback.stop(); + playback = { type: "idle" }; options.onState(active ? "paused" : "idle"); }; - const visibility = () => update(); - document.addEventListener("visibilitychange", visibility); + document.addEventListener("visibilitychange", update); return { active(value: boolean) { + if (playback.type === "destroyed") return; active = value; update(); }, destroy() { - active = false; - document.removeEventListener("visibilitychange", visibility); - stop?.(); - stop = undefined; + if (playback.type === "destroyed") return; + const previous = playback; + playback = { type: "destroyed" }; + document.removeEventListener("visibilitychange", update); + if (previous.type === "running") previous.stop(); options.canvas.width = 0; options.canvas.height = 0; options.onState("idle"); @@ -104,50 +98,6 @@ export function createViewer(options: ViewerOptions) { }; } -export function parseFramePacket(input: ArrayBuffer): FramePacket | undefined { - const data = new Uint8Array(input); - if (data.byteLength < 4) return; - const view = new DataView(input); - if (view.getUint32(0) === MAGIC) return metadataFrame(data, view); - return videoFrame(data, performance.now() * 1_000, false); -} - -function metadataFrame(data: Uint8Array, view: DataView): FramePacket | undefined { - if (data.byteLength < 16) return; - const header = FRAME_HEADERS[view.getUint8(4)]; - if (!header || data.byteLength <= header) return; - const pts = view.getBigUint64(8); - if (pts > BigInt(Number.MAX_SAFE_INTEGER)) return; - return videoFrame( - data.subarray(header), - Number(pts), - (view.getUint8(5) & 1) === 1, - header === 24 ? Number(view.getBigUint64(16)) / 1_000 : undefined, - ); -} - -function videoFrame( - bytes: Uint8Array, - timestamp: number, - key: boolean, - serverTimestamp?: number, -): FramePacket | undefined { - const start = nalStart(bytes, 0); - if (start === undefined || start >= bytes.byteLength) return; - const info = annexBInfo(bytes); - return { - bytes, - key: key || info.key, - codec: info.codec, - timestamp, - serverTimestamp, - }; -} - -export function annexBCodec(data: Uint8Array) { - return annexBInfo(data).codec; -} - export function contentPoint( bounds: Pick, size: { readonly width: number; readonly height: number }, @@ -179,86 +129,6 @@ export function contentPoint( }; } -export function multipartBoundary(contentType: string | null) { - if (!contentType || /[\r\n]/.test(contentType)) return; - const parts = contentType.split(";"); - if (parts.shift()?.trim().toLowerCase() !== "multipart/x-mixed-replace") return; - const value = parts - .map((part) => part.trim()) - .find((part) => part.toLowerCase().startsWith("boundary=")) - ?.slice("boundary=".length) - .trim(); - if (!value) return; - const boundary = value.startsWith('"') && value.endsWith('"') ? value.slice(1, -1) : value; - const normalized = boundary.startsWith("--") ? boundary.slice(2) : boundary; - if (!normalized) return; - return normalized; -} - -export function parseMjpegFrames(data: Uint8Array, boundary: string): MjpegFrames { - const delimiter = new TextEncoder().encode(`--${boundary}`); - const separator = new Uint8Array([13, 10, 13, 10]); - const frames: Uint8Array[] = []; - let cursor = 0; - let consumed = 0; - while (cursor < data.byteLength) { - const start = findBytes(data, delimiter, cursor); - if (start < 0) break; - const line = start + delimiter.byteLength; - if (line + 2 > data.byteLength) return { frames, rest: data.slice(start) }; - if (!hasCrlf(data, line)) { - cursor = line; - continue; - } - const headerEnd = findBytes(data, separator, line + 2); - if (headerEnd < 0) return { frames, rest: data.slice(start) }; - const headers = new TextDecoder("ascii").decode(data.subarray(line + 2, headerEnd)); - const bytes = jpegLength(headers); - const frameStart = headerEnd + separator.byteLength; - if (!bytes) { - cursor = headerEnd + separator.byteLength; - continue; - } - const frameEnd = frameStart + bytes; - if (frameEnd > data.byteLength) return { frames, rest: data.slice(start) }; - const frame = data.subarray(frameStart, frameEnd); - if (isJpeg(frame)) frames.push(frame); - consumed = frameEnd; - cursor = frameEnd; - } - return { frames, rest: data.slice(consumed) }; -} - -function hasCrlf(bytes: Uint8Array, offset: number): boolean { - return bytes[offset] === 13 && bytes[offset + 1] === 10; -} - -function jpegLength(headers: string): number | undefined { - const length = headers.match(/(?:^|\r\n)Content-Length:\s*(\d+)(?:\r\n|$)/i)?.[1]; - const contentType = headers - .match(/(?:^|\r\n)Content-Type:\s*([^\r\n]+)(?:\r\n|$)/i)?.[1] - .trim() - .toLowerCase(); - const bytes = length ? Number(length) : 0; - if ( - Number.isSafeInteger(bytes) && - bytes > 0 && - bytes <= MAX_JPEG_BUFFER && - contentType === "image/jpeg" - ) - return bytes; -} - -function isJpeg(frame: Uint8Array): boolean { - return ( - frame.byteLength >= 4 && - frame[0] === 0xff && - frame[1] === 0xd8 && - frame.at(-2) === 0xff && - frame.at(-1) === 0xd9 - ); -} - function playH264(options: ViewerOptions) { if (typeof VideoDecoder !== "function" || typeof EncodedVideoChunk !== "function") { options.onState("paused"); @@ -268,18 +138,15 @@ function playH264(options: ViewerOptions) { const url = new URL("/video?frame-meta=1", new URL(options.stream.baseUrl).origin); url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; const socket = new WebSocket(url, `opencode-device.${options.stream.token}`); - let decoder: VideoDecoder | undefined; - let codec: string | undefined; - let configuring = 0; - let waitingKeyframe = true; - let stopped = false; - let playing = false; + let state: H264State = { type: "waiting" }; + const configuration = new H264Configuration(); let startupTimer: ReturnType | undefined; + let decodeTimer: ReturnType | undefined; const waitForFrame = () => { clearTimeout(startupTimer); startupTimer = setTimeout( () => - fail("Android video did not produce a screen within 15 seconds. Refresh devices to retry."), + fail("Device video did not produce a screen within 15 seconds. Refresh devices to retry."), 15_000, ); }; @@ -293,121 +160,130 @@ function playH264(options: ViewerOptions) { } catch {} }; const reset = (request = false) => { - configuring++; - const previous = decoder; - const wasPlaying = playing; - decoder = undefined; - codec = undefined; - waitingKeyframe = true; - playing = false; - if (previous && previous.state !== "closed") previous.close(); - if (request && wasPlaying && !stopped) { + clearTimeout(decodeTimer); + decodeTimer = undefined; + const previous = state; + state = { type: "waiting" }; + if ("decoder" in previous && previous.decoder.state !== "closed") previous.decoder.close(); + if (request && previous.type === "playing") { waitForFrame(); options.onState("connecting"); } if (request) requestKeyframe(); }; - const fail = (message: string) => { - if (stopped) return; - stopped = true; + const stop = () => { + if (state.type === "stopped") return; + reset(); + state = { type: "stopped" }; clearTimeout(startupTimer); socket.close(); - reset(); + }; + const fail = (message: string) => { + if (state.type === "stopped") return; + stop(); options.onError(message.replaceAll(options.stream.token, "[redacted]").slice(0, 512)); options.onState("paused"); }; const decode = (frame: FramePacket, target: VideoDecoder) => { - if (stopped || decoder !== target || target.state !== "configured") return; + if (!("decoder" in state) || state.decoder !== target || target.state !== "configured") return; + state.waitingKeyframe = false; try { + decodeTimer ??= setTimeout(() => { + decodeTimer = undefined; + reset(true); + }, 15_000); target.decode( new EncodedVideoChunk({ type: frame.key ? "key" : "delta", timestamp: frame.timestamp, - data: frame.bytes, + data: configuration.bytes(frame), }), ); - waitingKeyframe = false; } catch (error) { fail(`H.264 decode failed: ${error instanceof Error ? error.message : String(error)}`); } }; const configure = (frame: FramePacket, nextCodec: string) => { reset(); - codec = nextCodec; - const generation = configuring; + const setup: DecoderSetup = { type: "configuring", codec: nextCodec, missed: false }; + state = setup; + const current = () => state === setup || ("decoder" in state && state.setup === setup); const config: VideoDecoderConfig = { codec: nextCodec, optimizeForLatency: true, hardwareAcceleration: "prefer-hardware", }; - void VideoDecoder.isConfigSupported(config) - .then((support) => { - if (stopped || generation !== configuring) return; - if (!support.supported) { - fail(`H.264 codec ${nextCodec} is not supported by this browser`); - return; - } - const target: VideoDecoder = new VideoDecoder({ - output(output) { - if (stopped || decoder !== target) { - output.close(); - return; - } - try { - draw(options, output, output.displayWidth, output.displayHeight); - if (!playing) { - playing = true; - clearTimeout(startupTimer); - options.onState("playing"); - } - } catch (error) { - fail( - `Device video draw failed: ${error instanceof Error ? error.message : String(error)}`, - ); - } finally { - output.close(); - } - }, - error(error) { - if (stopped || decoder !== target) return; - fail(`H.264 decoder failed: ${error.message}`); - }, - }); - try { - target.configure(support.config ?? config); - } catch (error) { + void createH264Decoder(config, { + current, + output(output) { + if (!("decoder" in state)) return; + const previous = state; + clearTimeout(decodeTimer); + decodeTimer = undefined; + draw(options, output, output.displayWidth, output.displayHeight); + if (state !== previous || previous.type === "playing") return; + state = { ...previous, type: "playing" }; + clearTimeout(startupTimer); + options.onState("playing"); + }, + error: fail, + }) + .then((target) => { + if (!target) return; + if (state !== setup) { target.close(); - fail( - `H.264 decoder configuration failed: ${error instanceof Error ? error.message : String(error)}`, - ); return; } - decoder = target; + state = { type: "decoding", setup, decoder: target, waitingKeyframe: true }; decode(frame, target); + if (current() && "decoder" in state && setup.missed) { + state.waitingKeyframe = true; + requestKeyframe(); + } }) .catch((error) => { - if (stopped || generation !== configuring) return; + if (!current()) return; fail( `H.264 decoder setup failed: ${error instanceof Error ? error.message : String(error)}`, ); }); }; - const receive = (frame: FramePacket) => { - if (decoder && decoder.decodeQueueSize > 2) return reset(true); - if (waitingKeyframe && !frame.key) return; - const nextCodec = frame.codec ?? codec; + const decoderCodec = () => + state.type === "configuring" ? state.codec : "decoder" in state ? state.setup.codec : undefined; + const updateConfiguration = (frame: FramePacket) => { + configuration.update(frame); + const codec = decoderCodec(); + if (codec && configuration.codec !== codec) reset(true); + }; + const decodeOrConfigure = (frame: FramePacket) => { + const nextCodec = configuration.codec ?? decoderCodec(); if (!nextCodec) return requestKeyframe(); - if (decoder && nextCodec === codec) return decode(frame, decoder); + if ("decoder" in state) return decode(frame, state.decoder); + if (!frame.key) return reset(true); configure(frame, nextCodec); }; + const receive = (frame: FramePacket) => { + updateConfiguration(frame); + if (!frame.picture) return; + if (state.type === "configuring") { + state.missed = true; + return; + } + if ("decoder" in state && state.decoder.decodeQueueSize > 2) return reset(true); + if (!frame.key && (!("decoder" in state) || state.waitingKeyframe)) return; + decodeOrConfigure(frame); + }; waitForFrame(); socket.addEventListener("open", requestKeyframe); socket.addEventListener("message", (event) => { - if (stopped) return; + if (state.type === "stopped") return; if (typeof event.data === "string") { const message = videoMessage(event.data); - if (message?.type === "error") fail(`Android video failed: ${message.message}`); - if (message?.type === "reset") reset(true); + if (message?.type === "error") fail(`Device video failed: ${message.message}`); + if (message?.type === "reset") { + configuration.clear(); + reset(true); + } return; } if (!(event.data instanceof ArrayBuffer)) return; @@ -419,80 +295,11 @@ function playH264(options: ViewerOptions) { fail(`Device video connection closed (code ${event.code}). Refresh devices to retry.`); }); - return () => { - stopped = true; - clearTimeout(startupTimer); - socket.close(); - reset(); - }; -} - -function playMjpeg(options: ViewerOptions) { - const abort = new AbortController(); - let reader: ReadableStreamDefaultReader | undefined; - let stopped = false; - let playing = false; - const render = async (bytes: Uint8Array) => { - const bitmap = await createImageBitmap(new Blob([bytes], { type: "image/jpeg" })); - try { - if (stopped) return; - draw(options, bitmap, bitmap.width, bitmap.height); - if (!playing) { - playing = true; - options.onState("playing"); - } - } finally { - bitmap.close(); - } - }; - options.onState("connecting"); - void fetch(new URL("/video", new URL(options.stream.baseUrl).origin), { - method: "GET", - headers: { Authorization: `Bearer ${options.stream.token}` }, - signal: abort.signal, - }) - .then(async (response) => { - if (!response.ok || !response.body) - throw new Error(`Device video returned ${response.status}`); - const boundary = multipartBoundary(response.headers.get("content-type")); - if (!boundary) throw new Error("Device video returned an invalid MJPEG content type"); - reader = response.body.getReader(); - let pending = new Uint8Array(); - while (!stopped) { - const next = await reader.read(); - if (next.done) break; - pending = append(pending, next.value); - const frames = parseMjpegFrames(pending, boundary); - pending = frames.rest; - const bytes = frames.frames.at(-1); - if (bytes) await render(bytes); - if (pending.byteLength > MAX_JPEG_BUFFER) - throw new Error("Device MJPEG stream exceeded 16 MiB without a complete frame"); - } - if (!stopped) throw new Error("Device video stream ended. Refresh devices to retry."); - }) - .catch((error) => { - if (!stopped) { - options.onState("paused"); - options.onError(error instanceof Error ? error.message : String(error)); - } - }) - .finally(async () => { - abort.abort(); - await reader?.cancel().catch(() => {}); - reader?.releaseLock(); - reader = undefined; - }); - - return () => { - stopped = true; - abort.abort(); - void reader?.cancel().catch(() => {}); - }; + return stop; } function draw(options: ViewerOptions, source: CanvasImageSource, width: number, height: number) { - if (!Number.isSafeInteger(width) || !Number.isSafeInteger(height) || width <= 0 || height <= 0) + if (!validVideoSize({ width, height })) throw new Error("Device video produced invalid screen dimensions"); if (options.canvas.width !== width || options.canvas.height !== height) { options.canvas.width = width; @@ -545,48 +352,3 @@ function validVideoSize(size: object): boolean { size.height > 0 ); } - -// Inspect once per packet. An SPS supplies both the codec and keyframe marker, so most keyframes stop at their header. -function annexBInfo(data: Uint8Array): { key: boolean; codec?: string } { - let key = false; - for (let index = 0; index < data.byteLength - 2; index++) { - const start = nalStart(data, index); - if (start === undefined) continue; - const type = data[start] & 0x1f; - if (type === 7) return { key: true, codec: spsCodec(data.subarray(start, start + 4)) }; - if (type === 5) key = true; - index = start; - } - return { key }; -} - -function nalStart(data: Uint8Array, index: number): number | undefined { - if (data[index] !== 0 || data[index + 1] !== 0) return; - if (data[index + 2] === 1) return index + 3; - if (data[index + 2] === 0 && data[index + 3] === 1) return index + 4; -} - -function spsCodec(sps: Uint8Array): string | undefined { - if (sps.byteLength < 4) return; - return `avc1.${[sps[1], sps[2], sps[3]].map((byte) => byte.toString(16).padStart(2, "0")).join("")}`; -} - -function append(left: Uint8Array, right: Uint8Array) { - const combined = new Uint8Array(left.byteLength + right.byteLength); - combined.set(left); - combined.set(right, left.byteLength); - return combined; -} - -function findBytes(data: Uint8Array, needle: Uint8Array, offset: number) { - for (let index = offset; index <= data.byteLength - needle.byteLength; index++) { - let matches = true; - for (let cursor = 0; cursor < needle.byteLength; cursor++) { - if (data[index + cursor] === needle[cursor]) continue; - matches = false; - break; - } - if (matches) return index; - } - return -1; -} diff --git a/test/video.ts b/test/video.ts new file mode 100644 index 0000000..0afa910 --- /dev/null +++ b/test/video.ts @@ -0,0 +1,57 @@ +export const sps = new Uint8Array([0, 0, 0, 1, 0x67, 0x64, 0, 0x29]); +export const pps = new Uint8Array([0, 0, 1, 0x68, 0xeb]); +export const idr = new Uint8Array([0, 0, 0, 1, 0x65, 0x88]); +export const delta = new Uint8Array([0, 0, 1, 0x41, 0x99]); +export const keyframe = new Uint8Array([...sps, ...idr]); +export const startCode = Buffer.from([0, 0, 0, 1]); + +export function avccRecord(tag: number, payload: Uint8Array): Buffer { + const header = Buffer.alloc(5); + header.writeUInt32BE(payload.byteLength + 1); + header[4] = tag; + return Buffer.concat([header, payload]); +} + +export function avccDescription(lengthBytes = 4): Buffer { + return avccRecord( + 1, + Buffer.concat([ + Buffer.from([1, 0x64, 0, 0x29, 0xfc | (lengthBytes - 1), 0xe1, 0, 4]), + sps.subarray(4), + Buffer.from([1, 0, 2]), + pps.subarray(3), + ]), + ); +} + +export function avccSample(tag: number, nals: readonly Uint8Array[], lengthBytes = 4): Buffer { + return avccRecord( + tag, + Buffer.concat( + nals.flatMap((nal) => { + const length = Buffer.alloc(lengthBytes); + length.writeUIntBE(nal.byteLength, 0, lengthBytes); + return [length, nal]; + }), + ), + ); +} + +export function metadataFrame( + version: number, + flags: number, + pts: bigint, + payload: Uint8Array, + serverTimestamp = 0n, +): Uint8Array { + const header = version === 2 ? 24 : 16; + const data = new Uint8Array(header + payload.byteLength); + const view = new DataView(data.buffer); + view.setUint32(0, 0x53454d55); + view.setUint8(4, version); + view.setUint8(5, flags); + view.setBigUint64(8, pts); + if (version === 2) view.setBigUint64(16, serverTimestamp); + data.set(payload, header); + return data; +}