On resource events, if applyable, attempt to alter channels to drop the
global subscription and set it to group channels
543278
public function onSiteResourceEventAlterChannels(Event $event)
    {
        if (!$this->groupManager || !$event instanceof ResourceEvent) {
            return;
        }

        $done = [];

        // Fetch site groups, and yes we don't really have a choice here
        $sites = $this->siteManager->getStorage()->loadAll($event->getResourceIdList(), false);
        foreach ($sites as $site) {
            $groupId = (int)$site->getGroupId();
            if ($groupId) {
                $done[$groupId] = $groupId;
            }
        }

        foreach ($done as $groupId) {
            $event->addArbitraryChanId('admin:site:' . $groupId);
        }
    }
331406
function(subscriptions){
      if (subscriptions) _.extend(this.subscriptions || {}, subscriptions);
      subscriptions = subscriptions || this.subscriptions;
      if (!subscriptions || _.isEmpty(subscriptions)) return;
      // Just to be sure we don't set duplicate
      this.unsetSubscriptions(subscriptions);

      _.each(subscriptions, function(subscription, channel){
        var once;
        if (subscription.$once) {
          subscription = subscription.$once;
          once = true;
        }
        if (_.isString(subscription)) {
          subscription = this[subscription];
        }
        Backbone.Mediator.subscribe(channel, subscription, this, once);
      }, this);
    }
1164760
def on_connect(self, connection):
        ""
        # NOTE: for python3, we can't pass bytestrings as keyword arguments
        # so we need to decode channel/pattern names back to unicode strings
        # before passing them to [p]subscribe.
        self.pending_unsubscribe_channels.clear()
        self.pending_unsubscribe_patterns.clear()
        if self.channels:
            channels = {}
            for k, v in iteritems(self.channels):
                channels[self.encoder.decode(k, force=True)] = v
            self.subscribe(**channels)
        if self.patterns:
            patterns = {}
            for k, v in iteritems(self.patterns):
                patterns[self.encoder.decode(k, force=True)] = v
            self.psubscribe(**patterns)
773006
private void reSubscribeByName(EventChannelStruct eventChannelStruct, String name) {

        //  Get the map and the callback structure for channel
        Hashtable<String, EventCallBackStruct>
                callBackMap = EventConsumer.getEventCallbackMap();
        EventCallBackStruct callbackStruct = null;
        Enumeration channelNames = callBackMap.keys();
        while (channelNames.hasMoreElements()) {
            String  key = (String) channelNames.nextElement();
            EventCallBackStruct eventStruct = callBackMap.get(key);
            if (eventStruct.channel_name.equals(name)) {
                callbackStruct = eventStruct;
            }
        }

        //  Get the callback structure
        if (callbackStruct!=null) {
            callbackStruct.consumer.reSubscribeByName(eventChannelStruct, name);
        }
    }
1318063
def subscribe(self, client, channel_name):
        """"""
        if channel_name not in self.channels:
            self.channels[channel_name] = channel = Channel(channel_name)
            channel.start()

        self.channels[channel_name].subscribe(client)
964600
@Override
    protected boolean reSubscribe(EventChannelStruct event_channel_struct,
                                  EventCallBackStruct callback_struct) {
        boolean retVal = true;

        try {
            DeviceData subscriber_in = new DeviceData();
            String[] subscriber_info = new String[4];
            subscriber_info[0] = callback_struct.device.name();
            subscriber_info[1] = callback_struct.attr_name;
            subscriber_info[2] = "subscribe";
            subscriber_info[3] = callback_struct.event_name;
            subscriber_in.insert(subscriber_info);
            event_channel_struct.adm_device_proxy.command_inout("EventSubscriptionChange", subscriber_in);
            event_channel_struct.heartbeat_skipped = false;
            event_channel_struct.last_subscribed = System.currentTimeMillis();
            callback_struct.last_subscribed = event_channel_struct.last_subscribed;
        } catch (Exception e) {
            retVal = false;
        }
        return retVal;
    }
198909
function (ev, comp, params, state) {
                for (var id in comp.__subscription) {
                    if (!_cs.isown(comp.__subscription, id))
                        continue;
                    var s = comp.__subscription[id];
                    if (   (   (state === "capturing" && s.capturing)
                            || (state === "targeting"               )
                            || (state === "spreading" && s.spreading)
                            || (state === "bubbling"  && s.bubbling ))
                        && ev.matches(s.name, s.spec)                 ) {

                        /*  verbosity  */
                        if (!params.silent)
                            $cs.debug(1, "event: " + comp.path("/") + ": dispatch " + ev.name() + " to subscriber on " + state);

                        /*  further annotate event object  */
                        ev.state(state);
                        ev.decline(false);

                        /*  call subscription method  */
                        var args = _cs.concat(
                            s.noevent ? [] : [ ev ],
                            s.args,
                            params.args
                        );
                        var result = s.func.apply(s.ctx, args);

                        /*  process return value  */
                        if (s.noevent && _cs.isdefined(result))
                            ev.result(params.resultstep(ev.result(), result));

                        /*  control the further dispatching  */
                        if (!ev.decline()) {
                            ev.dispatched(true);
                            if (params.firstonly)
                                ev.propagation(false);
                        }
                    }
                }
            }
649466
private <T> void manageSubscription(EventStream<T> stream, Consumer<T> consumer) {
        manageSubscription(stream.subscribe(consumer));
    }
