Wraps the JFAP Channel send method to allow tracing, retrieval
of Unique request numbers and setting of message priority.

@param sendSegType
@param priority
@param canPoolOnReceive

@throws SIConnectionDroppedException
@throws SIConnectionLostException
178368
protected void jfapSend(CommsByteBuffer buffer,
                           int sendSegType,
                           int priority,
                           boolean canPoolOnReceive,
                           ThrottlingPolicy throttlingPolicy)
      throws SIConnectionDroppedException,
             SIConnectionLostException
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "jfapSend",
                                           new Object[]{buffer, sendSegType, priority, canPoolOnReceive});

      if (buffer == null)
      {
         // The data list cannot be null
         SIErrorException e = new SIErrorException(
            nls.getFormattedMessage("NULL_DATA_LIST_PASSED_IN_SICO1046", null, null)
         );

         FFDCFilter.processException(e, CLASS_NAME + ".JFAPSend",
                                     CommsConstants.JFAPCOMMUNICATOR_SEND_01, this);

         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(tc, e.getMessage(), e);

         throw e;
      }

      int reqNum = 0;

      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
      {
         SibTr.debug(this, tc, "About to Send segment " 
                         + "conversation: "+ con + " " 
                         + JFapChannelConstants.getSegmentName(sendSegType)
                         + " - " + sendSegType 
                         + " (0x" + Integer.toHexString(sendSegType) + ") "
                         + "using request number "
                         + reqNum);
         
         SibTr.debug(this, tc, con.getFullSummary());
      }

      con.send(buffer, sendSegType, reqNum, priority, canPoolOnReceive, throttlingPolicy, null);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "jfapSend");
   }
573036
public final void sendMessage(ByteBuffer bb, byte msg_priority) {
    UDP_TCP_SendThread sendThread = _sendThread;
    if (sendThread == null) {
      // Sending threads are created lazily.
      // This is because we will intern all client nodes including the ones that have nothing to do with the cluster.
      // By delaying the initialization to the point when we actually want to send a message, we initialize the sending
      // thread just for the nodes that are really part of the cluster.
      // The other reason is client disconnect - when removing client we kill the reference to the sending thread, if we
      // still do need to communicate with the client later - we just recreate the thread.
      if (_removed_from_cloud) {
        Log.warn("Node " + this + " is not active in the cloud anymore but we want to communicate with it." +
                "Re-opening the communication channel.");
      }
      sendThread = startSendThread();
    }
    assert sendThread != null;
    sendThread.sendMessage(bb, msg_priority);
  }
633548
@Override
	public void send() throws IOException {
		if(logger.isDebugEnabled()) {
			logger.debug("send - method=" + this.getMethod());
		}
		checkReadOnly();
		// Cope with com.bea.sipservlet.tck.agents.api.javax_servlet_sip.SipServletMessageTest.testSend101
		// make sure a message received cannot be sent out
		checkMessageState();
		final Request request = (Request) super.message;
		final String requestMethod = getMethod();
		final SipApplicationDispatcher sipApplicationDispatcher = sipFactoryImpl.getSipApplicationDispatcher();
		final MobicentsSipSession session = getSipSession();
		final DNSServerLocator dnsServerLocator = sipApplicationDispatcher.getDNSServerLocator();
		Hop hop = null;
		// RFC 3263 support
		if(dnsServerLocator != null) {
			if(Request.CANCEL.equals(requestMethod)) {
				// RFC 3263 Section 4 : a CANCEL for a particular SIP request MUST be sent to the same SIP
				// server that the SIP request was delivered to.
				TransactionApplicationData inviteTxAppData = ((TransactionApplicationData)inviteTransactionToCancel.getApplicationData());
				if(inviteTxAppData != null && inviteTxAppData.getHops() != null) {
					hop = inviteTxAppData.getHops().peek();
				}
			} else {
				javax.sip.address.URI uriToResolve =  request.getRequestURI();
				RouteHeader routeHeader = (RouteHeader) request.getHeader(RouteHeader.NAME);
				if(routeHeader != null) {
					uriToResolve = routeHeader.getAddress().getURI();
				} else {
					// RFC5626 - see if we are to find a flow for this request.
					// Note: we should do this even if the "uriToResolve" is coming
					// from a route header but since we currently have not implemented
					// the correct things for a proxy scenario, only do it for UAS
					// scenarios. At least this will minimize the potential for messing
					// up right now...
					uriToResolve = resolveSipOutbound(uriToResolve);
				}
				String uriToResolveTransport = ((javax.sip.address.SipURI)uriToResolve).getTransportParam();
				boolean transportParamModified = false;
				if(session.getProxy() == null && session.getTransport() != null && uriToResolve.isSipURI() && uriToResolveTransport  == null &&
						// no need to modify the Request URI for UDP which is the default transport
						!session.getTransport().equalsIgnoreCase(ListeningPoint.UDP)) {
					try {
						((javax.sip.address.SipURI)uriToResolve).setTransportParam(session.getTransport());
						transportParamModified = true;
					} catch (ParseException e) {
						// nothing to do here, will never happen
					}
				}
				Queue<Hop> hops = dnsServerLocator.locateHops(uriToResolve);
				if(transportParamModified) {
					// Issue http://code.google.com/p/sipservlets/issues/detail?id=186
					// Resetting the transport to what is was before the modification to avoid modifying the route set
					if(uriToResolveTransport == null) {
						((javax.sip.address.SipURI)uriToResolve).removeParameter("transport");
					} else {
						try {
							((javax.sip.address.SipURI)uriToResolve).setTransportParam(uriToResolveTransport);
						} catch (ParseException e) {
							// nothing to do here, will never happen
						}
					}
				}
				if(hops != null && hops.size() > 0) {
					// RFC 3263 support don't remove the current hop, it will be the one to reuse for CANCEL and ACK to non 2xx transactions
					hop = hops.peek();
					transactionApplicationData.setHops(hops);
				}
			}
		}
		if(logger.isDebugEnabled()) {
			logger.debug("send - calling send(hop) - hop=" + hop);
		}
		send(hop);
	}
