File Explorer

1'use strict'2 3const { Buffer } = require('buffer')4const Transform = require('readable-stream').Transform5const duplexify = require('duplexify')6 7/* global wx */8let socketTask, proxy, stream9 10function buildProxy () {11  const proxy = new Transform()12  proxy._write = function (chunk, encoding, next) {13    socketTask.send({14      data: chunk.buffer,15      success: function () {16        next()17      },18      fail: function (errMsg) {19        next(new Error(errMsg))20      }21    })22  }23  proxy._flush = function socketEnd (done) {24    socketTask.close({25      success: function () {26        done()27      }28    })29  }30 31  return proxy32}33 34function setDefaultOpts (opts) {35  if (!opts.hostname) {36    opts.hostname = 'localhost'37  }38  if (!opts.path) {39    opts.path = '/'40  }41 42  if (!opts.wsOptions) {43    opts.wsOptions = {}44  }45}46 47function buildUrl (opts, client) {48  const protocol = opts.protocol === 'wxs' ? 'wss' : 'ws'49  let url = protocol + '://' + opts.hostname + opts.path50  if (opts.port && opts.port !== 80 && opts.port !== 443) {51    url = protocol + '://' + opts.hostname + ':' + opts.port + opts.path52  }53  if (typeof (opts.transformWsUrl) === 'function') {54    url = opts.transformWsUrl(url, opts, client)55  }56  return url57}58 59function bindEventHandler () {60  socketTask.onOpen(function () {61    stream.setReadable(proxy)62    stream.setWritable(proxy)63    stream.emit('connect')64  })65 66  socketTask.onMessage(function (res) {67    let data = res.data68 69    if (data instanceof ArrayBuffer) data = Buffer.from(data)70    else data = Buffer.from(data, 'utf8')71    proxy.push(data)72  })73 74  socketTask.onClose(function () {75    stream.end()76    stream.destroy()77  })78 79  socketTask.onError(function (res) {80    stream.destroy(new Error(res.errMsg))81  })82}83 84function buildStream (client, opts) {85  opts.hostname = opts.hostname || opts.host86 87  if (!opts.hostname) {88    throw new Error('Could not determine host. Specify host manually.')89  }90 91  const websocketSubProtocol =92    (opts.protocolId === 'MQIsdp') && (opts.protocolVersion === 3)93      ? 'mqttv3.1'94      : 'mqtt'95 96  setDefaultOpts(opts)97 98  const url = buildUrl(opts, client)99  socketTask = wx.connectSocket({100    url: url,101    protocols: [websocketSubProtocol]102  })103 104  proxy = buildProxy()105  stream = duplexify.obj()106  stream._destroy = function (err, cb) {107    socketTask.close({108      success: function () {109        cb && cb(err)110      }111    })112  }113 114  const destroyRef = stream.destroy115  stream.destroy = function () {116    stream.destroy = destroyRef117 118    const self = this119    setTimeout(function () {120      socketTask.close({121        fail: function () {122          self._destroy(new Error())123        }124      })125    }, 0)126  }.bind(stream)127 128  bindEventHandler()129 130  return stream131}132 133module.exports = buildStream134