import binding from './binding' ; import { Connection } from './connection' ; import { HttpErrorContext } from './errorcontexts' ; import { RequestSpan } from './tracing' ; import * as events from 'events' ; internal export enum HttpServiceType { Management = 'MGMT' , Views = 'VIEW' , Query = 'QUERY' , Search = 'SEARCH' , Analytics = 'ANALYTICS' , Eventing = 'EVENTING' } internal export enum HttpMethod { Get = 'GET' , Post = 'POST' , Put = 'PUT' , Delete = 'DELETE' } internal export interface HttpRequestOptions { type : HttpServiceType ; method : HttpMethod ; path : string ; contentType ? : string ; body ? : string | Buffer ; parentSpan ? : RequestSpan ; timeout ? : number ; } internal export interface HttpResponse { requestOptions : HttpRequestOptions ; statusCode : number ; headers : { [ key ] : string ; } ; body : Buffer ; } internal export class HttpExecutor { private _conn : Connection ; internal constructor ( conn ) { this . _conn = conn ; } streamRequest ( options ) { const emitter = new events . EventEmitter ( ) ; let lcbHttpType ; if ( options . type === HttpServiceType . Management ) { lcbHttpType = binding . LCB_HTTP_TYPE_MANAGEMENT ; } else if ( options . type === HttpServiceType . Views ) { lcbHttpType = binding . LCB_HTTP_TYPE_VIEW ; } else if ( options . type === HttpServiceType . Query ) { lcbHttpType = binding . LCB_HTTP_TYPE_QUERY ; } else if ( options . type === HttpServiceType . Search ) { lcbHttpType = binding . LCB_HTTP_TYPE_SEARCH ; } else if ( options . type === HttpServiceType . Analytics ) { lcbHttpType = binding . LCB_HTTP_TYPE_ANALYTICS ; } else if ( options . type === HttpServiceType . Eventing ) { lcbHttpType = binding . LCB_HTTP_TYPE_EVENTING ; } else { throw new Error ( 'unexpected http request type' ) ; } let lcbHttpMethod ; if ( options . method === HttpMethod . Get ) { lcbHttpMethod = binding . LCB_HTTP_METHOD_GET ; } else if ( options . method === HttpMethod . Post ) { lcbHttpMethod = binding . LCB_HTTP_METHOD_POST ; } else if ( options . method === HttpMethod . Put ) { lcbHttpMethod = binding . LCB_HTTP_METHOD_PUT ; } else if ( options . method === HttpMethod . Delete ) { lcbHttpMethod = binding . LCB_HTTP_METHOD_DELETE ; } else { throw new Error ( 'unexpected http request method' ) ; } const lcbTimeout = options . timeout ? options . timeout * 1000 : undefined ; this . _conn . httpRequest ( lcbHttpType , lcbHttpMethod , options . path , options . contentType , options . body , options . parentSpan , lcbTimeout , ( err , flags , data ) => { if ( ! ( flags & binding . LCBX_RESP_F_NONFINAL ) ) { if ( err ) { emitter . emit ( 'error' , err ) ; return ; } emitter . emit ( 'end' , data ) ; return ; } if ( err ) { throw new Error ( 'unexpected error on non-final callback' ) ; } emitter . emit ( 'data' , data ) ; } ) ; return emitter ; } async request ( options ) { return new Promise ( ( resolve , reject ) => { const emitter = this . streamRequest ( options ) ; emitter . on ( 'error' , ( err ) => { reject ( err ) ; } ) ; let dataCache = Buffer . allocUnsafe ( 0 ) ; emitter . on ( 'data' , ( data ) => { dataCache = Buffer . concat ( [ dataCache , data ] ) ; } ) ; emitter . on ( 'end' , ( meta ) => { const headers = { } ; for ( let i = 0 ; i < meta . headers . length ; i += 2 ) { const headerName = meta . headers [ i + 0 ] ; const headerValue = meta . headers [ i + 1 ] ; if ( headers [ headerName ] ) { headers [ headerName ] += ',' + headerValue ; } else { headers [ headerName ] = headerValue ; } } resolve ( { requestOptions : options , statusCode : meta . statusCode , headers : headers , body : dataCache , } ) ; } ) ; } ) ; } static errorContextFromResponse ( resp ) { return new HttpErrorContext ( { method : resp . requestOptions . method , request_path : resp . requestOptions . path , response_code : resp . statusCode , response_body : resp . body . toString ( ) , } ) ; } }