actual send on the SenderLink should happen only in this method & should run on Reactor Thread
483555
private void processSendWork()
	{
		synchronized (this.pendingSendLock)
		{
			if(!this.isSendLoopRunning)
			{
				this.isSendLoopRunning = true;
			}
			else
			{
				return;
			}
		}

		TRACE_LOGGER.debug("Processing pending sends to '{}'. Available link credit '{}'", this.sendPath, this.linkCredit);
		try
		{
			if(!this.ensureLinkIsOpen().isDone())
			{
				// Link recreation is pending
				return;
			}

			final Sender sendLinkCurrent = this.sendLink;
			while (sendLinkCurrent != null
					&& sendLinkCurrent.getLocalState() == EndpointState.ACTIVE && sendLinkCurrent.getRemoteState() == EndpointState.ACTIVE
					&& this.linkCredit > 0)
			{
				final WeightedDeliveryTag deliveryTag;
				final SendWorkItem<DeliveryState> sendData;
				synchronized (this.pendingSendLock)
				{
					deliveryTag = this.pendingSends.poll();
					if (deliveryTag == null)
					{
						TRACE_LOGGER.debug("There are no pending sends to '{}'.", this.sendPath);
						// Must be done inside this synchronized block
						this.isSendLoopRunning = false;
						break;
					}
					else
					{
						sendData = this.pendingSendsData.get(deliveryTag.getDeliveryTag());
						if(sendData == null)
						{
							TRACE_LOGGER.debug("SendData not found for this delivery. path:{}, linkName:{}, deliveryTag:{}", this.sendPath, this.sendLink.getName(), deliveryTag);
							continue;
						}
					}
				}

				if (sendData.getWork() != null && sendData.getWork().isDone())
				{
					// CoreSend could enqueue Sends into PendingSends Queue and can fail the SendCompletableFuture
					// (when It fails to schedule the ProcessSendWork on reactor Thread)
					this.pendingSendsData.remove(sendData);
					continue;
				}

				Delivery delivery = null;
				boolean linkAdvance = false;
				int sentMsgSize = 0;
				Exception sendException = null;

				try
				{
					delivery = sendLinkCurrent.delivery(deliveryTag.getDeliveryTag().getBytes());
					delivery.setMessageFormat(sendData.getMessageFormat());

					TransactionContext transaction = sendData.getTransaction();
					if (transaction != TransactionContext.NULL_TXN) {
						TransactionalState transactionalState = new TransactionalState();
						transactionalState.setTxnId(new Binary(transaction.getTransactionId().array()));
						delivery.disposition(transactionalState);
					}

					TRACE_LOGGER.debug("Sending message delivery '{}' to '{}'", deliveryTag.getDeliveryTag(), this.sendPath);
					sentMsgSize = sendLinkCurrent.send(sendData.getMessage(), 0, sendData.getEncodedMessageSize());
					assert sentMsgSize == sendData.getEncodedMessageSize() : "Contract of the ProtonJ library for Sender.Send API changed";

					linkAdvance = sendLinkCurrent.advance();
				}
				catch(Exception exception)
				{
					sendException = exception;
				}

				if (linkAdvance)
				{
					this.linkCredit--;
					sendData.setWaitingForAck();
				}
				else
				{
					TRACE_LOGGER.warn("Sendlink advance failed. path:{}, linkName:{}, deliveryTag:{}, sentMessageSize:{}, payloadActualSiz:{}",
							this.sendPath, this.sendLink.getName(), deliveryTag, sentMsgSize, sendData.getEncodedMessageSize());

					if (delivery != null)
					{
						delivery.free();
					}

					Exception completionException = sendException != null ? new OperationCancelledException("Send operation failed. Please see cause for more details", sendException)
							: new OperationCancelledException(String.format(Locale.US, "Send operation failed while advancing delivery(tag: %s) on SendLink(path: %s).", this.sendPath, deliveryTag));
					AsyncUtil.completeFutureExceptionally(sendData.getWork(), completionException);
				}
			}
		}
		finally
		{
			synchronized (this.pendingSendLock)
			{
				if(this.isSendLoopRunning)
				{
					this.isSendLoopRunning = false;
				}
			}
		}
	}
14420
def start
    @em_lock.synchronize{
      # TODO on event of the process ending this thread will be
      # shutdown before a local finalizer can be run,
      # would be good to gracefully shut this down / wait for completion
      @reactor_thread  = Thread.new {
        begin
          EventMachine.run
        rescue Exception => e
          # TODO option to autorestart the reactor on errors ?
          puts "Critical exception #{e}\n#{e.backtrace.join("\n")}"
        ensure
          @em_lock.synchronize { @reactor_thread = nil }
        end
      } unless @reactor_thread
    }
    sleep 0.01 until EventMachine.reactor_running? # XXX hack but needed
    self
  end
1046243
async def send(self, sender, **kwargs):
        '''  '''
        if not self.receivers:
            return []
        responses = []
        futures = []
        for receiver in self._get_receivers(sender):
            method = receiver()
            if callable(method):
                futures.append(method(sender=sender, **kwargs))
        if len(futures) > 0:
            responses = await asyncio.gather(*futures)
        return responses
