1
// Copyright 2007 The Closure Library Authors. All Rights Reserved.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//      http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS-IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14

15
/**
16
 * @fileoverview  Topic-based publish/subscribe channel implementation.
17
 *
18
 * @author [email protected] (Attila Bodis)
19
 */
20

21
goog.provide('goog.pubsub.PubSub');
22

23
goog.require('goog.Disposable');
24
goog.require('goog.array');
25
goog.require('goog.async.run');
26

27

28

29
/**
30
 * Topic-based publish/subscribe channel.  Maintains a map of topics to
31
 * subscriptions.  When a message is published to a topic, all functions
32
 * subscribed to that topic are invoked in the order they were added.
33
 * Uncaught errors abort publishing.
34
 *
35
 * Topics may be identified by any nonempty string, <strong>except</strong>
36
 * strings corresponding to native Object properties, e.g. "constructor",
37
 * "toString", "hasOwnProperty", etc.
38
 *
39
 * @constructor
40
 * @param {boolean=} opt_async Enable asynchronous behavior.  Recommended for
41
 *     new code.  See notes on the publish() method.
42
 * @extends {goog.Disposable}
43
 */
44
goog.pubsub.PubSub = function(opt_async) {
45
  goog.pubsub.PubSub.base(this, 'constructor');
46

47
  /**
48
   * The next available subscription key.  Internally, this is an index into the
49
   * sparse array of subscriptions.
50
   *
51
   * @private {number}
52
   */
53
  this.key_ = 1;
54

55
  /**
56
   * Array of subscription keys pending removal once publishing is done.
57
   *
58
   * @private {!Array<number>}
59
   * @const
60
   */
61
  this.pendingKeys_ = [];
62

63
  /**
64
   * Lock to prevent the removal of subscriptions during publishing. Incremented
65
   * at the beginning of {@link #publish}, and decremented at the end.
66
   *
67
   * @private {number}
68
   */
69
  this.publishDepth_ = 0;
70

71
  /**
72
   * Sparse array of subscriptions. Each subscription is represented by a tuple
73
   * comprising a topic identifier, a function, and an optional context object.
74
   * Each tuple occupies three consecutive positions in the array, with the
75
   * topic identifier at index n, the function at index (n + 1), the context
76
   * object at index (n + 2), the next topic at index (n + 3), etc. (This
77
   * representation minimizes the number of object allocations and has been
78
   * shown to be faster than an array of objects with three key-value pairs or
79
   * three parallel arrays, especially on IE.) Once a subscription is removed
80
   * via {@link #unsubscribe} or {@link #unsubscribeByKey}, the three
81
   * corresponding array elements are deleted, and never reused. This means the
82
   * total number of subscriptions during the lifetime of the pubsub channel is
83
   * limited by the maximum length of a JavaScript array to (2^32 - 1) / 3 =
84
   * 1,431,655,765 subscriptions, which should suffice for most applications.
85
   *
86
   * @private {!Array<?>}
87
   * @const
88
   */
89
  this.subscriptions_ = [];
90

91
  /**
92
   * Map of topics to arrays of subscription keys.
93
   *
94
   * @private {!Object<!Array<number>>}
95
   */
96
  this.topics_ = {};
97

98
  /**
99
   * @private @const {boolean}
100
   */
101
  this.async_ = Boolean(opt_async);
102
};
103
goog.inherits(goog.pubsub.PubSub, goog.Disposable);
104

105

106
/**
107
 * Subscribes a function to a topic.  The function is invoked as a method on
108
 * the given {@code opt_context} object, or in the global scope if no context
109
 * is specified.  Subscribing the same function to the same topic multiple
110
 * times will result in multiple function invocations while publishing.
111
 * Returns a subscription key that can be used to unsubscribe the function from
112
 * the topic via {@link #unsubscribeByKey}.
113
 *
114
 * @param {string} topic Topic to subscribe to.
115
 * @param {Function} fn Function to be invoked when a message is published to
116
 *     the given topic.
117
 * @param {Object=} opt_context Object in whose context the function is to be
118
 *     called (the global scope if none).
119
 * @return {number} Subscription key.
120
 */
121
goog.pubsub.PubSub.prototype.subscribe = function(topic, fn, opt_context) {
122
  var keys = this.topics_[topic];
123
  if (!keys) {
124
    // First subscription to this topic; initialize subscription key array.
125
    keys = this.topics_[topic] = [];
126
  }
127

128
  // Push the tuple representing the subscription onto the subscription array.
129
  var key = this.key_;
130
  this.subscriptions_[key] = topic;
131
  this.subscriptions_[key + 1] = fn;
132
  this.subscriptions_[key + 2] = opt_context;
133
  this.key_ = key + 3;
134

135
  // Push the subscription key onto the list of subscriptions for the topic.
136
  keys.push(key);
137

138
  // Return the subscription key.
139
  return key;
140
};
141

