/* (non-Javadoc)
@see com.ibm.ws.sib.processor.impl.interfaces.UpstreamControl#sendAckMessage(com.ibm.ws.sib.trm.topology.Cellule, long, int, com.ibm.websphere.sib.Reliability, com.ibm.ws.sib.utils.SIBUuid12)
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");
  }
666534
@Override
	public final void send(Message message) throws JMSException
    {
    	if (this.destination == null)
    		throw new UnsupportedOperationException("Destination was not set at creation time");
    	
    	// Setup message fields
    	setupMessage(destination,message,defaultDeliveryMode,defaultPriority,defaultTimeToLive);
    	
    	// Handle foreign message implementations
        message = MessageTools.normalize(message);

        sendToDestination(destination,false,message,defaultDeliveryMode,defaultPriority,defaultTimeToLive);
    }
1215594
protected synchronized void sendCommandToShard(VoidMessage message) {
        // if this node is shard - we just step over TCP/IP infrastructure
        // TODO: we want LocalTransport to be used in such cases
        if (nodeRole == NodeRole.SHARD) {
            message.setTargetId(shardIndex);
            messages.add(message);
            return;
        }


        //log.info("Sending CS: {}", message.getClass().getCanonicalName());

        message.setTargetId(targetIndex);
        DirectBuffer buffer = message.asUnsafeBuffer();

        long result = publicationForShards.offer(buffer);

        if (result < 0)
            for (int i = 0; i < 5 && result < 0; i++) {
                try {
                    // TODO: make this configurable
                    Thread.sleep(1000);
                } catch (Exception e) {
                }
                result = publicationForShards.offer(buffer);
            }

        // TODO: handle retransmit & backpressure separately

        if (result < 0)
            throw new RuntimeException("Unable to send message over the wire. Error code: " + result);
    }
170461
public void sendAreYouFlushedMessage(
    SIBUuid8 meUuid, 
    SIBUuid12 destUuid,
    SIBUuid8  busUuid,  
    long queryID, 
    SIBUuid12 streamID)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "sendAreYouFlushedMessage", 
        new Object[]{meUuid, new Long(queryID), streamID});
        
    InvalidOperationException e = 
      new InvalidOperationException(nls.getFormattedMessage(
        "INTERNAL_MESSAGING_ERROR_CWSIP0001",
        new Object[] {
          "com.ibm.ws.sib.processor.impl.PubSubOutputHandler",
          "1:1766:1.164.1.5" },
        null));
    
    // FFDC
    FFDCFilter.processException(
      e,
      "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendAreYouFlushedMessage",
      "1:1773:1.164.1.5",
      this);
      
    SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
      new Object[] {
        "com.ibm.ws.sib.processor.impl.PubSubOutputHandler",
        "1:1779:1.164.1.5" });
      
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "sendAreYouFlushedMessage", e);
    throw e;  
  }
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();
    	}
    }
758958
private void sendMessage(BeanMessageID type, Buffer payload) {
        Buffer buffer = new Buffer();
        buffer.writeByte((type.getRawValue() >> 8) & 0xff);
        buffer.writeByte(type.getRawValue() & 0xff);
        if (payload != null) {
            try {
                buffer.writeAll(payload);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        GattSerialMessage serialMessage = GattSerialMessage.fromPayload(buffer.readByteArray());
        gattClient.getSerialProfile().sendMessage(serialMessage.getBuffer());
    }
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");
    }
594201
public void sendMessage(String message) {
        EventBuilder eventBuilder = new EventBuilder().withMessage(message)
            .withLevel(Event.Level.INFO);
        sendEvent(eventBuilder);
    }
492962
@Override
    protected void sendMessage(@NonNull INDArray message, int iterationNumber, int epochNumber) {
        // here we'll send our stuff to other executores over the wire
        // and let's pray for udp broadcast availability

        // Send this message away
        // FIXME: do something with unsafe duplication, which is bad and used ONLY for local spark
        try (MemoryWorkspace wsO = Nd4j.getMemoryManager().scopeOutOfWorkspaces()) {
            long updateId = updatesCounter.getAndIncrement();

            val m = message.unsafeDuplication();
            ModelParameterServer.getInstance().sendUpdate(m, iterationNumber, epochNumber);
        }


        // heere we update local queue
        super.sendMessage(message, iterationNumber, epochNumber);
    }
594257
protected void sendMessage(DataOutputStream outputStream, Message message) throws IOException {
        long startNs = System.nanoTime();
        ProtoUtils.writeMessage(outputStream, message);
        if(streamStats != null) {
            streamStats.reportNetworkTime(operation,
                                          Utils.elapsedTimeNs(startNs, System.nanoTime()));
        }
    }
183016
private void sendLayoutMessage(final String msg) {
      if (sqw != null) {
          String packet = msg;
          String hdr = getPacketHeader(new Date().getTime());
          if(facilityPrinting || hdr.length() > 0) {
              StringBuffer buf = new StringBuffer(hdr);
              if(facilityPrinting) {
                  buf.append(facilityStr);
              }
              buf.append(msg);
              packet = buf.toString();
          }
          sqw.setLevel(6);
          sqw.write(packet);
      }
  }
1328785
public synchronized long sendMessage(AbstractRequestMessage message) {
    log.debug("Sending {} to {}", message, this);
    message.setTicketNumber(this.nextTicketNumber.getAndIncrement());
    this.outstandingRequests.put(Long.valueOf(message.getTicketNumber()),
        message);
    this.session.write(message);
    return message.getTicketNumber();
  }
687871
public void sendMessage(Message message)
    {
        ((DualMessageQueue)this.getMessageQueue()).getMessageStack().sendMessage(message);
        
        ((BaseMessage)message).setProcessedByClient(true);     // Make sure any server doesn't send this back up.

        super.sendMessage(message);
    }
1164890
@Override
    public <T> void send(NotificationType type, List<T> messages) throws NotificationException {
        String[] strMessages = new String[messages.size()];
        for (int index = 0; index < messages.size(); index++) {
            strMessages[index] = getMessageJson(messages.get(index));
        }
        sendInternal(type, strMessages);
    }
282640
public void sendMessage(String requestMessage, String queueName, String correlationId,
            final Queue replyQueue, int delaySeconds, int deliveryMode) throws JMSException {
        /**
         * If it's an internal message then always use jmsTemplate for ActiveMQ
         */

        if (jmsTemplate != null) {
            jmsTemplate.setDeliveryMode(deliveryMode);
            if (!StringUtils.isEmpty(queueName)) {
                jmsTemplate.send(queueName, new MDWMessageCreator(requestMessage, correlationId,
                        replyQueue, delaySeconds));

            }
            else {
                jmsTemplate.send(new MDWMessageCreator(requestMessage, correlationId, replyQueue));
            }
        }
    }
1365668
private void sendMessage(Crouton crouton, final int messageId) {
    final Message message = obtainMessage(messageId);
    message.obj = crouton;
    sendMessage(message);
  }
1216467
public void sendMessage(Message message) throws RemoteException
    {
        BaseTransport transport = this.createProxyTransport(SEND_MESSAGE);
        transport.addParam(MESSAGE, message);   // Don't use COMMAND
        transport.sendMessageAndGetReply();
    }