notify next when event comes, if the callback returns 'reNotifyOfNext', then re-subscribe using the same id
17542
function handleNotifyNext(e) {
  if (notifyOfNextSubs[e.type].length === 0) return;
  e.persist = blankFunction;
  const reNotifyOfNext = [];
  const reNotifyOfNextIDs = {};
  notifyOfNextSubs[e.type].forEach(sub => {
    if (sub.callback(e) === 'reNotifyOfNext') {
      reNotifyOfNextIDs[sub.id] = reNotifyOfNext.push(sub) - 1;
    }
  });
  notifyOfNextSubs[e.type] = reNotifyOfNext;
  subsIDs[e.type] = reNotifyOfNextIDs;
}
97098
def subscribe(obj, event, callback, event_state=None):
    
    if inspect.isclass(obj):
        cls = obj.__name__
    else:
        cls = obj.__class__.__name__

    if event_state is None:
        event_state = states.ANY

    event_key = '.'.join([cls, event, event_state])
    if event_key not in EVENT_HANDLERS:
        EVENT_HANDLERS[event_key] = []

    EVENT_HANDLERS[event_key].append(callback)
    return
1052585
def visit(hook_event, *data)
      each { |subscriber|
        synchronize { hook_event.notify(subscriber, *data) }
      }
    end
886821
function subscribe() {
  openSTNotification.subscribeEvent.rabbit(
    ['#'],
    {queue: 'openst_platform'},
    function (msgContent) {
      logger.debug('[RECEIVED]', msgContent, '\n');
    }
  ).catch(function (err) {
    logger.error(err);
  });
}
559150
function(name, callback, scope, once) {

            var events = null;
            if (!(events = this._events[name])) {
                events = this._events[name] = [];
            }

            events.push({
                callback: callback,
                scope: scope || this,
                once: once,
                fired: false
            });

            return callback;

        }
1215997
function subscribe(subscriber, immediate) {
        if (!~subscribers.indexOf(subscriber)) {
            subscribers.push(subscriber);
        }
        if (immediate) {
            subscriber(value);
        }
    }
1072547
function notify(listener) {
    var func = state == webdriver.promise.Deferred.State_.RESOLVED ?
        listener.callback : listener.errback;
    if (func) {
      flow.runInNewFrame_(goog.partial(func, value),
          listener.fulfill, listener.reject);
    } else if (state == webdriver.promise.Deferred.State_.REJECTED) {
      listener.reject(value);
    } else {
      listener.fulfill(value);
    }
  }
1156891
function on(eventName, callback) {
  	if (objectType(eventName) !== "string") {
  		throw new TypeError("eventName must be a string when registering a listener");
  	} else if (!inArray(eventName, SUPPORTED_EVENTS)) {
  		var events = SUPPORTED_EVENTS.join(", ");
  		throw new Error("\"" + eventName + "\" is not a valid event; must be one of: " + events + ".");
  	} else if (objectType(callback) !== "function") {
  		throw new TypeError("callback must be a function when registering a listener");
  	}

  	if (!LISTENERS[eventName]) {
  		LISTENERS[eventName] = [];
  	}

  	// Don't register the same callback more than once
  	if (!inArray(callback, LISTENERS[eventName])) {
  		LISTENERS[eventName].push(callback);
  	}
  }
472679
def once(event, &block)
      add_listener(event) do |*args, &callback|
        block.call(*args, &callback)
        :delete
      end
    end
69541
private boolean resubscribe() throws TransactionIdTooOldException,
      InterruptedException {
    for (NamespaceEventKey eventKey : watchedEvents.keySet()) {
      NamespaceEvent event = eventKey.getEvent(); 
      if (!subscribe(event.getPath(), EventType.fromByteValue(event.getType()),
          watchedEvents.get(eventKey))) {
        return false;
      }
    }
    return true;
  }
1147287
def subscribe(self, event):
        """"""
        uuids = event.data

        if not isinstance(uuids, list):
            uuids = [uuids]

        subscribed = []
        for uuid in uuids:
            try:
                self._add_subscription(uuid, event)
                subscribed.append(uuid)
            except KeyError:
                continue

        result = {
            'component': 'hfos.events.objectmanager',
            'action': 'subscribe',
            'data': {
                'uuid': subscribed, 'success': True
            }
        }
        self._respond(None, result, event)
1348940
private void notifyListenersIfUninhibited(R oldValue, R newValue) {
        if (inhibited) {
            inhibitCount++;
            lastInhibitedValue = newValue;
        } else {
            lastInhibitedValue = newValue; // Just in case, even though not really necessary
            lastNonInhibitedValue = newValue;
            doNotifyListeners(oldValue, newValue);
        }
    }
789147
public void _onNext(IN in) {
        if ( subscribers == null )
            return;
        openRequested--;
        try {
            OUT apply = processor.apply(in);
            if ( apply != null ) {
                forwardMessage(apply);
            } else {
                emitRequestNext();
            }
        } catch (Throwable err) {
            err.printStackTrace();
            forwardError(err);
        }
    }
866306
def subscribe(self, name, params=[], callback=None):
        """"""
        self._wait_for_connect()

        def subscribed(error, sub_id):
            if error:
                self._remove_sub_by_id(sub_id)
                if callback:
                    callback(error.get('reason'))
                return
            if callback:
                callback(None)
            self.emit('subscribed', name)

        if name in self.subscriptions:
            raise MeteorClientException('Already subcribed to {}'.format(name))

        sub_id = self.ddp_client.subscribe(name, params, subscribed)
        self.subscriptions[name] = {
            'id': sub_id,
            'params': params
        }
1202346
def subscribe(caller = nil, &block)
      if block_given?
        notify_when = block.call(state)
        @listeners << { obj: block.binding.receiver, notify_when: notify_when }
      else
        @listeners << { obj: caller }
      end
    end
968373
function subscribe() {
    console.log(' > ', 'init', ['xmlrpc_bin://' + thisHost + ':2031', 'test123']);
    rpcClient.methodCall('init', ['xmlrpc_bin://' + thisHost + ':2031', 'test123'], function (err, res) {
        console.log(err, res);
    });
}
502953
function Subscriber(opts) {
  opts = opts || {};

  this._channel = opts.channel || 'default';
  this._subscribers = [];

  /* for mocking */
  var addEventListener = opts.addEventListener || window.addEventListener;
  var attachEvent = opts.attachEvent || window.attachEvent;

  var _this = this;
  function onMessage(e) {
    var event = parse(e.data);
    if (event.channel === _this._channel) {
      for (var i = 0, len = _this._subscribers.length; i < len; i++) {
        _this._subscribers[i].apply({}, event.args);
      }
    }
  }

  if (addEventListener) {
    addEventListener('message', onMessage, false);
  } else {
    attachEvent('onmessage', onMessage, false);
  }
}