597033
@Handler
    public void onPurge(Purge event, IOSubchannel channel) {
        // Needn't close this more than once
        event.stop();
        channel.respond(new Close());
    }
903407
private void handleChannelCategory(JsonNode channel) {
        long serverId = channel.get("guild_id").asLong();
        api.getPossiblyUnreadyServerById(serverId).ifPresent(server -> {
            ChannelCategory channelCategory = ((ServerImpl) server).getOrCreateChannelCategory(channel);
            ServerChannelCreateEvent event = new ServerChannelCreateEventImpl(channelCategory);
            api.getEventDispatcher().dispatchServerChannelCreateEvent((DispatchQueueSelector) server, server, event);
        });
    }
117654
private void removeOverlappingSubscriptionEvents(final LinkedList<SubscriptionEvent> events) {
        final Iterator<SubscriptionEvent> iterator = events.iterator();
        final Map<String, DefaultSubscriptionEvent> prevPerService = new HashMap<String, DefaultSubscriptionEvent>();
        while (iterator.hasNext()) {
            final DefaultSubscriptionEvent current = (DefaultSubscriptionEvent) iterator.next();
            final DefaultSubscriptionEvent prev = prevPerService.get(current.getServiceName());
            if (prev != null) {
                if (current.overlaps(prev)) {
                    iterator.remove();
                } else {
                    prevPerService.put(current.getServiceName(), current);
                }
            } else {
                prevPerService.put(current.getServiceName(), current);
            }
        }
    }
905780
function(client, subscriptions, backlog, channel, callback, done_callback) {
    if ( addToBacklog(client, backlog, unsubscribeBrowser, [client, subscriptions, backlog, channel, callback, done_callback]) ) return;
    if (subscriptions[channel]===undefined)
        return;
    subscriptions[channel].splice(subscriptions[channel].indexOf(callback), 1);
    if (subscriptions[channel].length===0) {
        delete subscriptions[channel];
        client.unsubscribe(channel, {onSuccess : function() {
            // If there is a done_callback defined, execute it
            if (done_callback!==undefined) done_callback();
        }});
    }
}
589869
@Handler(priority = 1000)
    public void onProtocolSwitchAccepted(
            ProtocolSwitchAccepted event, IOSubchannel channel) {
        event.requestEvent().associated(Session.class)
            .ifPresent(session -> {
                channel.setAssociated(Session.class, session);
            });
    }
893009
private function doBeforeCompileLaziness(): void
	{
		$builder = $this->getContainerBuilder();
		$dispatcher = $builder->getDefinition($this->prefix('dispatcher'));

		$subscribers = $builder->findByType(EventSubscriberInterface::class);
		foreach ($subscribers as $name => $subscriber) {
			$events = call_user_func([$subscriber->getEntity(), 'getSubscribedEvents']);

			/**
			 * ['eventName' => 'methodName']
			 * ['eventName' => ['methodName', $priority]]
			 * ['eventName' => [['methodName1', $priority], ['methodName2']]]
			 */
			foreach ($events as $event => $args) {
				if (is_string($args)) {
					if (!method_exists((string) $subscriber->getType(), $args)) {
						throw new ServiceCreationException(sprintf('Event listener %s does not have callable method %s', $subscriber->getType(), $args));
					}

					$dispatcher->addSetup('addSubscriberLazy', [$event, $name]);
				} elseif (is_string($args[0])) {
					if (!method_exists((string) $subscriber->getType(), $args[0])) {
						throw new ServiceCreationException(sprintf('Event listener %s does not have callable method %s', $subscriber->getType(), $args[0]));
					}

					$dispatcher->addSetup('addSubscriberLazy', [$event, $name]);
				} else {
					foreach ($args as $arg) {
						if (!method_exists((string) $subscriber->getType(), $arg[0])) {
							throw new ServiceCreationException(sprintf('Event listener %s does not have callable method %s', $subscriber->getType(), $arg[0]));
						}

						$dispatcher->addSetup('addSubscriberLazy', [$event, $name]);
					}
				}
			}
		}
	}
657809
def bind_channel(self, channel, callback):
        """"""

        # Remove the following list when deprecation time expires
        if channel in self.CHANNELS:
            warning = (
                "The event name '{}' will soon be deprecated. Use "
                "the real event name '{}' instead."
            ).format(channel, self.CHANNELS[channel])

            self.handle_error(warning)
            channel = self.CHANNELS[channel]
        # -------------------------------------------------------

        if channel == self.EVENT_NAME_ERROR:
            self.error_callback = callback
        elif channel == self.EVENT_NAME_RESULTS:
            self.socketIO.on(channel, partial(self.unpack_results, callback))
        else:
            self.socketIO.on(channel, callback)
501475
def drain_events(self, allowed_methods=None, timeout=None):
        """"""
        return self.wait_multi(self.channels.values(), timeout=timeout)
875617
function(client, subscriptions, channel, callback, done_callback) {
    if (subscriptions[channel]===undefined)
        return;
    subscriptions[channel].splice(subscriptions[channel].indexOf(callback), 1);
    if (subscriptions[channel].length===0) {
        delete subscriptions[channel];
        client.unsubscribe(channel, function() {
            // If there is a done_callback defined, execute it
            if (done_callback!==undefined) done_callback();
        });
    }
}