import { CacheHandler , IgnoreCacheError } from "@mediaurl/cache" ; import { TaskRequest , TaskResponse } from "@mediaurl/schema" ; import { EventEmitter } from "events" ; import { v4 as uuid4 } from "uuid" ; import { AddonClass } from "../addon" ; import { SilentError } from "../errors" ; import { SendResponseFn } from "../types" ; export class Responder { queue : string [ ] ; emitter : EventEmitter ; sendResponse : null | SendResponseFn ; constructor ( fn ) { this . queue = [ ] ; this . emitter = new EventEmitter ( ) ; this . sendResponse = fn ; } async send ( statusCode , body , queueTimeout = 30 * 1000 ) { const id = uuid4 ( ) . toString ( ) ; this . queue . push ( id ) ; if ( this . queue [ 0 ] !== id ) { console . debug ( `Task queue length ${ this . queue . length }, waiting...` ) ; await new Promise < void > ( ( resolve , reject ) => { const on = ( ) => { if ( this . queue [ 0 ] === id ) { this . emitter . removeListener ( "event" , on ) ; resolve ( ) ; } } ; this . emitter . addListener ( "event" , on ) ; setTimeout ( ( ) => { this . emitter . removeListener ( "event" , on ) ; const i = this . queue . indexOf ( id ) ; if ( i !== - 1 ) this . queue . splice ( i , 1 ) ; reject ( new IgnoreCacheError ( "Waiting for task slot timed out" ) ) ; } , queueTimeout ) ; } ) ; } const sendResponse = this . sendResponse ; if ( sendResponse === null ) { console . warn ( "Send response is not set, client connection is broken" ) ; } else { this . sendResponse = null ; await sendResponse ( statusCode , body ) ; } return id ; } setSendResponse ( id , fn ) { if ( this . queue [ 0 ] !== id ) { throw new Error ( `First queue element is not the current id` ) ; } this . queue . shift ( ) ; this . sendResponse = fn ; this . emitter . emit ( "event" ) ; } } export const sendTask = async ( testMode , responder , cache , taskRequestData , timeout = 30 * 1000 ) : Promise < TaskResponse [ "data" ] > => { if ( testMode ) { throw new Error ( `Can not run client task "${ taskRequestData . type }" in test mode` ) ; } const task = { kind : "taskRequest" , id : uuid4 ( ) . toString ( ) , data : taskRequestData , } ; await cache . set ( `task.wait-${ task . id }` , "1" , timeout * 2 ) ; const id = await responder . send ( 200 , task ) ; let data ; try { data = await cache . waitKey ( `task.response-${ task . id }` , timeout , true ) ; } catch ( error ) { if ( error . message === "Wait timed out" ) { console . warn ( `Task ${ task . id } timed out, destroying connection to client` ) ; responder . setSendResponse ( id , null ) ; throw new SilentError ( "Task timed out" ) ; } throw error ; } const { responseChannel , response } = JSON . parse ( data ) ; responder . setSendResponse ( id , async ( statusCode , body ) => { const data = JSON . stringify ( { statusCode , body } ) ; await cache . set ( `task.response-${ responseChannel }` , data , timeout ) ; } ) ; if ( response . type === "error" ) throw new Error ( `Client error: ${ response . error }` ) ; return response ; } ; type HandleTaskProps = { cache : CacheHandler ; addon : AddonClass ; timeout ? : number ; input : any ; sendResponse : SendResponseFn ; } ; export const handleTask = async ( { cache , addon , timeout = 120 * 1000 , input , sendResponse , } ) => { cache = cache . clone ( { prefix : addon . getId ( ) , ... addon . getDefaultCacheOptions ( ) , } ) ; const task = input ; if ( ! ( await cache . get ( `task.wait-${ task . id }` ) ) ) { console . warn ( `Task ${ task . id } not found or timed out` ) ; await sendResponse ( 500 , { error : "Task not found or timed out" } ) ; return ; } await cache . delete ( `task.wait-${ task . id }` ) ; const responseChannel = uuid4 ( ) . toString ( ) ; const raw = JSON . stringify ( { responseChannel , response : task . data } ) ; await cache . set ( `task.response-${ task . id }` , raw ) ; const data = await cache . waitKey ( `task.response-${ responseChannel }` , timeout , true ) ; const { statusCode , body } = JSON . parse ( data ) ; await sendResponse ( statusCode , body ) ; } ;