142

143
/**
144
 * Subscribes a single-use function to a topic.  The function is invoked as a
145
 * method on the given {@code opt_context} object, or in the global scope if
146
 * no context is specified, and is then unsubscribed.  Returns a subscription
147
 * key that can be used to unsubscribe the function from the topic via
148
 * {@link #unsubscribeByKey}.
149
 *
150
 * @param {string} topic Topic to subscribe to.
151
 * @param {Function} fn Function to be invoked once and then unsubscribed when
152
 *     a message is published to the given topic.
153
 * @param {Object=} opt_context Object in whose context the function is to be
154
 *     called (the global scope if none).
155
 * @return {number} Subscription key.
156
 */
157
goog.pubsub.PubSub.prototype.subscribeOnce = function(topic, fn, opt_context) {
158
  // Keep track of whether the function was called.  This is necessary because
159
  // in async mode, multiple calls could be scheduled before the function has
160
  // the opportunity to unsubscribe itself.
161
  var called = false;
162

163
  // Behold the power of lexical closures!
164
  var key = this.subscribe(topic, function(var_args) {
165
    if (!called) {
166
      called = true;
167

168
      // Unsubuscribe before calling function so the function is unscubscribed
169
      // even if it throws an exception.
170
      this.unsubscribeByKey(key);
171

172
      fn.apply(opt_context, arguments);
173
    }
174
  }, this);
175
  return key;
176
};
177

178

179
/**
180
 * Unsubscribes a function from a topic.  Only deletes the first match found.
181
 * Returns a Boolean indicating whether a subscription was removed.
182
 *
183
 * @param {string} topic Topic to unsubscribe from.
184
 * @param {Function} fn Function to unsubscribe.
185
 * @param {Object=} opt_context Object in whose context the function was to be
186
 *     called (the global scope if none).
187
 * @return {boolean} Whether a matching subscription was removed.
188
 */
189
goog.pubsub.PubSub.prototype.unsubscribe = function(topic, fn, opt_context) {
190
  var keys = this.topics_[topic];
191
  if (keys) {
192
    // Find the subscription key for the given combination of topic, function,
193
    // and context object.
194
    var subscriptions = this.subscriptions_;
195
    var key = goog.array.find(keys, function(k) {
196
      return subscriptions[k + 1] == fn && subscriptions[k + 2] == opt_context;
197
    });
198
    // Zero is not a valid key.
199
    if (key) {
200
      return this.unsubscribeByKey(key);
201
    }
202
  }
203

204
  return false;
205
};
206

207

208
/**
209
 * Removes a subscription based on the key returned by {@link #subscribe}.
210
 * No-op if no matching subscription is found.  Returns a Boolean indicating
211
 * whether a subscription was removed.
212
 *
213
 * @param {number} key Subscription key.
214
 * @return {boolean} Whether a matching subscription was removed.
215
 */
216
goog.pubsub.PubSub.prototype.unsubscribeByKey = function(key) {
217
  var topic = this.subscriptions_[key];
218
  if (topic) {
219
    // Subscription tuple found.
220
    var keys = this.topics_[topic];
221

222
    if (this.publishDepth_ != 0) {
223
      // Defer removal until after publishing is complete, but replace the
224
      // function with a no-op so it isn't called.
225
      this.pendingKeys_.push(key);
226
      this.subscriptions_[key + 1] = goog.nullFunction;
227
    } else {
228
      if (keys) {
229
        goog.array.remove(keys, key);
230
      }
231
      delete this.subscriptions_[key];
232
      delete this.subscriptions_[key + 1];
233
      delete this.subscriptions_[key + 2];
234
    }
235
  }
236

237
  return !!topic;
238
};
239

240

241
/**
242
 * Publishes a message to a topic.  Calls functions subscribed to the topic in
243
 * the order in which they were added, passing all arguments along.
244
 *
245
 * If this object was created with async=true, subscribed functions are called
246
 * via goog.async.run().  Otherwise, the functions are called directly, and if
247
 * any of them throw an uncaught error, publishing is aborted.
248
 *
249
 * @param {string} topic Topic to publish to.
250
 * @param {...*} var_args Arguments that are applied to each subscription
251
 *     function.
252
 * @return {boolean} Whether any subscriptions were called.
253
 */
