v0.3.0 released

.eslintrc.yaml

@@ -0,0 +1,22 @@
+parser: "@babel/eslint-parser"
+parserOptions:
+  ecmaVersion: 2020
+  sourceType: module
+  requireConfigFile: false
+env:
+  es2020: true
+  node: true
+extends:
+  - airbnb-base
+rules:
+  max-len: [2, 180]
+  valid-jsdoc: 1
+overrides:
+    - env:
+        mocha: true
+        node: true
+      plugins:
+        - mocha
+      files:
+        - "tests/*.test.js"
+        - "tests/**/*.test.js"

LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2022 James ZHANG(TheNorthMemory)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,36 @@
+# Taobao Message Channel (tmc.js)
+
+Events driven and chained Taobao Message Channel(TMC) for NodeJS
+
+## 使用
+
+`npm i tmc.js`
+
+```js
+import TMC from 'tmc.js';
+
+new TMC('your_app_key', 'your_app_secret')
+.on('taobao_trade_TradeChanged', msg => console.info(msg))
+.taobao_trade_TradeClose(msg => console.info(msg))
+.connect();
+```
+
+## API
+
+**`new TMC(appKey: string, appSecret: BinaryLike, groupName?: string | object, options?: object)`**
+
+| 参数 | 类型 | 说明 |
+| --- | --- | --- |
+| appKey | `string` | 应用ID |
+| appSecret | `BinaryLike` | 应用密钥 |
+| groupName | `string` | 消息分组(可选，默认`default`) |
+| options | `object` | 消费端配置参数(可选) |
+| options.pullRequestInterval | `number` | 定时发送`pullRequest`请求时间间隔(默认`5000`毫秒) |
+| options.onErrorReconnection | `number` | 当消费端出现错误，重试连接间隔(默认`15000`毫秒) |
+| options.onCloseReconnection | `number` | 当消费端断开连接，重试连接间隔(默认`3000`毫秒) |
+| options.autoParseContentJson | `boolean` | 自动解析推送消息`$.content.content`字段为对象(默认`true`) |
+| options.autoReplyConfirmation | `boolean` | 以推送的`$.content.id`字段自动`Confirm`消息(默认`true`) |
+
+## License
+
+[MIT](LICENSE)

index.js

@@ -0,0 +1,16 @@
+const TaoMessageConsumer = require('./lib/consumer');
+const HeaderType = require('./lib/header-type');
+const MessageFields = require('./lib/message-fields');
+const MessageKind = require('./lib/message-kind');
+const MessageType = require('./lib/message-type');
+const Message = require('./lib/message');
+const ValueFormat = require('./lib/value-format');
+
+module.exports = TaoMessageConsumer;
+module.exports.default = TaoMessageConsumer;
+module.exports.HeaderType = HeaderType;
+module.exports.MessageFields = MessageFields;
+module.exports.MessageKind = MessageKind;
+module.exports.MessageType = MessageType;
+module.exports.Message = Message;
+module.exports.ValueFormat = ValueFormat;

index.mjs

@@ -0,0 +1,10 @@
+import TaoMessageConsumer from './lib/consumer.js';
+import HeaderType from './lib/header-type.js';
+import MessageFields from './lib/message-fields.js';
+import MessageKind from './lib/message-kind.js';
+import MessageType from './lib/message-type.js';
+import Message from './lib/message.js';
+import ValueFormat from './lib/value-format.js';
+
+export { HeaderType, MessageFields, MessageKind, MessageType, Message, ValueFormat };
+export default TaoMessageConsumer;

lib/consumer.js