132065
protected void sendInfoPacket(String channel) {
		Tree msg = registry.getDescriptor();
		msg.put("ver", PROTOCOL_VERSION);
		msg.put("sender", nodeID);
		msg.put("seq", registry.getTimestamp());
		publish(channel, msg);
	}
519599
def send(self, timeout=None):
        """"""
        if self.sigint_event and is_main_thread():
            with ReplacedSigIntHandler(self.sigint_handler):
                return self._send(timeout)
        else:
            return self._send(timeout)
1167843
@Override
	public void send(final CEMI frame, final BlockingMode mode) throws KNXConnectionClosedException
	{
		if (frame.getMessageCode() != CEMILData.MC_LDATA_IND)
			throw new KNXIllegalArgumentException("cEMI frame is not an L-Data.ind");
		try {
			if (loopbackEnabled) {
				synchronized (loopbackFrames) {
					loopbackFrames.add((CEMILData) frame);
				}
				logger.trace("add to multicast loopback frame buffer: {}", frame);
			}
			// filter IP system broadcasts and always send them unsecured
			if (RoutingSystemBroadcast.isSystemBroadcast(frame)) {
				final byte[] buf = PacketHelper.toPacket(new RoutingSystemBroadcast(frame));
				final DatagramPacket p = new DatagramPacket(buf, buf.length, systemBroadcast, DEFAULT_PORT);
				if (sysBcastSocket != null)
					sysBcastSocket.send(p);
				else
					socket.send(p);
			}
			else
				super.send(frame, NonBlocking);
			// we always succeed...
			setState(OK);
		}
		catch (final IOException e) {
			close(CloseEvent.INTERNAL, "communication failure", LogLevel.ERROR, e);
			throw new KNXConnectionClosedException("connection closed (" + e.getMessage() + ")");
		}
		catch (final KNXTimeoutException ignore) {}
		catch (final InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}
58550
function send () {
      sent = true;
      // next tick it to avoid express catching runtime error in the callback
      //   getting caught by express and triggering app's error handler
      process.nextTick(function () {
        res.emit('finish');
        cb(null, res, body);
      });
    }
990795
function sendCommand(method, url, payload, skipQueue) {
    if (!self.connected) {
      debug('Queueing %s %s', method, url);
      return new P(function(resolve, reject) {
        var p = P.method(function() {
          debug('Dequeueing %s %s', method, url);
          return _sendCommand(method, url, payload)
            .then(resolve)
            .catch(reject);
        });
        if (skipQueue) {
          pendingCommands.splice(0, 0, p);
        } else {
          pendingCommands.push(p);
        }
      });
    }
    return _sendCommand(method, url, payload);
  }
40334
public static Object invokeNewN(Class senderClass, Class receiver, Object arguments) throws Throwable {
        try {
            return InvokerHelper.invokeConstructorOf(receiver, arguments);
        } catch (GroovyRuntimeException gre) {
            throw unwrap(gre);
        }
    }
629720
public final void send (final ProtocolDataUnit protocolDataUnit) throws InternetSCSIException {

        try {
            senderReceiver.sendOverWire(protocolDataUnit);
        } catch (IOException e) {
            throw new InternetSCSIException(e);
        } catch (InterruptedException e) {
            throw new InternetSCSIException(e);
        }
    }
506359
private void send(final byte[] buffer, final Session destination) {
        synchronized (connections) {
            // execute asynchronously to avoid slower clients from interfering with faster clients
            final ExecutorService connectionThread = connectionThreads.get(destination);
            if (connectionThread == null) {
                // Maybe the client has disconnected in the mean time.
                return;
            }
            connectionThread.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        // for the case that the runnable was committed shortly before the connection was closed.
                        if (destination.isOpen()) {
                            // FIXME replace with getAsyncRemote and removeconnectionThreads as soon as
                            // getAsyncRemote on tomcat is thread-safe
                            destination.getBasicRemote().sendBinary(ByteBuffer.wrap(buffer));
                        }
                    } catch (final IOException e) {
                        try {
                            if (destination.isOpen()) {
                                destination.close(new CloseReason(CloseCodes.PROTOCOL_ERROR, "Failed to send data."));
                            }
                        } catch (final IOException e1) {
                            // The outer exception already indicated that something went wrong.
                            ignore(e1);
                        }
                        handleClientError(destination, e);
                        connectionCloses(destination);
                    }
                }
            });
        }
    }
334288
private void doSend(byte[] msg, boolean wait, KNXAddress dst)
		throws KNXAckTimeoutException, KNXLinkClosedException
	{
		if (closed)
			throw new KNXLinkClosedException("link closed");
		try {
			logger.info("send message to " + dst + (wait ? ", wait for ack" : ""));
			logger.trace("EMI " + DataUnitBuilder.toHex(msg, " "));
			conn.send(msg, wait);
			logger.trace("send to " + dst + " succeeded");
		}
		catch (final KNXPortClosedException e) {
			logger.error("send error, closing link", e);
			close();
			throw new KNXLinkClosedException("link closed, " + e.getMessage());
		}
	}