1273935
@Override
    public void sendAckMessage(SIBUuid8 sourceMEUuid,
                               SIBUuid12 destUuid,
                               SIBUuid8 busUuid,
                               long ackPrefix,
                               int priority,
                               Reliability reliability,
                               SIBUuid12 stream,
                               boolean consolidate) throws SIResourceException
    {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "sendAckMessage", new Long(ackPrefix));

        ControlAck ackMsg = createControlAckMessage();

        // As we are using the Guaranteed Header - set all the attributes as
        // well as the ones we want.
        SIMPUtils.setGuaranteedDeliveryProperties(ackMsg,
                                                  _messageProcessor.getMessagingEngineUuid(),
                                                  sourceMEUuid,
                                                  stream,
                                                  null,
                                                  destUuid,
                                                  ProtocolType.UNICASTOUTPUT,
                                                  GDConfig.PROTOCOL_VERSION);

        ackMsg.setPriority(priority);
        ackMsg.setReliability(reliability);

        ackMsg.setAckPrefix(ackPrefix);

        // Send Ack messages at the priority of the original message +1
        sendToME(ackMsg, sourceMEUuid, busUuid, priority + 1);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendAckMessage");
    }
502158
@Override
	public final void send(Message message, int deliveryMode, int priority, long timeToLive) throws JMSException
    {
    	if (this.destination == null)
    		throw new UnsupportedOperationException("Destination was not set at creation time");

        // Setup message fields
    	setupMessage(destination,message,deliveryMode,priority,timeToLive);
    	
    	// Handle foreign message implementations
        message = MessageTools.normalize(message);
        
    	sendToDestination(destination,false,message,deliveryMode,priority,timeToLive);
    }
1012196
@Override
	protected final void sendToDestination(Destination destination, boolean destinationOverride, Message srcMessage, int deliveryMode, int priority, long timeToLive) throws JMSException
    {
        // Check that the destination was specified
        if (destination == null)
            throw new InvalidDestinationException("Destination not specified");  // [JMS SPEC]

        // Create an internal copy if necessary
        AbstractMessage message = MessageTools.makeInternalCopy(srcMessage);
        
        externalAccessLock.readLock().lock();
        try
		{
    		checkNotClosed();
    		
	        // Dispatch to session
            ((LocalSession)session).dispatch(message);
		}
        finally
        {
        	externalAccessLock.readLock().unlock();
        }
    }
1042908
@Override
	protected final void sendToDestination(Destination destination, boolean destinationOverride, Message srcMessage, int deliveryMode, int priority, long timeToLive) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	    	checkNotClosed();
	    	
	    	((RemoteSession)session).dispatch(srcMessage);
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
25632
private void send(final byte[] buffer, final Object destination) {
        if (LOG.isTraceEnabled()) {
            LOG.trace("Sending from thread: id: " + Thread.currentThread().getName() + ", name: "
                    + Thread.currentThread().getName());
        }

        final WsOutbound outbound = ((MessageInbound) destination).getWsOutbound();

        final ExecutorService executorService = connectionThreads.get(destination);
        // execute asynchronously to avoid slower clients from interfering with faster clients
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    outbound.writeBinaryMessage(ByteBuffer.wrap(buffer));
                } catch (final IOException e) {
                    LOG.warn("Sending data to a client failed. Closing connection to this client.");
                    try {
                        outbound.close(1002, null);
                        // CHECKSTYLE:OFF
                    } catch (final IOException e1) {
                        // Maybe the connection is already closed. This is no exceptional state but rather the
                        // default in
                        // this case. So it's safe to ignore this exception.
                    }
                    // CHECKSTYLE:ON
                    connectionCloses((SynchronizeFXTomcatConnection) destination);
                }
            }
        });
    }
