to be sent downstream
176918
public List sendValueMessages(
    List msgList,
    long completedPrefix,
    boolean requestedOnly,
    int priority,
    Reliability reliability,
    SIBUuid12 stream) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "sendValueMessages",
        new Object[] { msgList,
                       new Long(completedPrefix),
                       new Boolean(requestedOnly)});

    // We will send the messages using MPIO
    // Work out the cellule pair needed by the
    // messageTransmitter
    SIBUuid8[] fromTo = new SIBUuid8[1];
    fromTo[0] = _targetMEUuid;

    JsMessage jsMsg = null;

    TickRange tickRange = null;
    MessageItem msgItem = null;
    long msgId = -1;
    List<TickRange> expiredMsgs = null;
    
    for (int i = 0; i < msgList.size(); i++)
    {
      tickRange = (TickRange)msgList.get(i);
     
      // Get the messageStore Id from the stream
      msgId = tickRange.itemStreamIndex;

      // Retrieve the message from the non-persistent ItemStream
      msgItem =
        _destinationHandler.
          getPubSubRealization().
            retrieveMessageFromItemStream(msgId);
                  
      // If the item wasn't found it has expired  
      if ( msgItem == null )
      {
        if ( expiredMsgs == null )
          expiredMsgs = new ArrayList<TickRange>();
        
        expiredMsgs.add(tickRange);   

        // In this case send Silence instead      
        ControlMessage cMsg = createSilenceMessage(tickRange.valuestamp,
                                                   completedPrefix,
                                                   priority,
                                                   reliability,
                                                   stream);

        ((ControlSilence)cMsg).setRequestedOnly(requestedOnly);
             
        // If the destination in a Link add Link specific properties to message
        if( _isLink )
        {
          cMsg = addLinkProps(cMsg);
          cMsg.setRoutingDestination( _routingDestination );  
        }
                  
        // Send the message to the MessageTransmitter
        // Send at priority+1 if this is a response to a Nack
        if( requestedOnly )
          _mpio.sendDownTree(fromTo, priority+1, cMsg);
        else
          _mpio.sendDownTree(fromTo, priority, cMsg);
  
      }
      else
      {
        try
        {
          // PM34074
          // Retrieve a copy of the message from MessageItem
          jsMsg = msgItem.getMessage().getReceived();
        }
        catch(MessageCopyFailedException e)
        {
          FFDCFilter.processException(
            e,
            "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendValueMessages",
            "1:960:1.164.1.5",
            this);
            
          if (tc.isEntryEnabled())
            SibTr.exit(tc, "sendValueMessages", "SIErrorException: "+e);  
          throw new SIErrorException(e);
        }
      
        // Modify the streamId if necessary     
        if( jsMsg.getGuaranteedStreamUUID() !=  stream)
        {
          jsMsg.setGuaranteedStreamUUID(stream);
        }

        // Are there Completed ticks after this Value
        // If so we need to adjust the message to reflect this
        if (tickRange.endstamp > tickRange.valuestamp)
        {
          jsMsg.setGuaranteedValueEndTick(tickRange.endstamp);
        }
      
        // Update the completedPrefix to current value
        jsMsg.setGuaranteedValueCompletedPrefix(completedPrefix);

        // Set the requestedOnly flag
        jsMsg.setGuaranteedValueRequestedOnly(requestedOnly);
      
        // If the destination in a Link add Link specific properties to message
        if( _isLink )
        {
          jsMsg = addLinkProps(jsMsg);
          jsMsg.setRoutingDestination( _routingDestination );  
        }
      
        if (TraceComponent.isAnyTracingEnabled() && UserTrace.tc_mt.isDebugEnabled())
        {
          DestinationHandler dest = 
            _messageProcessor.getDestinationManager().getDestinationInternal(_destinationHandler.getUuid(), false);
          
          if (dest != null)
            UserTrace.traceOutboundSend(jsMsg, 
                                        _neighbour.getUUID(), 
                                        dest.getName(),
                                        dest.isForeignBus() || dest.isLink(),
                                        dest.isMQLink(),
                                        dest.isTemporary());
          else
            UserTrace.traceOutboundSend(jsMsg, 
                                        _neighbour.getUUID(), 
                                        _destinationHandler.getUuid().toString().toString(),
                                        false,
                                        false,
                                        false);

        }
        
        // Send the message to the MessageTransmitter
        // Send at priority+1 if this is a response to a Nack
        if( requestedOnly )
          _mpio.sendDownTree(fromTo, priority+1, jsMsg);
        else
          _mpio.sendDownTree(fromTo, priority, jsMsg);
                        
      }
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "sendValueMessages", expiredMsgs);
      
    return expiredMsgs;  
  }
962179
public static void transfer(final InputStream input,
			final OutputStream output) throws Exception {
		final byte[] data = new byte[SIZE];
		while (true) {
			final int count = input.read(data, 0, SIZE);
			if (count == EOF) {
				break;
			}
			output.write(data, 0, count);
		}
	}
