When multiple BroadcastChannels with the same name
are created in a single node-process, we can access them directly and emit messages.
This might not happen often in production
but will speed up things when this module is used in unit-tests.
190686
function emitOverFastPath(state, msgObj, messageJson) {
  if (!state.options.node.useFastPath) return; // disabled

  var others = OTHER_INSTANCES[state.channelName].filter(function (s) {
    return s !== state;
  });
  var checkObj = {
    time: msgObj.time,
    senderUuid: msgObj.uuid,
    token: msgObj.token
  };
  others.filter(function (otherState) {
    return _filterMessage(checkObj, otherState);
  }).forEach(function (otherState) {
    otherState.messagesCallback(messageJson);
  });
}
892720
def channels_create(self, name, **kwargs):
        """"""
        return self.__call_api_post('channels.create', name=name, kwargs=kwargs)
688436
function BoundedBroadcast(source, max) {
    EventEmitter.call(this);
    this.cache = []; this.index = -1; this.max = max || 500; // cyclic cache
    this.source = source;
    var self = this;
    this.source.on('bounded-broadcast-receive', function(socket, message){
        self.receive(message);
    });
}
814796
public synchronized int getNumStartedChainsUsingChannel(String channelName) {
        int numStartedChains = 0;
        ChannelContainer channelContainer = this.channelRunningMap.get(channelName);
        if (null != channelContainer) {
            for (Chain chain : channelContainer.getChainMap().values()) {
                if (chain.getState() == RuntimeState.STARTED) {
                    numStartedChains++;
                }
            }
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "getNumStartedChainsUsingChannel: " + channelName + "=" + numStartedChains);
        }
        return numStartedChains;
    }
93713
@Override
    public boolean sequenceForReplay(VoltMessage message)
    {
        boolean canDeliver = false;
        long sequenceWithUniqueId = Long.MIN_VALUE;

        boolean commandLog = (message instanceof TransactionInfoBaseMessage &&
                (((TransactionInfoBaseMessage)message).isForReplay()));

        boolean sentinel = message instanceof MultiPartitionParticipantMessage;

        boolean replay = commandLog || sentinel;
        boolean sequenceForReplay = m_isLeader && replay;

        if (replay) {
            sequenceWithUniqueId = ((TransactionInfoBaseMessage)message).getUniqueId();
        }

        if (sequenceForReplay) {
            InitiateResponseMessage dupe = m_replaySequencer.dedupe(sequenceWithUniqueId,
                    (TransactionInfoBaseMessage) message);
            if (dupe != null) {
                // Duplicate initiate task message, send response
                m_mailbox.send(dupe.getInitiatorHSId(), dupe);
            }
            else if (!m_replaySequencer.offer(sequenceWithUniqueId, (TransactionInfoBaseMessage) message)) {
                canDeliver = true;
            }
            else {
                deliverReadyTxns();
            }

            // If it's a DR sentinel, send an acknowledgement
            if (sentinel && !commandLog) {
                MultiPartitionParticipantMessage mppm = (MultiPartitionParticipantMessage) message;
                final InitiateResponseMessage response = new InitiateResponseMessage(mppm);
                ClientResponseImpl clientResponse =
                        new ClientResponseImpl(ClientResponseImpl.UNEXPECTED_FAILURE,
                                new VoltTable[0], ClientResponseImpl.IGNORED_TRANSACTION);
                response.setResults(clientResponse);
                m_mailbox.send(response.getInitiatorHSId(), response);
            }
        }
        else {
            if (replay) {
                // Update last seen and last polled uniqueId for replicas
                m_replaySequencer.updateLastSeenUniqueId(sequenceWithUniqueId,
                        (TransactionInfoBaseMessage) message);
                m_replaySequencer.updateLastPolledUniqueId(sequenceWithUniqueId,
                        (TransactionInfoBaseMessage) message);
            }

            canDeliver = true;
        }

        return canDeliver;
    }
552757
def _bind_channels(self, events, channels):
        
        for channel_name in channels:
            if channel_name in self.channels:
                channel = self.pusher.subscribe(channel_name)
                if isinstance(events, list):
                    for event in events:
                        channel.bind(event, channels[channel_name])
                else:
                    channel.bind(events, channels[channel_name])