1053743
void send(final InterfaceFeature feature) throws KNXConnectionClosedException, KNXAckTimeoutException, InterruptedException {
		synchronized (lock) {
			final TunnelingFeature get = TunnelingFeature.newGet(channelId, getSeqSend(), feature);
			send(get);
		}
	}
464739
void send(final long destinations[], final VoltMessage message) {
        if (!m_isUp) {
            hostLog.warn("Failed to send VoltMessage because connection to host " +
                    CoreUtils.getHostIdFromHSId(destinations[0])+ " is closed");
            return;
        }
        if (destinations.length == 0) {
            return;
        }

        // if this link is "gone silent" for partition tests, just drop the message on the floor
        if (!m_linkCutForTest.get()) {
            m_network.enqueue(
                    new DeferredSerialization() {
                        @Override
                        public final void serialize(final ByteBuffer buf) throws IOException {
                            buf.putInt(buf.capacity() - 4);
                            buf.putLong(message.m_sourceHSId);
                            buf.putInt(destinations.length);
                            for (int ii = 0; ii < destinations.length; ii++) {
                                buf.putLong(destinations[ii]);
                            }
                            message.flattenToBuffer(buf);
                            buf.flip();
                        }

                        @Override
                        public final void cancel() {
                        /*
                         * Can this be removed?
                         */
                        }

                        @Override
                        public String toString() {
                            return message.getClass().getName();
                        }

                        @Override
                        public int getSerializedSize() {
                            final int len = 4            /* length prefix */
                                    + 8            /* source hsid */
                                    + 4            /* destinationCount */
                                    + 8 * destinations.length  /* destination list */
                                    + message.getSerializedSize();
                            return len;
                        }
                    });
        }

        long current_time = EstTime.currentTimeMillis();
        long current_delta = current_time - m_lastMessageMillis.get();
        /*
         * Try and give some warning when a connection is timing out.
         * Allows you to observe the liveness of the host receiving the heartbeats
         */
        if (isPrimary() && current_delta > m_logRate) {
            rateLimitedLogger.log(
                    "Have not received a message from host "
                        + hostnameAndIPAndPort() + " for " + (current_delta / 1000.0) + " seconds",
                        current_time);
        }
        // NodeFailureFault no longer immediately trips FHInputHandler to
        // set m_isUp to false, so use both that and m_closing to
        // avoid repeat reports of a single node failure
        if ((!m_closing && m_isUp) &&
                isPrimary() &&
                current_delta > m_deadHostTimeout)
        {
            if (m_deadReportsCount.getAndIncrement() == 0) {
                hostLog.error("DEAD HOST DETECTED, hostname: " + hostnameAndIPAndPort());
                hostLog.info("\tcurrent time: " + current_time);
                hostLog.info("\tlast message: " + m_lastMessageMillis);
                hostLog.info("\tdelta (millis): " + current_delta);
                hostLog.info("\ttimeout value (millis): " + m_deadHostTimeout);
                VoltDB.dropStackTrace("Timed out foreign host " + hostnameAndIPAndPort() + " with delta " + current_delta);
            }
            m_hostMessenger.reportForeignHostFailed(m_hostId);
        }
    }
1155698
@Override
    public void sendFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID) throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "sendFlushedMessage", new Object[] { streamID });

        // This flush should be broadcast to all downstream neighbors 
        // for this cell as it is the result of a startFlush()

        // This is a bit of a kludge since we may be sending flushes to cells
        // which have no information about a stream.
        HashMap allPubSubOutputHandlers = _destination.getAllPubSubOutputHandlers();
        try
        {
            Iterator itr = allPubSubOutputHandlers.values().iterator();

            while (itr.hasNext())
            {
                // Get the appropriate target cellule and forward the message
                PubSubOutputHandler handler = (PubSubOutputHandler) itr.next();

                // Note that the null Cellule we pass in here is ignored
                // as each OutputHandler knows its targetCellule 
                handler.sendFlushedMessage(null, streamID);

                // Also, tell the handler to remove any information for this
                // stream since it's just been flushed.
                handler.removeStream(streamID);
            }
        } finally
        {
            // By calling the getAllPubSubOutputHandlers it will lock the 
            // handlers
            _destination.unlockPubsubOutputHandlers();
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendFlushedMessage");
    }