455855
def _process_outgoing_msg(self, sink_iter):
        
        LOG.debug('NetworkController processing outgoing request list.')
        # TODO(PH): We should try not to sent routes from bgp peer that is not
        # in established state.
        from ryu.services.protocols.bgp.model import (
            FlexinetOutgoingRoute)
        while self.is_connected:
            # sink iter is Sink instance and next is blocking so this isn't
            # active wait.
            for outgoing_msg in sink_iter:
                if not self.is_connected:
                    self._socket.close()
                    return
                if isinstance(outgoing_msg, FlexinetOutgoingRoute):
                    rpc_msg = _create_prefix_notification(outgoing_msg, self)
                else:
                    raise NotImplementedError(
                        'Do not handle out going message of type %s' %
                        outgoing_msg.__class__)
                if rpc_msg:
                    self._sendall(rpc_msg)
            self.pause(0)

        # Stop incoming connection.
        if self.green_in:
            self.green_in.kill()
1020125
def _send_chunk(self, index, chunk):
        
        self._pending_chunks += 1
        self.outbox.put((index, chunk))
1247262
def _process_outgoing(self, xmlstream, token):
        

        self._send_stanza(xmlstream, token)
        # try to send a bulk
        while True:
            try:
                token = self._active_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            self._send_stanza(xmlstream, token)

        if self._sm_enabled:
            self._logger.debug("sending SM req")
            xmlstream.send_xso(nonza.SMRequest())
477671
function pipeEvents(source, destination, event) {
  source.on(event, function (data) {
    destination.emit(event, data);
  });
}
177991
def send(self, packet, interfaces=None):
        """"""
        interfaces = interfaces or self.interfaces  # default to all interfaces
        interfaces = interfaces if hasattr(interfaces, '__iter__') else [interfaces]

        for interface in interfaces:
            for f in self.filters:
                packet = f.tx(packet, interface)  # run outgoing packet through the filters
            if packet:
                # if not dropped, log the transmit and pass it to the interface's send method
                # self.log("OUT     ", ("<"+",".join(i.name for i in interfaces)+">").ljust(30), packet.decode())
                interface.send(packet)
1326605
private void push(double[][] flow, double[] excess, int u, int v) {
        double send = Math.min(excess[u], graph[u][v] - flow[u][v]);
        flow[u][v] += send;
        flow[v][u] -= send;
        excess[u] -= send;
        excess[v] += send;
    }
899861
async def send_to(self, destination: discord.abc.Messageable):
        

        self.message = await destination.send(**self.send_kwargs)

        # add the close reaction
        await self.message.add_reaction(self.emojis.close)

        # if there is more than one page, and the reactions haven't been sent yet, send navigation emotes
        if not self.sent_page_reactions and self.page_count > 1:
            await self.send_all_reactions()

        if self.task:
            self.task.cancel()

        self.task = self.bot.loop.create_task(self.wait_loop())
861493
private static PumpStreamHandler redirectOutputAlsoTo(PumpStreamHandler pumps, OutputStream output) {
    if (output == null)
      throw new IllegalArgumentException("OutputStream must be provided.");
    OutputStream current = pumps.getOut();
    if (current != null && !(current instanceof NullOutputStream)) {
      output = new TeeOutputStream(current, output);
    }
    return new PumpStreamHandler(output, pumps.getErr(), pumps.getInput());
  }
114219
void send(byte streamCode, byte[] b, int offset, int len) throws IOException {
    synchronized (writeLock) {
      out.writeInt(len);
      out.writeByte(streamCode);
      out.write(b, offset, len);
    }
    out.flush();
  }
1082468
private void deliver() {
    try {
      deliverUnsafe();
    } catch (Throwable t) {
      // This should never happen. If does, it means we are in an inconsistent state and should
      // close the stream and further processing should be prevented. This is accomplished by
      // purposefully leaving the lock non-zero and notifying the outerResponseObserver of the
      // error. Care must be taken to avoid calling close twice in case the first invocation threw
      // an error.
      innerController.cancel();
      if (!finished) {
        outerResponseObserver.onError(t);
      }
    }
  }
342792
private boolean transfer(long total,
                             GlobusInputStream in,
                             GlobusOutputStream out)
        throws IOException {

        byte [] buffer       = new byte[bufferSize];
        int bytes            = 0;
        long totalBytes      = total;
        long transferedBytes = 0;

        if (total == -1) {
            while( (bytes = in.read(buffer)) != -1) {
                out.write(buffer, 0, bytes);
                out.flush();

                if (listeners != null) {
                    transferedBytes += bytes;
                    fireUrlTransferProgressEvent(totalBytes,
                                                 transferedBytes);
                }
                if (isCanceled()) return false;
            }
        } else {
            while ( total != 0 ) {

                bytes = bufferSize;
                if (total < bufferSize) bytes = (int)total;

                bytes = in.read(buffer);
                out.write(buffer, 0, bytes);
                out.flush();
                total -= bytes;

                if (listeners != null) {
                    transferedBytes += bytes;
                    fireUrlTransferProgressEvent(totalBytes,
                                                 transferedBytes);
                }
                if (isCanceled()) return false;
            }
        }

        return true;
    }
1016377
@Override
  public void deliver(WriteStream os, OutHttp2 outHttp)
    throws IOException
  {
    outHttp.writeFlow(_streamId, _credit);
  }
830130
def send(self, data):
        
        self.stdin.write(data)
        self.stdin.flush()
1291386
@Override
  public void deliver(WriteStream os, OutHttp2 out)
    throws IOException
  {
    _stream.writeCont(out);
  }
953933
def outgoing(message)
      (m = filter(message, :outgoing)) && each {|l| l.outgoing(m)}
    end