import { Readable } from 'stream' ; import { ReadableStream } from 'web-streams-polyfill/ponyfill' ; if ( typeof TextDecoder === 'undefined' && typeof require !== 'undefined' ) { ( global as any ) . TextDecoder = require ( 'util' ) . TextDecoder ; } export const readableNodeToWeb = < T > ( nodeStream ) => { if ( ! ( nodeStream instanceof Readable ) ) { return nodeStream ; } return new ReadableStream < T > ( { start ( controller ) { nodeStream . pause ( ) ; nodeStream . on ( 'data' , ( chunk ) => { controller . enqueue ( chunk ) ; nodeStream . pause ( ) ; } ) ; nodeStream . on ( 'end' , ( ) => controller . close ( ) ) ; nodeStream . on ( 'error' , ( e ) => controller . error ( e ) ) ; } , pull ( _controller ) { nodeStream . resume ( ) ; } , cancel ( _reason ) { nodeStream . pause ( ) ; } } ) ; } ; export class NodeReadable < T > extends Readable { _webStream : ReadableStream ; _reader : ReadableStreamDefaultReader ; _reading : boolean ; constructor ( webStream , options ? ) { super ( options ) ; this . _webStream = webStream ; this . _reader = webStream . getReader ( ) ; this . _reading = false ; } _read ( size ) { if ( this . _reading ) { return ; } this . _reading = true ; const doRead = ( size ) => { this . _reader . read ( ) . then ( ( res ) => { if ( res . done ) { this . push ( null ) ; return ; } if ( this . push ( res . value ) ) { return doRead ( size ) ; } else { this . _reading = false ; } } ) ; } ; doRead ( size ) ; } } export const readableWebToNode = < T > ( webStream ) => { return new NodeReadable ( webStream ) ; } ; export const streamHandler = < T > ( response ) => { let isReader ; let cancellationRequest = false ; return new ReadableStream < T > ( { start ( controller ) { const reader = readableNodeToWeb ( response ) . getReader ( ) ; isReader = reader ; const decoder = new TextDecoder ( ) ; let dataBuffer = '' ; const processResult = ( result ) => { if ( result . done ) { if ( cancellationRequest ) { return ; } dataBuffer = dataBuffer . trim ( ) ; if ( dataBuffer . length !== 0 ) { try { const dataLine = JSON . parse ( dataBuffer ) ; controller . enqueue ( dataLine ) ; } catch ( e ) { controller . error ( e ) ; return ; } } controller . close ( ) ; return ; } const data = decoder . decode ( result . value , { stream : true } ) ; dataBuffer += data ; const lines = dataBuffer . split ( '\n' ) ; for ( const line of lines ) { const l = line . trim ( ) ; if ( l . length > 0 ) { try { controller . enqueue ( JSON . parse ( l ) ) ; } catch ( e ) { controller . error ( e ) ; cancellationRequest = true ; reader . cancel ( undefined ) ; return ; } } } dataBuffer = lines . length > 1 ? lines [ lines . length - 1 ] : '' ; reader . read ( ) . then ( processResult ) ; } ; reader . read ( ) . then ( processResult ) ; } , cancel ( reason ? ) { cancellationRequest = true ; isReader . cancel ( reason ) ; } } ) ; } ;