@@ -0,0 +1,206 @@
+const { createHash, createSecretKey } = require('crypto');
+const { networkInterfaces } = require('os');
+const { EventEmitter, captureRejectionSymbol } = require('events');
+
+const WebSocket = require('ws');
+const JSONB = require('json-bigint')({ useNativeBigInt: true });
+
+const logger = require('./logger');
+const Message = require('./message');
+const { PullRequest, Confirm } = require('./message-kind');
+const {
+  CONNECT, SEND, SENDACK, CONNECTACK,
+} = require('./message-type');
+const {
+  APP_KEY, GROUP_NAME, INTRANET_IP, TIMESTAMP, SIGN, SDK, TYPE, CODE, TOKEN, CONTENT, ID, TOPIC,
+} = require('./message-fields');
+
+const kAddress = Symbol('kAddress');
+const kAppKey = Symbol('kAppKey');
+const kAppSecret = Symbol('kAppSecret');
+const kGroupName = Symbol('kGroupName');
+const kToken = Symbol('kToken');
+const kOptions = Symbol('kOptions');
+const kWebSocket = Symbol('kWebSocket');
+
+function ipAddr() {
+  return Object.values(networkInterfaces()).flat().find(
+    ({ internal, family, address }) => !internal && (family === 'IPv4' || family === 4) && address !== '127.0.0.1',
+  )?.address;
+}
+
+class TaoMessageComsumer extends EventEmitter {
+  [kAddress] = 'ws://mc.api.taobao.com/';
+
+  [kGroupName] = 'default';
+
+  [kAppKey];
+
+  [kAppSecret];
+
+  [kToken];
+
+  [kOptions] = {
+    pullRequestInterval: 5e3,
+    onErrorReconnection: 15e3,
+    onCloseReconnection: 3e3,
+    autoParseContentJson: true,
+    autoReplyConfirmation: true,
+  };
+
+  [kWebSocket];
+
+  constructor(appKey, appSecret, groupName, options = {}) {
+    super({ captureRejections: true });
+
+    this[kAppKey] = appKey;
+    this[kAppSecret] = createSecretKey(appSecret);
+    if (groupName && typeof groupName === 'string') { this[kGroupName] = groupName; }
+    this[kOptions] = {
+      ...this[kOptions],
+      ...(typeof groupName === 'object' ? groupName : options),
+    };
+
+    return /* eslint-disable-line no-constructor-return */ new Proxy(this, {
+      get(target, topic) {
+        if (typeof topic === 'symbol') { return target[topic]; }
+        if (!Object.prototype.hasOwnProperty.call(target, topic) && !(topic in target)) {
+          Object.defineProperty(target, topic, {
+            value: function subscriber(callback) {
+              if (typeof callback !== 'function') {
+                throw new TypeError('The callback must be a function here.');
+              }
+              return this.on(topic, callback);
+            },
+          });
+        }
+
+        return target[topic];
+      },
+    });
+  }
+
+  sign(timestamp) {
+    const sk = this[kAppSecret].export();
+    return [
+      ['app_key', this[kAppKey]],
+      ['group_name', this[kGroupName]],
+      ['timestamp', `${timestamp}`],
+    ].reduce(
+      (h, [k, v]) => h.update(k).update(v),
+      createHash('md5').update(sk),
+    ).update(sk).digest('hex').toUpperCase();
+  }
+
+  onopen() {
+    const now = `${Date.now()}`;
+    const msg = new Message(CONNECT)
+      .with(APP_KEY, this[kAppKey])
+      .with(GROUP_NAME, this[kGroupName])
+      .with(TIMESTAMP, now)
+      .with(SIGN, this.sign(now))
+      .with(SDK, this.constructor.name)
+      .with(INTRANET_IP, ipAddr());
+
+    logger.onopen(JSONB.stringify(msg));
+
+    this[kWebSocket]?.send(msg.buffer, { mask: true, binary: true });
+
+    if (this[kOptions].intervalId) { clearInterval(this[kOptions].intervalId); }
+
+    this[kOptions].intervalId = setInterval(() => this.onpull(), this[kOptions].pullRequestInterval);
+  }
+
+  onpull() {
+    const msg = new Message(SEND, PullRequest).with(TOKEN, this[kToken]);
+    logger.onpull(JSONB.stringify(msg));
+    this[kWebSocket]?.send(msg.buffer, { mask: true, binary: true });
+  }
+
+  onping(data) {
+    logger.onping('The channel is onping with data [%s].', JSONB.stringify(data));
+    this[kWebSocket]?.pong(data, { mask: true }, true);
+  }
+
+  onerror(err) {
+    logger.onerror('The channel is onerror(%s), let us reconnect.', JSONB.stringify(err));
+    this.reconnect(this[kOptions].onErrorReconnection);
+  }
+
+  onclose(...arg) {
+    logger.onclose('The channel is onclose(%s), let us reconnect.', JSONB.stringify(arg));
+    this.reconnect(this[kOptions].onCloseReconnection);
+  }
+
+  [`process${CONNECTACK}`](msg) {
+    if (!msg[CODE] && msg[TOKEN]) { this[kToken] = msg[TOKEN]; }
+  }
+
+  [`process${SEND}`](msg) {
+    if (msg && msg[CONTENT] && msg[CONTENT][TOPIC]) { this.emit(msg[CONTENT][TOPIC], msg); }
+
+    if (this[kOptions].autoReplyConfirmation && msg && msg[CONTENT] && msg[CONTENT][ID]) {
+      const reply = new Message(SEND, Confirm).with(TOKEN, this[kToken]).with(ID, msg[CONTENT][ID]);
+      logger.onmessage[SEND][Confirm](JSONB.stringify(reply));
+      this[kWebSocket]?.send(reply.buffer, { mask: true, binary: true });
+    }
+  }
+
+  [`process${CONNECT}`]() { /* eslint-disable-line class-methods-use-this */ }
+
+  [`process${SENDACK}`]() { /* eslint-disable-line class-methods-use-this */ }
+
+  processundefined() { /* eslint-disable-line class-methods-use-this */ }
+
+  [captureRejectionSymbol](err) { this.onerror(err); }
+
+  onmessage(data, isBinary = true) {
+    if (!isBinary) { return; }
+
+    const msg = Message.from(data);
+
+    if (
+      this[kOptions].autoParseContentJson
+      && msg[CONTENT][CONTENT]
+      && typeof msg[CONTENT][CONTENT] === 'string'
+      && /^\{.*\}$/.test(msg[CONTENT][CONTENT])) {
+      Reflect.set(msg[CONTENT], CONTENT, JSONB.parse(msg[CONTENT][CONTENT]));
+    }
+
+    logger.onmessage[msg[TYPE]](JSONB.stringify(msg));
+
+    this[`process${msg[TYPE]}`](msg);
+  }
+
+  reconnect(ms) {
+    this[kWebSocket]?.terminate();
+
+    if (this[kOptions].intervalId) { this[kOptions].intervalId = clearInterval(this[kOptions].intervalId); }
+    if (this[kOptions].timeoutId) { this[kOptions].timeoutId = clearTimeout(this[kOptions].timeoutId); }
+
+    this[kWebSocket] = undefined;
+    this[kOptions].timeoutId = setTimeout(() => this.connect(), ms);
+
+    return this;
+  }
+
+  connect(address) {
+    if (address) { this[kAddress] = address; }
+
+    this[kWebSocket]?.close();
+
+    const ws = new WebSocket(this[kAddress]);
+    Reflect.set(this, kWebSocket, ws);
+
+    ws.on('open', () => this.onopen());
+    ws.on('ping', (...arg) => this.onping(...arg));
+    ws.on('error', (err) => this.onerror(err));
+    ws.on('close', (...arg) => this.onclose(...arg));
+    ws.on('message', (...arg) => this.onmessage(...arg));
+
+    return this;
+  }
+}
+
+module.exports = TaoMessageComsumer;
+module.exports.default = TaoMessageComsumer;

