'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