'use strict'
|
|
const { Buffer } = require('buffer')
|
const Transform = require('readable-stream').Transform
|
const duplexify = require('duplexify')
|
|
/* global wx */
|
let socketTask, proxy, stream
|
|
function buildProxy () {
|
const proxy = new Transform()
|
proxy._write = function (chunk, encoding, next) {
|
socketTask.send({
|
data: chunk.buffer,
|
success: function () {
|
next()
|
},
|
fail: function (errMsg) {
|
next(new Error(errMsg))
|
}
|
})
|
}
|
proxy._flush = function socketEnd (done) {
|
socketTask.close({
|
success: function () {
|
done()
|
}
|
})
|
}
|
|
return proxy
|
}
|
|
function setDefaultOpts (opts) {
|
if (!opts.hostname) {
|
opts.hostname = 'localhost'
|
}
|
if (!opts.path) {
|
opts.path = '/'
|
}
|
|
if (!opts.wsOptions) {
|
opts.wsOptions = {}
|
}
|
}
|
|
function buildUrl (opts, client) {
|
const protocol = opts.protocol === 'wxs' ? 'wss' : 'ws'
|
let url = protocol + '://' + opts.hostname + opts.path
|
if (opts.port && opts.port !== 80 && opts.port !== 443) {
|
url = protocol + '://' + opts.hostname + ':' + opts.port + opts.path
|
}
|
if (typeof (opts.transformWsUrl) === 'function') {
|
url = opts.transformWsUrl(url, opts, client)
|
}
|
return url
|
}
|
|
function bindEventHandler () {
|
socketTask.onOpen(function () {
|
stream.setReadable(proxy)
|
stream.setWritable(proxy)
|
stream.emit('connect')
|
})
|
|
socketTask.onMessage(function (res) {
|
let data = res.data
|
|
if (data instanceof ArrayBuffer) data = Buffer.from(data)
|
else data = Buffer.from(data, 'utf8')
|
proxy.push(data)
|
})
|
|
socketTask.onClose(function () {
|
stream.end()
|
stream.destroy()
|
})
|
|
socketTask.onError(function (res) {
|
stream.destroy(new Error(res.errMsg))
|
})
|
}
|
|
function buildStream (client, opts) {
|
opts.hostname = opts.hostname || opts.host
|
|
if (!opts.hostname) {
|
throw new Error('Could not determine host. Specify host manually.')
|
}
|
|
const websocketSubProtocol =
|
(opts.protocolId === 'MQIsdp') && (opts.protocolVersion === 3)
|
? 'mqttv3.1'
|
: 'mqtt'
|
|
setDefaultOpts(opts)
|
|
const url = buildUrl(opts, client)
|
socketTask = wx.connectSocket({
|
url: url,
|
protocols: [websocketSubProtocol]
|
})
|
|
proxy = buildProxy()
|
stream = duplexify.obj()
|
stream._destroy = function (err, cb) {
|
socketTask.close({
|
success: function () {
|
cb && cb(err)
|
}
|
})
|
}
|
|
const destroyRef = stream.destroy
|
stream.destroy = function () {
|
stream.destroy = destroyRef
|
|
const self = this
|
setTimeout(function () {
|
socketTask.close({
|
fail: function () {
|
self._destroy(new Error())
|
}
|
})
|
}, 0)
|
}.bind(stream)
|
|
bindEventHandler()
|
|
return stream
|
}
|
|
module.exports = buildStream
|