1013414
@SuppressWarnings("PMD.UseVarargs")
    public void add(EventBase<?> event, Channel[] channels) {
        add(new EventChannelsTuple(event, channels));
    }
928183
function broadcastQueuedMessages() {
            var message;
            while (messageQueue.length) {
                message = messageQueue.shift();
                broadcast(message.name, message.data);
            }
            messageQueue = null;
        }
108801
protected boolean processChannelMessage(SshChannelMessage msg)
			throws SshException {

		try {
			switch (msg.getMessageId()) {
			case SSH_MSG_CHANNEL_REQUEST:

				String requesttype = msg.readString();
				boolean wantreply = msg.read() != 0;
				byte[] requestdata = new byte[msg.available()];
				msg.read(requestdata);

				if (Log.isDebugEnabled()) {
					Log.debug(this,
							"Received SSH_MSG_CHANNEL_REQUEST id=" + channelid
									+ " rid=" + remoteid + " request="
									+ requesttype + " wantreply=" + wantreply);
				}
				channelRequest(requesttype, wantreply, requestdata);
				return true;
			case SSH_MSG_WINDOW_ADJUST:

				if (Log.isDebugEnabled()) {
					msg.mark(4);
					int len = (int) msg.readInt();
					Log.debug(this,
							"Received SSH_MSG_WINDOW_ADJUST id=" + channelid
									+ " rid=" + remoteid + " window="
									+ remotewindow.available() + " adjust="
									+ len);
					msg.reset();
				}

				return false;
			case SSH_MSG_CHANNEL_DATA:

				if (Log.isDebugEnabled()) {
					Log.debug(
							this,
							"Received SSH_MSG_CHANNEL_DATA id=" + channelid
									+ " rid=" + remoteid + " len="
									+ (msg.available() - 4) + " window="
									+ localwindow.available());
				}

				// Is the channels InputStream currently in a blocking read
				// operation? if it is then we can leave the message for it
				// to process (and subsequently break out of the block) or
				// if not we should process it here so that data events
				// are fired in a timely fashion
				if (autoConsumeInput) {
					localwindow.consume(msg.available() - 4);
					if (localwindow.available() <= localwindow.getInitialSize() / 2) {
						adjustWindow(localwindow.getInitialSize()
								- localwindow.available());
					}
				}

				for (Enumeration<ChannelEventListener> e = listeners.elements(); e
						.hasMoreElements();) {
					(e.nextElement()).dataReceived(Ssh2Channel.this,
							msg.array(), msg.getPosition() + 4,
							msg.available() - 4);
				}

				return autoConsumeInput;

			case SSH_MSG_CHANNEL_EXTENDED_DATA:

				if (Log.isDebugEnabled()) {
					Log.debug(this,
							"Received SSH_MSG_CHANNEL_EXTENDED_DATA id="
									+ channelid + " rid=" + remoteid + " len="
									+ (msg.available() - 4) + " window="
									+ localwindow.available());
				}

				int type = (int) ByteArrayReader.readInt(msg.array(),
						msg.getPosition());

				if (autoConsumeInput) {
					localwindow.consume(msg.available() - 8);
					if (localwindow.available() <= localwindow.getInitialSize() / 2) {
						adjustWindow(localwindow.getInitialSize()
								- localwindow.available());
					}
				}

				for (Enumeration<ChannelEventListener> e = listeners.elements(); e
						.hasMoreElements();) {
					(e.nextElement()).extendedDataReceived(Ssh2Channel.this,
							msg.array(), msg.getPosition() + 8,
							msg.available() - 8, type);
				}

				return autoConsumeInput;

			case SSH_MSG_CHANNEL_CLOSE:

				if (Log.isDebugEnabled()) {
					Log.debug(this,
							"Received SSH_MSG_CHANNEL_CLOSE id=" + channelid
									+ " rid=" + remoteid);
				}

				// Synchronize on message store and close
				synchronized (this) {
					if (!closing) {
						synchronized (ms) {
							if (!ms.isClosed())
								ms.close();
						}
					}
				}

				checkCloseStatus(true);
				return false;

			case SSH_MSG_CHANNEL_EOF:

				if (Log.isDebugEnabled()) {
					Log.debug(this, "Received SSH_MSG_CHANNEL_EOF id="
							+ channelid + " rid=" + remoteid);
				}

				isRemoteEOF = true;

				for (Enumeration<ChannelEventListener> e = listeners.elements(); e
						.hasMoreElements();) {
					(e.nextElement()).channelEOF(Ssh2Channel.this);
				}

				// Fire the EOF event, do we need this on the listener?
				channelEOF();

				if (isLocalEOF) {
					close();
				}
				return false;

			default:
				return false;
			}
		} catch (IOException ex) {
			throw new SshException(ex, SshException.INTERNAL_ERROR);
		}

	}