lib/header-type.js

@@ -0,0 +1,29 @@
+const EndOfHeaders = 0;
+const Custom = 1;
+const StatusCode = 2;
+const StatusPhrase = 3;
+const Flag = 4;
+const Token = 5;
+
+class HeaderType {
+  static EndOfHeaders = EndOfHeaders;
+
+  static Custom = Custom;
+
+  static StatusCode = StatusCode;
+
+  static StatusPhrase = StatusPhrase;
+
+  static Flag = Flag;
+
+  static Token = Token;
+}
+
+module.exports = HeaderType;
+module.exports.default = HeaderType;
+module.exports.EndOfHeaders = EndOfHeaders;
+module.exports.Custom = Custom;
+module.exports.StatusCode = StatusCode;
+module.exports.StatusPhrase = StatusPhrase;
+module.exports.Flag = Flag;
+module.exports.Token = Token;

lib/logger.js

@@ -0,0 +1,18 @@
+const { debuglog } = require('util');
+
+const logger = debuglog('tmc');
+
+logger.onping = debuglog('tmc:onping');
+logger.onopen = debuglog('tmc:onopen');
+logger.onpull = debuglog('tmc:onpull');
+logger.onerror = debuglog('tmc:onerror');
+logger.onclose = debuglog('tmc:onclose');
+logger.onmessage = debuglog('tmc:onmessage');
+logger.onmessage[0] = debuglog('tmc:onmessage:connect');
+logger.onmessage[1] = debuglog('tmc:onmessage:connectack');
+logger.onmessage[2] = debuglog('tmc:onmessage:send');
+logger.onmessage[3] = debuglog('tmc:onmessage:sendack');
+logger.onmessage[2][2] = debuglog('tmc:onmessage:send:confirm');
+
+module.exports = logger;
+module.exports.default = logger;