import assert from "assert" ; import { Request } from "@miniflare/core" ; import type { FetchEvent } from "@miniflare/core" ; import { Compatibility , NoOpLog , PluginContext } from "@miniflare/shared" ; import { noop , triggerPromise , useMiniflare , useServer , } from "@miniflare/shared-test" ; import { CloseEvent , MessageEvent , WebSocket , WebSocketPair , WebSocketPlugin , } from "@miniflare/web-sockets" ; import test from "ava" ; const log = new NoOpLog ( ) ; const compat = new Compatibility ( ) ; const rootPath = process . cwd ( ) ; const ctx = { log , compat , rootPath , globalAsyncIO : true } ; test ( "WebSocketPlugin: setup: includes WebSocket stuff in globals" , ( t ) => { const plugin = new WebSocketPlugin ( ctx ) ; const globals = plugin . setup ( ) . globals ! ; t . is ( globals . MessageEvent , MessageEvent ) ; t . is ( globals . CloseEvent , CloseEvent ) ; t . is ( globals . WebSocketPair , WebSocketPair ) ; t . true ( typeof globals . WebSocket === "function" ) ; t . is ( globals . fetch , plugin . fetch ) ; } ) ; test ( "WebSocketPlugin: setup: fetch refuses unknown protocols if compatibility flag enabled" , async ( t ) => { const compat = new Compatibility ( undefined , [ "fetch_refuses_unknown_protocols" , ] ) ; const plugin = new WebSocketPlugin ( { log , compat , rootPath , globalAsyncIO : true , } ) ; const { globals } = await plugin . setup ( ) ; const upstream = ( await useServer ( t , ( req , res ) => res . end ( "upstream" ) ) ) . http ; upstream . protocol = "ftp:" ; await t . throwsAsync ( async ( ) => globals ?. fetch ( upstream ) , { instanceOf : TypeError , message : `Fetch API cannot load: ${ upstream . toString ( ) }` , } ) ; } ) ; test ( "WebSocketPlugin: setup: fetch throws outside request handler unless globalAsyncIO set" , async ( t ) => { const upstream = ( await useServer ( t , ( req , res ) => res . end ( "upstream" ) ) ) . http ; let plugin = new WebSocketPlugin ( { log , compat , rootPath } ) ; let { globals } = await plugin . setup ( ) ; await t . throwsAsync ( globals ?. fetch ( upstream ) , { instanceOf : Error , message : /^Some functionality, such as asynchronous I\/O/ , } ) ; plugin = new WebSocketPlugin ( { log , compat , rootPath , globalAsyncIO : true } ) ; globals = ( await plugin . setup ( ) ) . globals ; await globals ?. fetch ( upstream ) ; } ) ; test ( "WebSocketPlugin: setup: blocks direct WebSocket construction" , ( t ) => { const plugin = new WebSocketPlugin ( ctx ) ; const WebSocketProxy = plugin . setup ( ) . globals ! . WebSocket ; t . throws ( ( ) => new WebSocketProxy ( ) , { instanceOf : Error , message : "Failed to construct 'WebSocket': the constructor is not implemented." , } ) ; t . throws ( ( ) => WebSocketProxy ( ) , { instanceOf : Error , message : "Failed to construct 'WebSocket': the constructor is not implemented." , } ) ; const pair = new WebSocketPair ( ) ; t . true ( pair [ 0 ] instanceof WebSocketProxy ) ; } ) ; test ( "WebSocketPlugin: fetch, reload, dispose: closes WebSockets" , async ( t ) => { const plugin = new WebSocketPlugin ( ctx ) ; let [ eventTrigger , eventPromise ] = triggerPromise < { code : number ; reason : string ; } > ( ) ; const server = await useServer ( t , noop , ( ws ) => { ws . addEventListener ( "close" , eventTrigger ) ; } ) ; let res = await plugin . fetch ( server . ws , { headers : { upgrade : "websocket" } , } ) ; let webSocket = res . webSocket ; t . not ( webSocket , undefined ) ; assert ( webSocket ) ; webSocket . accept ( ) ; plugin . reload ( ) ; let event = await eventPromise ; t . is ( event . code , 1012 ) ; t . is ( event . reason , "Service Restart" ) ; [ eventTrigger , eventPromise ] = triggerPromise ( ) ; res = await plugin . fetch ( server . ws , { headers : { upgrade : "websocket" } , } ) ; webSocket = res . webSocket ; t . not ( webSocket , undefined ) ; assert ( webSocket ) ; webSocket . accept ( ) ; plugin . dispose ( ) ; event = await eventPromise ; t . is ( event . code , 1012 ) ; t . is ( event . reason , "Service Restart" ) ; } ) ; test ( "WebSocketPlugin: fetch, reload: ignores already closed WebSockets" , async ( t ) => { const plugin = new WebSocketPlugin ( ctx ) ; const [ eventTrigger , eventPromise ] = triggerPromise < { code : number ; reason : string ; } > ( ) ; const server = await useServer ( t , noop , ( ws ) => { ws . addEventListener ( "close" , eventTrigger ) ; } ) ; const res = await plugin . fetch ( server . ws , { headers : { upgrade : "websocket" } , } ) ; const webSocket = res . webSocket ; t . not ( webSocket , undefined ) ; assert ( webSocket ) ; webSocket . accept ( ) ; webSocket . close ( 1000 , "Test Closure" ) ; plugin . reload ( ) ; const event = await eventPromise ; t . is ( event . code , 1000 ) ; t . is ( event . reason , "Test Closure" ) ; } ) ; test ( "MiniflareCore: sends and responds to web socket messages" , async ( t ) => { const script = `(${ ( ( ) => { const sandbox = globalThis as any ; sandbox . addEventListener ( "fetch" , ( e ) => { const [ client , worker ] = Object . values ( new sandbox . WebSocketPair ( ) ) ; worker . accept ( ) ; worker . addEventListener ( "message" , ( e ) => { worker . send ( e . data ) ; } ) ; worker . send ( "hello client" ) ; e . respondWith ( new sandbox . Response ( null , { status : 101 , webSocket : client , } ) ) ; } ) ; } ) . toString ( ) })()` ; const mf = useMiniflare ( { WebSocketPlugin } , { script } ) ; const res = await mf . dispatchFetch ( new Request ( "http://localhost:8787/" ) ) ; t . not ( res . webSocket , undefined ) ; assert ( res . webSocket ) ; const [ eventTrigger , eventPromise ] = triggerPromise < void > ( ) ; const messages = [ ] ; res . webSocket . addEventListener ( "message" , ( e ) => { messages . push ( e . data ) ; if ( e . data === "hello worker" ) eventTrigger ( ) ; } ) ; res . webSocket . accept ( ) ; res . webSocket . send ( "hello worker" ) ; await eventPromise ; t . deepEqual ( messages , [ "hello client" , "hello worker" ] ) ; } ) ;