faas/ix-server/connector.js

var Stream = require('stream')

var io = require('socket.io-client')
var SocketIoStream = require('socket.io-stream')
var Request = require('request')

var Base = require('../base')
  , Crypto = require('./crypto')
  , Logger = require('./logger')
  , PolyglotRelay = require('./polyglot_relay')
  , Publisher = require('./publisher')

var EVENTS = [ 'connect', 'connect_error', 'connect_timeout', 'reconnect', 'reconnect_error', 'reconnect_failed', 'error', 'disconnect', 'open', 'close', 'reconnect_attempt' ]
var PROXY_HTTP_HOST_USE_PROXY = 'USE_PROXY'

var debug = Base.logger('ix-connector')

var socket, proxyData, upstreamUrl, isUpstreamHttps
var isRelay = false
var isHttpProxy = false
var deviceData = {}

exports.connectToProxy = function(data, cb_err_data){
  debug('connectToProxy', data)
  Logger.loadFilter(function(){
    Publisher.withDeviceData(function(err, devData){
      deviceData = devData
      proxyData = data
      if (Base.env.IX_PROXY_HOST_OVERRIDE) { proxyData.proxyUrl = "https://"+Base.env.IX_PROXY_HOST_OVERRIDE }
      console.log('Connecting to '+proxyData.proxyUrl+'...')
      socket = io(proxyData.proxyUrl)
      setDebugHandlers(socket, EVENTS)
      setDebugHandlers(socket.io, EVENTS, 'manager:')
      socket.on('connect', function(){ onConnect(cb_err_data) })
      if (isHttpProxy) {
        var streamSocket = SocketIoStream(socket)
        setDebugHandlers(streamSocket, EVENTS, 'stream:')
        streamSocket.on('http-proxy', onHttpProxy)
      }
      else {
        socket.on('ix-call', onIxCall)
        socket.on('ix-ping', onIxPing)
        socket.on('ix-x', onIxX)
      }
    })
  })
}

exports.connectRelayToProxy = function(data, cb_err_data){
  debug('connectRelayToProxy', data)
  isRelay = true
  exports.connectToProxy(data, cb_err_data)
}

exports.connectHttpToProxy = function(upstream, data, cb_err_data){
  upstreamUrl = upstream || Base.env.IX_HTTP_PROXY
  if (upstreamUrl) {
    upstreamUrl = upstreamUrl.trim().replace(/\/$/, '')
  }
  if (!upstreamUrl) { cb_err_data("No upstream server defined."); return }
  debug('connectHttpToProxy', upstreamUrl, data)
  isHttpProxy = true
  isUpstreamHttps = upstreamUrl.slice(0,6) === 'https:'
  exports.connectToProxy(data, cb_err_data)
}

var setDebugHandlers = function(socket, events, prefix){
  prefix = prefix || ''
  var i, event
  for (i = 0; i < events.length; i++) {
    event = events[i]
    socket.on(event, debugHandlerFor(prefix+event))
  }
}

var debugHandlerFor = function(type){
  return function(){
    var args = Array.prototype.slice.call(arguments, 0)
    args.unshift('event:'+type)
    debug.apply(this, args)
  }
}

var onConnect = function(cb_err_data){
  debug('connect', arguments)
  socket.emit('set-client', { apiConfig: Base.apiConfig(), proxyUrl: proxyData.proxyUrl, proxyExport: proxyData.proxyExport, device: deviceData }, function(err, data){
    if (err) {
      console.log('Connect error: Unable to complete connection to '+proxyData.proxyUrl+': '+JSON.stringify(err))
    }
    else {
      console.log('Connected to '+proxyData.proxyUrl+'.')
    }
    // TODO: if (err) { retryConnection }
    debug('connect result', err, data)
    cb_err_data(err, data)
  })
}

var onIxCall = function(data, responder){
  debug('ix-call', arguments)
  try {
    Logger.logCallRequest(data)
    responder = Logger.wrapResponder(data, debug.wrap(responder))
  }
  catch(e){}
  try {
    if (isRelay) {
      Publisher.ixCall(data, responder, PolyglotRelay.invoker)
    }
    else {
      Publisher.ixCallNodejs(data, responder)
    }
  }
  catch(e) {
    responder(e)
  }
}

var onIxPing = function(replyData, responder){
  debug('ix-ping', replyData)
  responder(null, replyData)
}

var onIxX = function(cipherParams, responder){
  debug('ix-x', cipherParams)
  Crypto.decrypt(cipherParams, function(err, data){
    if (err) { responder.call(err); return }
    onIxCall(data, Crypto.encryptingResponder(responder))
  })
}

var onHttpProxy = function(faasReq, sioStream, cb_proxyResponse){
  Logger.logHttpRequest(faasReq)
  sioStream.allowHalfOpen = true // Let reading end before writing closes.
  cb_proxyResponse = debug.wrap(cb_proxyResponse)
  var url = upstreamUrl + faasReq.url
  var fwdOptions = {
    url: url
    , headers: faasReq.headers
    , method: faasReq.method
    , followRedirect: false
  }
  var httpHost = Base.env.IX_PROXY_HTTP_HOST
  if (isUpstreamHttps || httpHost === PROXY_HTTP_HOST_USE_PROXY) {
    delete fwdOptions.headers.host
  }
  if (httpHost && httpHost !== PROXY_HTTP_HOST_USE_PROXY) {
    fwdOptions.headers.host = httpHost
  }
  debug('http-proxy request', fwdOptions)
  var fwdReq = Request(fwdOptions)
  var proxyResponse = {}
  var responseStarted = false
  faasReq.responseBytes = 0
  var counter = new Stream.Transform()
  counter._transform = function(chunk, encoding, callback){
    debug('http-proxy response chunk', chunk)
    faasReq.responseBytes += chunk.length
    callback(null, chunk)
  }
  fwdReq.on('error', function(err){
    debug('fwdReq.error', err, proxyResponse)
    if (responseStarted) {
      console.error('duplicate ERROR response attempted for '+url)
      sioStream.emit('error', err)
    }
    else {
      responseStarted = true
      proxyResponse.error = err.toString()
      cb_proxyResponse(proxyResponse)
    }
    try {
      fwdReq.abort()
      fwdReq.destroy()
      sioStream.unpipe()
      sioStream.on('data', function(data){ debug('fwdReq.error data', data) })
      sioStream.end()
    }
    catch (e) { console.error(e) }
    Logger.logHttpResponseError(faasReq, proxyResponse)
  })
  fwdReq.on('response', function(response){
    Logger.logHttpResponseStart(faasReq, response)
    debug('fwdReq.response', response.headers, response.statusCode, response.statusMessage, proxyResponse)
    if (responseStarted) {
      console.error('duplicate SUCCESS response attempted for '+url)
      return
    }
    responseStarted = true
    proxyResponse.lag = faasReq.ts.responseStart - faasReq.ts.requestStart
    proxyResponse.headers = response.headers
    proxyResponse.statusCode = response.statusCode
    proxyResponse.statusMessage = response.statusMessage
    cb_proxyResponse(proxyResponse)
    fwdReq.pipe(counter).pipe(sioStream)
  })
  fwdReq.on('end', function(){
    Logger.logHttpResponseEnd(faasReq)
  })
  sioStream.pipe(fwdReq)
}