791516
public void send(CEMILData msg, boolean waitForCon) throws KNXTimeoutException,
		KNXLinkClosedException
	{
		doSend(createEMI(msg), waitForCon, msg.getDestination());
	}
737189
@Override
    public void onReceive(Object message) throws Exception {
        try {
            if (message instanceof RequestWorkerMsgType) {
                switch ((RequestWorkerMsgType) message) {
                case PROCESS_REQUEST:
                    tryCount++;

                    if (tryCount == 1) {
                        sender = getSender();

                        ConnectionlessBootstrap udpClient = bootStrapUdpClient();

                        // Start the connection attempt.
                        ChannelFuture future = udpClient
                                .connect(new InetSocketAddress(targetHost,
                                        udpMeta.getUdpPort()));

                        // first schedule timeout before the sync wait
                        timeoutDuration = Duration.create(
                                actorMaxOperationTimeoutSec, TimeUnit.SECONDS);

                        timeoutMessageCancellable = getContext()
                                .system()
                                .scheduler()
                                .scheduleOnce(
                                        timeoutDuration,
                                        getSelf(),
                                        RequestWorkerMsgType.PROCESS_ON_TIMEOUT,
                                        getContext().system().dispatcher(),
                                        getSelf());

                        // Wait until the connection attempt succeeds or fails.
                        channel = future.awaitUninterruptibly().getChannel();
                        ChannelFuture requestFuture = null;

                        // Sends the line to server need. line ending
                        requestFuture = channel.write(udpMeta.getCommand()
                                + "\r\n");

                        // Wait until all messages are flushed before closing
                        // the channel.
                        if (requestFuture != null) {
                            requestFuture.await();
                        }
                    } else {
                        getLogger().error(
                                "duplicated PROCESS_REQUEST msg. ignore...");
                    }
                    break;

                case CANCEL:
                    if (sender == null)
                        sender = getSender();
                    getLogger().info(
                            "UDP Request was CANCELLED.................{}",
                            targetHost);
                    cancelCancellable();
                    reply(null, true, PcConstants.REQUEST_CANCELED,
                            PcConstants.REQUEST_CANCELED, PcConstants.NA,
                            PcConstants.NA_INT);
                    break;

                case PROCESS_ON_EXCEPTION:

                    String errorSummary = PcErrorMsgUtils.replaceErrorMsg(cause
                            .toString());
                    String stackTrace = PcStringUtils.printStackTrace(cause);
                    cancelCancellable();
                    reply(null, true, errorSummary, stackTrace, PcConstants.NA,
                            PcConstants.NA_INT);

                    break;

                case PROCESS_ON_TIMEOUT:

                    getLogger().error("PROCESS_ON_TIMEOUT.................{}",
                            targetHost);
                    cancelCancellable();

                    String errorMsg = String
                            .format("UdpWorker Timedout after %d SEC (no response but no exception catched). Check URL: may be very slow or stuck.",
                                    actorMaxOperationTimeoutSec);

                    reply(null, true, errorMsg, errorMsg, PcConstants.NA,
                            PcConstants.NA_INT);
                    break;

                case CHECK_FUTURE_STATE:
                default:
                    sender = getSender();
                    this.cause = new ActorMessageTypeInvalidException(
                            "ActorMessageTypeInvalidException error for on "
                                    + this.targetHost);
                    getSelf().tell(RequestWorkerMsgType.PROCESS_ON_EXCEPTION,
                            getSelf());
                    break;
                }
            } else {
                unhandled(message);
                sender = getSender();
                this.cause = new ActorMessageTypeInvalidException(
                        "ActorMessageTypeInvalidException error for UDP on "
                                + this.targetHost);
                getSelf().tell(RequestWorkerMsgType.PROCESS_ON_EXCEPTION,
                        getSelf());
            }
        } catch (Exception e) {
            this.cause = e;
            getSelf()
                    .tell(RequestWorkerMsgType.PROCESS_ON_EXCEPTION, getSelf());
        }
    }
527393
def send(self, url, data, headers):
        
        eventlet.spawn(self._send_payload, (url, data, headers))
255731
def on_send(node)
      name     = node.children[1].to_s
      name     = SEND_MAPPING.fetch(name, name)
      callback = "on_send_#{name}"

      value_stack.add_stack

      execute_callback(callback, node)
    end
1340751
public void sendReliably() throws Rel100Exception {
		final int statusCode = getStatus();
		if(statusCode == 100 || statusCode >= 200) {
			throw new Rel100Exception(Rel100Exception.NOT_1XX);
		}
		if(!Request.INVITE.equals(originalRequest.getMethod())) {
			throw new Rel100Exception(Rel100Exception.NOT_INVITE);
		}		
		if(!containsRel100(originalRequest.getMessage())) {
			throw new Rel100Exception(Rel100Exception.NO_REQ_SUPPORT);
		}
		try {
			send(true);
		} catch (IOException e) {
			// sendReliably doesn't throw IOException where send() does...
			throw new IllegalStateException(e);
		}
	}