1279702
public int send (Connection connection, Object object, SocketAddress address) throws IOException {
		DatagramChannel datagramChannel = this.datagramChannel;
		if (datagramChannel == null) throw new SocketException("Connection is closed.");
		synchronized (writeLock) {
			try {
				try {
					serialization.write(connection, writeBuffer, object);
				} catch (Exception ex) {
					throw new KryoNetException("Error serializing object of type: " + object.getClass().getName(), ex);
				}
				writeBuffer.flip();
				int length = writeBuffer.limit();
				datagramChannel.send(writeBuffer, address);

				lastCommunicationTime = System.currentTimeMillis();

				boolean wasFullWrite = !writeBuffer.hasRemaining();
				return wasFullWrite ? length : -1;
			} finally {
				writeBuffer.clear();
			}
		}
	}
1267
public void sendAckMessage(
    SIBUuid8 meUuid,
    SIBUuid12 destUuid,
    SIBUuid8  busUuid,  
    long ackPrefix,
    int priority,
    Reliability reliability,
    SIBUuid12 streamID,
    boolean consolidate)
    throws SIResourceException
  {  
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "sendAckMessage",
        new Object[] { meUuid, 
                       new Long(ackPrefix),
                       new Integer(priority),
                       reliability,
                       streamID });

    ControlAck newAckMsg =
      createControlAckMessage(
        priority,
        reliability,
        streamID);
    
    newAckMsg.setAckPrefix(ackPrefix);        
    
    try
    {
      _parentInputHandler.handleControlMessage(null, newAckMsg);
    }
    catch (SIException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendAckMessage",
        "1:1731:1.164.1.5",
        this);
        
      SibTr.exception(tc, e);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      {
        SibTr.exception(tc, e);
        SibTr.exit(tc, "sendAckMessage", e);
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "sendAckMessage");
  }
196167
public void send(PeerMessage message) throws IllegalStateException {
    logger.trace("Sending msg {} to {}", message.getType(), this);
    if (this.isConnected()) {
      ByteBuffer data = message.getData();
      data.rewind();
      connectionManager.offerWrite(new WriteTask(socketChannel, data, new WriteListener() {
        @Override
        public void onWriteFailed(String message, Throwable e) {
          if (e == null) {
            logger.info(message);
          } else if (e instanceof ConnectionClosedException){
            logger.debug(message, e);
            unbind(true);
          } else {
            LoggerUtils.warnAndDebugDetails(logger, message, e);
          }

        }

        @Override
        public void onWriteDone() {
        }
      }), 1, TimeUnit.SECONDS);
    } else {
      logger.trace("Attempting to send a message to non-connected peer {}!", this);
      unbind(true);
    }
  }
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);
                    }
                }
            });
        }
    }
70852
static void rcvCloseProducerSess(CommsByteBuffer request, Conversation conversation,
                                     int requestNumber, boolean allocatedFromBufferPool,
                                     boolean partOfExchange)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "rcvCloseProducerSess",
                        new Object[]
                        {
                         request,
                         conversation,
                         "" + requestNumber,
                         "" + allocatedFromBufferPool,
                         "" + partOfExchange
                        });
        ConversationState convState = (ConversationState) conversation.getAttachment();

        short connectionObjectID = request.getShort(); // BIT16 ConnectionObjectId
        short producerObjectID = request.getShort(); // BIT16 SyncProducerSessionId

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        {
            SibTr.debug(tc, "connectionObjectID", connectionObjectID);
            SibTr.debug(tc, "producerObjectID", producerObjectID);
        }

        ProducerSession producerSession = ((ProducerSession) convState.getObject(producerObjectID));

        try
        {
            producerSession.close();
            convState.removeObject(producerObjectID);

            try
            {
                conversation.send(poolManager.allocate(),
                                  JFapChannelConstants.SEG_CLOSE_PRODUCER_SESS_R,
                                  requestNumber,
                                  JFapChannelConstants.PRIORITY_MEDIUM,
                                  true,
                                  ThrottlingPolicy.BLOCK_THREAD,
                                  null);
            } catch (SIException e)
            {
                FFDCFilter.processException(e,
                                            CLASS_NAME + ".rcvCloseProducerSess",
                                            CommsConstants.STATICCATPRODUCER_CLOSE_01);

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, e.getMessage(), e);

                SibTr.error(tc, "COMMUNICATION_ERROR_SICO2024", e);
            }
        } catch (SIException e)
        {
            //No FFDC code needed - processor will have already FFDC'ed any interesting ones....
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc, e.getMessage(), e);

            StaticCATHelper.sendExceptionToClient(e,
                                                  CommsConstants.STATICCATPRODUCER_CLOSE_02, // d186970
                                                  conversation, requestNumber); // f172297

        }

        request.release(allocatedFromBufferPool);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "rcvCloseProducerSess");
    }
1039202
private void sendBuffer(final HttpResponseStatus status,
                          final ChannelBuffer buf) {
    final String contentType = (api_version < 1 ? guessMimeType(buf) :
      serializer.responseContentType());
    sendBuffer(status, buf, contentType);
  }