How to use streamx - 5 common examples

const Union = require('./')
const { Readable } = require('streamx')

const sorted1 = Readable.from([{ key: 'a' }, { key: 'b' }, { key: 'c' }])
const sorted2 = Readable.from([{ key: 'b' }, { key: 'd' }])

const u = new Union(sorted1, sorted2)

u.on('data', function (data) {
  console.log(data)
})

u.on('end', function () {
  console.log('no more data')
})

const Union = require('./')
const { Readable } = require('streamx')

const sorted1 = Readable.from([{ key: 'a' }, { key: 'b' }, { key: 'c' }])
const sorted2 = Readable.from([{ key: 'b' }, { key: 'd' }])

const u = new Union(sorted1, sorted2)

u.on('data', function (data) {
  console.log(data)
})

u.on('end', function () {
  console.log('no more data')
})

* automatically.
     */
    MsgType[MsgType["End"] = 2] = "End";
    /**
     * Signal to the other side that we are completely closing this channel and will not read or
     * write any more data.
     *
     * Message body is an empty Buffer.
     *
     */
    MsgType[MsgType["Destroy"] = 3] = "Destroy";
})(MsgType = exports.MsgType || (exports.MsgType = {}));
/**
 * Allows many Duplex streams to be sent over a single Duplex.
 */
class Multiplex extends streamx_1.Duplex {
    constructor() {
        super();
        /**
         * Called when a remote frame is decoded by SMC.
         */
        this.onReceivedMsg = (remoteId, type, data) => {
            switch (type) {
                case MsgType.Start: {
                    const name = data.toString();
                    const channel = this.getOrCreateChannel(name);
                    this.remoteChannels.set(remoteId, channel);
                    break;
                }
                case MsgType.Data:
                    this.getChannelByRemoteId(remoteId).push(data);
                    break;

if (!channel)
            throw new Error(`Unknown remote channelId: ${id}`);
        return channel;
    }
    getOrCreateChannel(name) {
        return Misc_1.getOrCreate(this.channels, name, () => {
            const id = this.getNextId();
            return new Channel(name, id, (type, msg) => this.sendMsg(id, name, type, msg));
        });
    }
    getNextId() {
        return this.nextId++;
    }
}
exports.default = Multiplex;
class Channel extends streamx_1.Duplex {
    constructor(name, id, send) {
        super({
            highWaterMark: 0,
            mapWritable: (data) => (typeof data === 'string' ? Buffer.from(data) : data),
        });
        this.name = name;
        this.id = id;
        this.send = send;
        this.send(MsgType.Start, Buffer.from(this.name));
    }
    /**
     * Calls .end() and returns a promise that resolves when the channel is fully closed. Channels
     * are fully closed when both sides have called .end()
     */
    close() {
        return new Promise((res) => {

options[key] = opts[key];
    } else {
      if (opts.headers) {
        options.headers = options.headers || {};
        extend(options.headers, opts.headers);
      }
    }
  }
  var res = yield httpx.request(url, options);
  if (res.statusCode < 200 || res.statusCode > 204) {
    var err = new Error("url: " + url + ", status code: " + res.statusCode);
    err.name = "WeChatAPIError";
    throw err;
  }

  var buffer = yield streamx.read(res);
  var contentType = res.headers['content-type'] || '';
  if (contentType.indexOf('application/json') !== -1) {
    var data;
    try {
      data = JSON.parse(buffer);
    } catch (ex) {
      var err = new Error('JSON.parse error. buffer is ' + buffer.toString());
      err.name = "WeChatAPIError";
      throw err;
    }
    if (data && data.errcode) {
      var err = new Error(data.errmsg);
      err.name = 'WeChatAPIError';
      err.code = data.errcode;
      throw err;
    }