import { AdbClient } from './AdbClient' ; import { Message } from './message' ; import { Options } from './Options' ; import { toHex32 } from './Helpers' ; import { SyncFrame } from './SyncFrame' ; import { AsyncBlockingQueue } from './Queues' ; export class Stream { private static nextId = 1 ; private messageQueue = new AsyncBlockingQueue < Message > ( ) ; constructor ( readonly client , readonly service , readonly localId , readonly remoteId , private options ) { } async close ( ) { await this . write ( 'CLSE' ) ; if ( this . options . debug ) { console . log ( `Closed stream ${ this . service }` ) ; console . log ( ` local_id: 0x${ toHex32 ( this . localId ) }` ) ; console . log ( ` remote_id: 0x${ toHex32 ( this . remoteId ) }` ) ; } this . client . unregisterStream ( this ) ; } consumeMessage ( msg ) { if ( msg . header . arg0 === 0 || msg . header . arg0 !== this . remoteId || msg . header . arg1 === 0 || msg . header . arg1 !== this . localId ) { return false ; } this . messageQueue . enqueue ( msg ) ; return true ; } async write ( cmd , data ? ) { const message = this . newMessage ( cmd , data ) ; await this . client . sendMessage ( message ) ; } async read ( ) { return this . messageQueue . dequeue ( ) ; } param { string } remotePath returns { Promise < Blob > } async pull ( remotePath ) { const encoder = new TextEncoder ( ) ; const encodedFilename = encoder . encode ( remotePath ) ; const recvFrame = new SyncFrame ( 'RECV' , encodedFilename . byteLength ) ; const wrteRecvMessage = this . newMessage ( 'WRTE' , recvFrame . toDataView ( ) ) ; await this . client . sendMessage ( wrteRecvMessage ) ; const wrteRecvResponse = await this . read ( ) ; if ( wrteRecvResponse . header . cmd !== 'OKAY' ) { throw new Error ( 'WRTE/RECV failed: ' + wrteRecvResponse ) ; } const wrteFilenameMessage = this . newMessage ( 'WRTE' , new DataView ( encodedFilename . buffer ) ) ; await this . client . sendMessage ( wrteFilenameMessage ) ; const wrteFilenameResponse = await this . read ( ) ; if ( wrteFilenameResponse . header . cmd !== 'OKAY' ) { throw new Error ( 'WRTE/filename failed: ' + wrteFilenameResponse ) ; } const okayMessage = this . newMessage ( 'OKAY' ) ; let fileDataMessage = await this . read ( ) ; await this . client . sendMessage ( okayMessage ) ; let syncFrame = SyncFrame . fromDataView ( new DataView ( fileDataMessage . data ! . buffer . slice ( 0 , 8 ) ) ) ; let buffer = new Uint8Array ( fileDataMessage . data ! . buffer . slice ( 8 ) ) ; const chunks = [ ] ; while ( syncFrame . cmd !== 'DONE' ) { while ( syncFrame . byteLength >= buffer . byteLength ) { fileDataMessage = await this . read ( ) ; await this . client . sendMessage ( okayMessage ) ; const newLength = buffer . byteLength + fileDataMessage . data ! . byteLength ; const newBuffer = new Uint8Array ( newLength ) ; newBuffer . set ( buffer , 0 ) ; newBuffer . set ( new Uint8Array ( fileDataMessage . data ! . buffer ) , buffer . byteLength ) ; buffer = newBuffer ; } chunks . push ( buffer . slice ( 0 , syncFrame . byteLength ) . buffer ) ; buffer = buffer . slice ( syncFrame . byteLength ) ; syncFrame = SyncFrame . fromDataView ( new DataView ( buffer . slice ( 0 , 8 ) . buffer ) ) ; buffer = buffer . slice ( 8 ) ; } return new Blob ( chunks ) ; } private newMessage ( cmd , data ? ) { return Message . newMessage ( cmd , this . localId , this . remoteId , this . options . useChecksum , data ) ; } static async open ( adbClient , service , options ) { const localId = Stream . nextId ++ ; let remoteId = 0 ; const m = Message . open ( localId , remoteId , service , options . useChecksum ) ; await adbClient . sendMessage ( m ) ; let response ; do { response = await adbClient . awaitMessage ( ) ; } while ( response . header . arg1 !== localId ) ; if ( response . header . cmd !== 'OKAY' ) { throw new Error ( 'OPEN Failed' ) ; } remoteId = response . header . arg0 ; if ( options . debug ) { console . log ( `Opened stream ${ service }` ) ; console . log ( ` local_id: 0x${ toHex32 ( localId ) }` ) ; console . log ( ` remote_id: 0x${ toHex32 ( remoteId ) }` ) ; } const stream = new Stream ( adbClient , service , localId , remoteId , options ) ; adbClient . registerStream ( stream ) ; return stream ; } }