1229867
private boolean reconnectToChannel(String name) {
        boolean reConnected = false;
        Enumeration callbackStructs = event_callback_map.elements();
        while (callbackStructs.hasMoreElements()) {
            EventCallBackStruct eventCallBackStruct = (EventCallBackStruct) callbackStructs.nextElement();
            if (eventCallBackStruct.channel_name.equals(name) && (eventCallBackStruct.callback != null)) {
                try {
                    EventChannelStruct channelStruct = channel_map.get(name);
                    DevVarLongStringArray   lsa =
                            ZMQutils.getEventSubscriptionInfoFromAdmDevice(
                                    channelStruct.adm_device_proxy,
                                    eventCallBackStruct.device.name(), eventCallBackStruct.attr_name, eventCallBackStruct.event_name);

                    //  Re Connect heartbeat
                    ZMQutils.connectHeartbeat(channelStruct.adm_device_proxy.get_tango_host(),
                                channelStruct.adm_device_proxy.name(), lsa, true);
                    reConnected = true;
                } catch (DevFailed e1) {
                    //Except.print_exception(e1);
                    reConnected = false;
                }
                break;
            }
        }
        return reConnected;
    }
559657
function broadcast(messageName, data) {
            var i, len, instance, messages;
            for (i = 0, len = instances.length; i < len; ++i) {
                instance = instances[i];
                if (!instance) {
                    continue;
                }
                messages = instance.messages || [];

                if (util.inArray(messageName, messages) !== -1) {
                    if (util.isFn(instance.onmessage)) {
                        instance.onmessage.call(instance, messageName, data);
                    }
                }
            }
        }
969376
def module_broadcast(m, broadcast_fn, *args, **kwargs):
    
    apply_leaf(m, lambda x: module_apply_broadcast(x, broadcast_fn, args, kwargs))
669932
def broadcast(*sinks_):
    
    @push
    def bc():
        sinks = [s() for s in sinks_]
        while True:
            msg = yield
            for s in sinks:
                s.send(msg)

    return bc
