222

schangxiang@126.com

2025-06-13 6a8393408d8cefcea02b7a598967de8dc1e565c2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
'use strict';
 
const debug = require('debug')('cluster-client');
const is = require('is-type-of');
const Base = require('sdk-base');
const assert = require('assert');
const utils = require('../utils');
// Symbols
const {
  init,
  logger,
  isReady,
  innerClient,
  subscribe,
  unSubscribe,
  publish,
  invoke,
  subInfo,
  pubInfo,
  closeHandler,
  close,
  singleMode,
  createClient,
} = require('../symbol');
 
class WrapperBase extends Base {
  /**
   * Share Connection among Multi-Process Mode
   *
   * @param {Object} options
   *  - {Number} port - the port
   *  - {Transcode} transcode - serialze / deseriaze methods
   *  - {Boolean} isLeader - wehether is leader or follower
   *  - {Number} maxWaitTime - leader startup max time (ONLY effective on isLeader is true)
   *  - {Function} createRealClient - to create the real client instance
   * @constructor
   */
  constructor(options) {
    super(options);
    this[subInfo] = new Map();
    this[pubInfo] = new Map();
    this[init]().catch(err => { this.ready(err); });
  }
 
  get isClusterClientLeader() {
    return this[innerClient] && this[innerClient].isLeader;
  }
 
  get [singleMode]() {
    return false;
  }
 
  /**
   * log instance
   * @property {Logger} ClusterClient#[logger]
   */
  get [logger]() {
    return this.options.logger;
  }
 
  async [createClient]() {
    throw new Error('not implement');
  }
 
  /**
   * initialize, to  leader or follower
   *
   * @return {void}
   */
  async [init]() {
    this[innerClient] = await this[createClient]();
 
    // events delegate
    utils.delegateEvents(this[innerClient], this);
 
    // re init when connection is close
    if (this[closeHandler]) {
      this[innerClient].on('close', this[closeHandler]);
    }
 
    // wait leader/follower ready
    await this[innerClient].ready();
 
    // subscribe all
    for (const registrations of this[subInfo].values()) {
      for (const args of registrations) {
        this[innerClient].subscribe(args[0], args[1]);
      }
    }
    // publish all
    for (const reg of this[pubInfo].values()) {
      this[innerClient].publish(reg);
    }
 
    if (!this[isReady]) {
      this[isReady] = true;
      this.ready(true);
    }
  }
 
  /**
   * do subscribe
   *
   * @param {Object} reg - subscription info
   * @param {Function} listener - callback function
   * @return {void}
   */
  [subscribe](reg, listener) {
    assert(is.function(listener), `[ClusterClient:${this.options.name}] subscribe(reg, listener) listener should be a function`);
 
    debug('[ClusterClient:%s] subscribe %j', this.options.name, reg);
    const key = this.options.formatKey(reg);
    const registrations = this[subInfo].get(key) || [];
    registrations.push([ reg, listener ]);
    this[subInfo].set(key, registrations);
 
    if (this[isReady]) {
      this[innerClient].subscribe(reg, listener);
    }
  }
 
  /**
   * do unSubscribe
   *
   * @param {Object} reg - subscription info
   * @param {Function} listener - callback function
   * @return {void}
   */
  [unSubscribe](reg, listener) {
    debug('[ClusterClient:%s] unSubscribe %j', this.options.name, reg);
    const key = this.options.formatKey(reg);
    const registrations = this[subInfo].get(key) || [];
    const newRegistrations = [];
    if (listener) {
      for (const arr of registrations) {
        if (arr[1] !== listener) {
          newRegistrations.push(arr);
        }
      }
    }
    this[subInfo].set(key, newRegistrations);
 
    if (this[isReady]) {
      this[innerClient].unSubscribe(reg, listener);
    }
  }
 
  /**
   * do publish
   *
   * @param {Object} reg - publish info
   * @return {void}
   */
  [publish](reg) {
    debug('[ClusterClient:%s] publish %j', this.options.name, reg);
    const key = this.options.formatKey(reg);
    this[pubInfo].set(key, reg);
 
    if (this[isReady]) {
      this[innerClient].publish(reg);
    }
  }
 
  /**
   * invoke a method asynchronously
   *
   * @param {String} method - the method name
   * @param {Array} args - the arguments list
   * @param {Function} callback - callback function
   * @return {void}
   */
  [invoke](method, args, callback) {
    if (!this[isReady]) {
      this.ready(err => {
        if (err) {
          callback && callback(err);
          return;
        }
        this[innerClient].invoke(method, args, callback);
      });
      return;
    }
 
    debug('[ClusterClient:%s] invoke method: %s, args: %j', this.options.name, method, args);
    this[innerClient].invoke(method, args, callback);
  }
 
  async [close]() {
    try {
      // close after ready, in case of innerClient is initializing
      await this.ready();
    } catch (err) {
      // ignore
    }
 
    const client = this[innerClient];
    if (client) {
      // prevent re-initializing
      if (this[closeHandler]) {
        client.removeListener('close', this[closeHandler]);
      }
      if (client.close) {
        await utils.callFn(client.close.bind(client));
      }
    }
  }
}
 
module.exports = WrapperBase;