/**
 * This implementation is largely taken from the Apache project and reimplemented in TypeScript.
 *
 * The orginal project can be found here:
 * https://github.com/apache/thrift/blob/master/lib/nodejs/lib/thrift/buffered_transport.js
 */ import * as binary from '../binary' ; import { InputBufferUnderrunError } from '../errors' ; import { TTransport } from './TTransport' ; const DEFAULT_READ_BUFFER_SIDE = 1024 ; export class BufferedTransport extends TTransport { public static receiver ( data ) { const reader = new BufferedTransport ( Buffer . alloc ( data . length ) ) ; data . copy ( reader . buffer , 0 , 0 ) ; return reader ; } private readCursor : number ; private writeCursor : number ; private outBuffers : Array < Buffer > ; private outCount : number ; constructor ( buffer ? ) { super ( buffer || Buffer . alloc ( DEFAULT_READ_BUFFER_SIDE ) ) ; this . readCursor = 0 ; this . writeCursor = buffer !== undefined ? buffer . length : 0 ; this . outBuffers = [ ] ; this . outCount = 0 ; } public remaining ( ) { const remainingSize = this . writeCursor - this . readCursor ; const remainingBuffer = Buffer . alloc ( remainingSize ) ; if ( remainingSize > 0 ) { this . buffer . copy ( remainingBuffer , 0 , this . readCursor , this . writeCursor ) ; } return remainingBuffer ; } public commitPosition ( ) { const unreadSize = this . writeCursor - this . readCursor ; const bufSize = unreadSize * 2 > DEFAULT_READ_BUFFER_SIDE ? unreadSize * 2 : DEFAULT_READ_BUFFER_SIDE ; const buf = Buffer . alloc ( bufSize ) ; if ( unreadSize > 0 ) { this . buffer . copy ( buf , 0 , this . readCursor , this . writeCursor ) ; } this . readCursor = 0 ; this . writeCursor = unreadSize ; this . buffer = buf ; } public rollbackPosition ( ) { this . readCursor = 0 ; } public isOpen ( ) { return true ; } public open ( ) { return true ; } public close ( ) { return true ; } public read ( len ) { this . ensureAvailable ( len ) ; const buf = Buffer . alloc ( len ) ; this . buffer . copy ( buf , 0 , this . readCursor , this . readCursor + len ) ; this . readCursor += len ; return buf ; } public readByte ( ) { this . ensureAvailable ( 1 ) ; return binary . readByte ( this . buffer [ this . readCursor ++ ] ) ; } public readI16 ( ) { this . ensureAvailable ( 2 ) ; const i16 = binary . readI16 ( this . buffer , this . readCursor ) ; this . readCursor += 2 ; return i16 ; } public readI32 ( ) { this . ensureAvailable ( 4 ) ; const i32 = binary . readI32 ( this . buffer , this . readCursor ) ; this . readCursor += 4 ; return i32 ; } public readDouble ( ) { this . ensureAvailable ( 8 ) ; const d = binary . readDouble ( this . buffer , this . readCursor ) ; this . readCursor += 8 ; return d ; } public readString ( len ) { this . ensureAvailable ( len ) ; const str = this . buffer . toString ( 'utf8' , this . readCursor , this . readCursor + len ) ; this . readCursor += len ; return str ; } public readAll ( ) { return this . readString ( this . buffer . length - this . readCursor ) ; } public consume ( len ) { this . readCursor += len ; } public write ( buf ) { if ( buf instanceof Buffer ) { this . outBuffers . push ( buf ) ; this . outCount += buf . length ; } else { throw new TypeError ( `Expected buffer but found type ${ typeof buf }` ) ; } } public flush ( ) { if ( this . outCount < 1 ) { return Buffer . alloc ( 0 ) ; } const msg = Buffer . alloc ( this . outCount ) ; let pos = 0 ; this . outBuffers . forEach ( ( buf ) : void => { buf . copy ( msg , pos , 0 ) ; pos += buf . length ; } ) ; this . outBuffers = [ ] ; this . outCount = 0 ; return msg ; } private ensureAvailable ( len ) { if ( this . readCursor + len > this . buffer . length ) { throw new InputBufferUnderrunError ( ) ; } } }