254
goog.pubsub.PubSub.prototype.publish = function(topic, var_args) {
255
  var keys = this.topics_[topic];
256
  if (keys) {
257
    // Copy var_args to a new array so they can be passed to subscribers.
258
    // Note that we can't use Array.slice or goog.array.toArray for this for
259
    // performance reasons. Using those with the arguments object will cause
260
    // deoptimization.
261
    var args = new Array(arguments.length - 1);
262
    for (var i = 1, len = arguments.length; i < len; i++) {
263
      args[i - 1] = arguments[i];
264
    }
265

266
    if (this.async_) {
267
      // For each key in the list of subscription keys for the topic, schedule
268
      // the function to be applied to the arguments in the appropriate context.
269
      for (i = 0; i < keys.length; i++) {
270
        var key = keys[i];
271
        goog.pubsub.PubSub.runAsync_(
272
            this.subscriptions_[key + 1], this.subscriptions_[key + 2], args);
273
      }
274
    } else {
275
      // We must lock subscriptions and remove them at the end, so we don't
276
      // adversely affect the performance of the common case by cloning the key
277
      // array.
278
      this.publishDepth_++;
279

280
      try {
281
        // For each key in the list of subscription keys for the topic, apply
282
        // the function to the arguments in the appropriate context.  The length
283
        // of the array must be fixed during the iteration, since subscribers
284
        // may add new subscribers during publishing.
285
        for (i = 0, len = keys.length; i < len; i++) {
286
          var key = keys[i];
287
          this.subscriptions_[key + 1].apply(
288
              this.subscriptions_[key + 2], args);
289
        }
290
      } finally {
291
        // Always unlock subscriptions, even if a subscribed method throws an
292
        // uncaught exception. This makes it possible for users to catch
293
        // exceptions themselves and unsubscribe remaining subscriptions.
294
        this.publishDepth_--;
295

296
        if (this.pendingKeys_.length > 0 && this.publishDepth_ == 0) {
297
          var pendingKey;
298
          while ((pendingKey = this.pendingKeys_.pop())) {
299
            this.unsubscribeByKey(pendingKey);
300
          }
301
        }
302
      }
303
    }
304

305
    // At least one subscriber was called.
306
    return i != 0;
307
  }
308

309
  // No subscribers were found.
310
  return false;
311
};
312

313

314
/**
315
 * Runs a function asynchronously with the given context and arguments.
316
 * @param {!Function} func The function to call.
317
 * @param {*} context The context in which to call {@code func}.
318
 * @param {!Array} args The arguments to pass to {@code func}.
319
 * @private
320
 */
321
goog.pubsub.PubSub.runAsync_ = function(func, context, args) {
322
  goog.async.run(function() { func.apply(context, args); });
323
};
324

325

326
/**
327
 * Clears the subscription list for a topic, or all topics if unspecified.
328
 * @param {string=} opt_topic Topic to clear (all topics if unspecified).
329
 */
330
goog.pubsub.PubSub.prototype.clear = function(opt_topic) {
331
  if (opt_topic) {
332
    var keys = this.topics_[opt_topic];
333
    if (keys) {
334
      goog.array.forEach(keys, this.unsubscribeByKey, this);
335
      delete this.topics_[opt_topic];
336
    }
337
  } else {
338
    this.subscriptions_.length = 0;
339
    this.topics_ = {};
340
    // We don't reset key_ on purpose, because we want subscription keys to be
341
    // unique throughout the lifetime of the application.  Reusing subscription
342
    // keys could lead to subtle errors in client code.
343
  }
344
};
345

346

347
/**
348
 * Returns the number of subscriptions to the given topic (or all topics if
349
 * unspecified). This number will not change while publishing any messages.
350
 * @param {string=} opt_topic The topic (all topics if unspecified).
351
 * @return {number} Number of subscriptions to the topic.
352
 */
353
goog.pubsub.PubSub.prototype.getCount = function(opt_topic) {
354
  if (opt_topic) {
355
    var keys = this.topics_[opt_topic];
356
    return keys ? keys.length : 0;
357
  }
358

359
  var count = 0;
360
  for (var topic in this.topics_) {
361
    count += this.getCount(topic);
362
  }
363

364
  return count;
365
};
366

367

368
/** @override */
369
goog.pubsub.PubSub.prototype.disposeInternal = function() {
370
  goog.pubsub.PubSub.base(this, 'disposeInternal');
371
  this.clear();
372
  this.pendingKeys_.length = 0;
373
};
374

375