456814
function _getServerStreamingInterceptor(method_definition, channel, emitter) {
  var deserialize = common.wrapIgnoreNull(
    method_definition.responseDeserialize);
  var serialize = method_definition.requestSerialize;
  return function (options) {
    var batch_state = {
      completed_ops: [],
      deferred_batches: []
    };
    var call = getCall(channel, method_definition.path, options);
    var final_requester = {};
    var first_listener;
    var get_listener = function() {
      return first_listener;
    };
    final_requester.start = function(metadata, listener) {
      first_listener = listener;
      metadata = metadata.clone();
      var metadata_batch = {
        [grpc.opType.SEND_INITIAL_METADATA]: metadata._getCoreRepresentation(),
        [grpc.opType.RECV_INITIAL_METADATA]: true
      };
      var callback = function(err, response) {
        if (err) {
          // The call has stopped for some reason. A non-OK status will arrive
          // in the other batch.
          return;
        }
        first_listener.onReceiveMetadata(
          Metadata._fromCoreRepresentation(response.metadata));
      };
      batch_state = _startBatchIfReady(call, metadata_batch, batch_state,
                                       callback);
      var status_batch = {
        [grpc.opType.RECV_STATUS_ON_CLIENT]: true
      };
      call.startBatch(status_batch, function(err, response) {
        if (err) {
          emitter.emit('error', err);
          return;
        }
        response.status.metadata = Metadata._fromCoreRepresentation(
          response.status.metadata);
        first_listener.onReceiveStatus(response.status);
      });
    };
    final_requester.sendMessage = function(argument) {
      var message = serialize(argument);
      if (options) {
        message.grpcWriteFlags = options.flags;
      }
      var send_batch = {
        [grpc.opType.SEND_MESSAGE]: message
      };
      var callback = function(err, response) {
        if (err) {
          // The call has stopped for some reason. A non-OK status will arrive
          // in the other batch.
          return;
        }
      };
      batch_state = _startBatchIfReady(call, send_batch, batch_state, callback);
    };
    final_requester.halfClose = function() {
      var batch = {
        [grpc.opType.SEND_CLOSE_FROM_CLIENT]: true
      };
      batch_state = _startBatchIfReady(call, batch, batch_state, function() {});
    };
    final_requester.recvMessageWithContext = function(context) {
      var recv_batch = {
        [grpc.opType.RECV_MESSAGE]: true
      };
      var callback = _getStreamReadCallback(emitter, call,
        get_listener, deserialize);
      batch_state = _startBatchIfReady(call, recv_batch, batch_state, callback);
    };
    final_requester.cancel = function() {
      call.cancel();
    };
    final_requester.cancelWithStatus = function(code, details) {
      call.cancelWithStatus(code, details)
    };
    final_requester.getPeer = function() {
      return call.getPeer();
    };
    return new InterceptingCall(null, final_requester);
  };
}
525192
@Override
    public synchronized ChannelData[] getRunningChannels() {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "getRunningChannels");
        }

        // Note, runtime has child data objects so duplicates may be found.
        List<ChannelData> list = new ArrayList<ChannelData>();
        for (ChannelContainer channel : this.channelRunningMap.values()) {
            list.add(channel.getChannelData().getParent());
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "getRunningChannels");
        }
        return list.toArray(new ChannelData[list.size()]);
    }
703989
private synchronized boolean createChannel() {

        final String url = settings.getBounceProxyUrl().buildCreateChannelUrl(channelId);

        HttpPost postCreateChannel = httpRequestFactory.createHttpPost(URI.create(url.trim()));
        postCreateChannel.setConfig(defaultRequestConfig);
        postCreateChannel.addHeader(httpConstants.getHEADER_X_ATMOSPHERE_TRACKING_ID(), receiverId);
        postCreateChannel.addHeader(httpConstants.getHEADER_CONTENT_TYPE(), httpConstants.getAPPLICATION_JSON());

        channelUrl = null;

        CloseableHttpResponse response = null;
        boolean created = false;
        try {
            response = httpclient.execute(postCreateChannel);
            StatusLine statusLine = response.getStatusLine();
            int statusCode = statusLine.getStatusCode();
            String reasonPhrase = statusLine.getReasonPhrase();

            switch (statusCode) {
            case HttpURLConnection.HTTP_OK:
            case HttpURLConnection.HTTP_CREATED:
                try {
                    Header locationHeader = response.getFirstHeader(httpConstants.getHEADER_LOCATION());
                    channelUrl = locationHeader != null ? locationHeader.getValue() : null;
                } catch (Exception e) {
                    throw new JoynrChannelMissingException("channel url was null");
                }
                break;

            default:
                logger.error("createChannel channelId: {} failed. status: {} reason: {}",
                             channelId,
                             statusCode,
                             reasonPhrase);
                throw new JoynrChannelMissingException("channel url was null");
            }

            created = true;
            logger.info("createChannel channelId: {} returned channelUrl {}", channelId, channelUrl);

        } catch (ClientProtocolException e) {
            logger.error("createChannel ERROR reason: {}", e.getMessage());
        } catch (IOException e) {
            logger.error("createChannel ERROR reason: {}", e.getMessage());
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException e) {
                    logger.error("createChannel ERROR reason: {}", e.getMessage());
                }
            }
        }

        return created;
    }
745759
public Channel newChannel(String name) throws InvalidArgumentException {
        clientCheck();
        if (Utils.isNullOrEmpty(name)) {
            throw new InvalidArgumentException("Channel name can not be null or empty string.");
        }

        synchronized (channels) {

            if (channels.containsKey(name)) {
                throw new InvalidArgumentException(format("Channel by the name %s already exists", name));
            }
            logger.trace("Creating channel :" + name);
            Channel newChannel = Channel.createNewInstance(name, this);

            channels.put(name, newChannel);
            return newChannel;

        }

    }