/* (non-Javadoc)
@see com.ibm.ws.sib.processor.impl.interfaces.UpstreamControl#sendAreYouFlushedMessage(com.ibm.ws.sib.trm.topology.Cellule, long, com.ibm.ws.sib.utils.SIBUuid12)
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;  
  }
198053
protected void sendDatagram (UpstreamMessage msg)
        throws IOException
    {
        // reset the stream and write our connection id and hash placeholder
        _bout.reset();
        _uout.writeInt(_client.getConnectionId());
        _uout.writeLong(0L);

        // write the datagram through the sequencer
        _sequencer.writeDatagram(msg);

        // flip the buffer and make sure it's not too long
        ByteBuffer buf = _bout.flip();
        int size = buf.remaining();
        if (size > Client.MAX_DATAGRAM_SIZE) {
            log.warning("Dropping oversized datagram", "size", size, "msg", msg);
            return;
        }

        // compute the hash
        buf.position(12);
        _digest.update(buf);
        byte[] hash = _digest.digest(_secret);

        // insert the first 64 bits of the hash
        buf.position(4);
        buf.put(hash, 0, 8).rewind();

        // send the datagram
        writeDatagram(buf);

        // notify the tracker
        _client.getMessageTracker().messageSent(true, size, msg);
    }
123624
public void sendFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
       SibTr.entry(tc, "sendFlushedMessage", new Object[]{ignore, streamID});
       
    ControlFlushed flushMsg = createControlFlushed(_targetMEUuid, streamID);
    
    // If the destination in a Link add Link specific properties to message
    if( _isLink )
    {
      flushMsg = (ControlFlushed)addLinkProps(flushMsg);
    }
    
    _mpio.sendToMe(_targetMEUuid, SIMPConstants.MSG_HIGH_PRIORITY, flushMsg);   
   
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "sendFlushedMessage");
  }
65691
@Override
    public void processMessage() {
        VectorAggregation aggregation = new VectorAggregation(rowIndex, (short) voidConfiguration.getNumberOfShards(),
                        getShardIndex(), storage.getArray(key).getRow(rowIndex).dup());
        aggregation.setOriginatorId(this.getOriginatorId());

        clipboard.pin(aggregation);

        DistributedVectorMessage dvm = new DistributedVectorMessage(key, rowIndex);
        dvm.setOriginatorId(this.originatorId);

        if (voidConfiguration.getNumberOfShards() > 1)
            transport.sendMessageToAllShards(dvm);
        else {
            aggregation.extractContext(this);
            aggregation.processMessage();
        }
    }
29028
private void sendMessage(NetworkMessage message)
    {
        try (ByteArrayOutputStream encode = message.encode())
        {
            final byte[] encoded = encode.toByteArray();
            // Message header
            out.writeByte(NetworkMessageSystemId.USER_MESSAGE);
            out.writeByte(message.getClientId());
            out.writeByte(message.getClientDestId());
            out.writeByte(message.getType());
            // Message content
            out.writeInt(encoded.length);
            out.write(encoded);
            out.flush();

            final int headerSize = 8;
            bandwidth += headerSize + encoded.length;
        }
        catch (final IOException exception)
        {
            Verbose.exception(exception, "Unable to send the message for client: ", String.valueOf(clientId));
        }
    }
1035342
private void sendHandshake(ChannelHandlerContext ctx) {
    boolean needToFlush = false;

    // Iterate until there is nothing left to write.
    while (true) {
      buffer = getOrCreateBuffer(ctx.alloc());
      try {
        handshaker.getBytesToSendToPeer(buffer);
      } catch (GeneralSecurityException e) {
        throw new RuntimeException(e);
      }
      if (!buffer.isReadable()) {
        break;
      }

      needToFlush = true;
      @SuppressWarnings("unused") // go/futurereturn-lsc
      Future<?> possiblyIgnoredError = ctx.write(buffer);
      buffer = null;
    }

    // If something was written, flush.
    if (needToFlush) {
      ctx.flush();
    }
  }
112544
protected ControlRequestFlush createControlRequestFlush() throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createControlRequestFlush");

        ControlRequestFlush rflushMsg = null;

        // Create new message
        try
        {
            rflushMsg = _cmf.createNewControlRequestFlush();
        } catch (MessageCreateFailedException e)
        {
            // FFDC
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.PtoPInputHandler.createControlRequestFlush",
                                        "1:1717:1.323",
                                        this);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            {
                SibTr.exception(tc, e);
                SibTr.exit(tc, "createControlRequestFlush", e);
            }

            SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                        new Object[] {
                                      "com.ibm.ws.sib.processor.impl.PtoPInputHandler",
                                      "1:1729:1.323",
                                      e });

            throw new SIResourceException(
                            nls.getFormattedMessage(
                                                    "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                    new Object[] {
                                                                  "com.ibm.ws.sib.processor.impl.PtoPInputHandler",
                                                                  "1:1737:1.323",
                                                                  e },
                                                    null),
                            e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createControlRequestFlush", rflushMsg);

        return rflushMsg;
    }
197131
public void messageEventOccurred(
    int event,
    SIMPMessage msg,
    TransactionCommon tran)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "messageEventOccurred", 
        new Object[]{new Integer(event), msg, tran});
        
    InvalidOperationException e = 
      new InvalidOperationException(nls.getFormattedMessage(
        "INTERNAL_MESSAGING_ERROR_CWSIP0001",
        new Object[] {
          "com.ibm.ws.sib.processor.impl.PubSubOutputHandler",
          "1:603:1.164.1.5" },
        null));
    
    // FFDC
    FFDCFilter.processException(
      e,
      "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.messageEventOccurred",
      "1:610:1.164.1.5",
      this);
    
    SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
    new Object[] {
      "com.ibm.ws.sib.processor.impl.PubSubOutputHandler",
      "1:616:1.164.1.5" } );
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "messageEventOccurred", e);
    throw e;  
  }
816829
public void sendNotFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID, long requestID) 
  throws SIResourceException 
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "sendNotFlushedMessage", 
        new Object[]{ignore, streamID, new Long(requestID)});

    ControlNotFlushed notFlushed = createControlNotFlushed(streamID, requestID);
    notFlushed = sourceStreamManager.stampNotFlushed(notFlushed);
    
    // If the destination in a Link add Link specific properties to message
    if( isLink )
    {
      notFlushed = (ControlNotFlushed)addLinkProps(notFlushed);
      // The following is set so we can create the linkReceiver correctly on the 
      // other side of the link - 499581
      notFlushed.setRoutingDestination(routingDestination);
    }
    else if( this.isSystemOrTemp )
    {
      // If the destination is system or temporary then  the 
      // routingDestination into th message
      notFlushed.setRoutingDestination(routingDestination); 
    }
    
 
    mpio.sendToMe(routingMEUuid, SIMPConstants.MSG_HIGH_PRIORITY, notFlushed);
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "sendNotFlushedMessage");
  }
329505
@Override
  public boolean acceptInboundMessage(Object msg) throws Exception {
    if (msg instanceof ChunkFetchRequest) {
      return false;
    } else {
      return super.acceptInboundMessage(msg);
    }
  }
932188
@Override
    public void sendSilenceMessage(
                                   long startStamp,
                                   long endStamp,
                                   long completedPrefix,
                                   boolean requestedOnly,
                                   int priority,
                                   Reliability reliability,
                                   SIBUuid12 stream)
    {
        // NOP, the PubSubInputHandler doesn't originate
        // silence.  The associated output handlers do it
        // for us.
    }
490991
public void sendSharedObjectMessage(String name, int currentVersion, boolean persistent, Set<ISharedObjectEvent> events) {
        // create a new sync message for every client to avoid concurrent access through multiple threads
        SharedObjectMessage syncMessage = state.getEncoding() == Encoding.AMF3 ? new FlexSharedObjectMessage(null, name, currentVersion, persistent) : new SharedObjectMessage(null, name, currentVersion, persistent);
        syncMessage.addEvents(events);
        try {
            // get the channel for so updates
            Channel channel = getChannel(3);
            if (log.isTraceEnabled()) {
                log.trace("Send to channel: {}", channel);
            }
            channel.write(syncMessage);
        } catch (Exception e) {
            log.warn("Exception sending shared object", e);
        }
    }
1281017
private void sendAck()
   {
     if (tc.isEntryEnabled())
       SibTr.entry(tc, "sendAck");
     
     try
     {
       upControl.sendAckMessage(null, null, null, iack, priority, reliability, streamID, false);
     }
     catch (SIException e)
     {
       // FFDC
       FFDCFilter.processException(
         e,
         "com.ibm.ws.sib.processor.gd.InternalInputStream.sendAck",
         "1:161:1.42",
         this);
         
       SibTr.exception(tc, e);

       //TODO don't know what to do here - rethrow?!
     } 
     
     if (tc.isEntryEnabled())
       SibTr.exit(tc, "sendAck");
   }
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);
    }
1032770
public void commitAddActiveMessage()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "commitAddActiveMessage");

    // Lock the active message counter lock
    synchronized(maxActiveMessageLock)
    {
      // Move the prepared add to the real count
      currentActiveMessages++;
      preparedActiveMessages--;
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Active Messages: current: " + currentActiveMessages +
                          ", prepared: " + preparedActiveMessages +
                          ", maximum: " + maxActiveMessages +
                          " (suspended: " + consumerSetSuspended + ")");
      
      // We should never go negative, if we do something has gone wrong - FFDC
      if(preparedActiveMessages < 0)
      {
        SIErrorException e = new SIErrorException(
            nls.getFormattedMessage(
              "INTERNAL_MESSAGING_ERROR_CWSIP0001",
              new Object[] {
                "com.ibm.ws.sib.processor.impl.JSConsumerSet.commitAddActiveMessage",
                "1:499:1.16",
                Integer.valueOf(preparedActiveMessages),
                Integer.valueOf(currentActiveMessages),
                Integer.valueOf(maxActiveMessages),
                Boolean.valueOf(consumerSetSuspended)
                },
              null));

          FFDCFilter.processException(
            e,
            "com.ibm.ws.sib.processor.impl.JSConsumerSet.commitAddActiveMessage",
            "1:510:1.16",
            this);

          SibTr.exception(tc, e);
          SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
            new Object[] {
              "com.ibm.ws.sib.processor.impl.JSConsumerSet.commitAddActiveMessage",
              "1:517:1.16" });
      }
      
      // If we currently hold the write lock (as we will always either hold
      // the read lock of the write lock we know this check will return the right
      // answer (i.e. it can't change under us)) we must have thought this add
      // would take us up to the limit and therefore require a suspend.
      if(maxActiveMessagePrepareLock.isWriteLockedByCurrentThread())
      {
        // If the limit has been reached, suspend the set
        if(currentActiveMessages == maxActiveMessages)
        {
          // All this does is mark the set as suspended. We leave it up to the members
          // of the set to notice that the set is suspended and suspend themselves.
          // This is more efficient than going round and suspending everyone, only to
          // have to resume then all again in a second. Hopefully the set will be
          // resumed before most of them notice it was suspended.
          consumerSetSuspended = true;

          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "JSConsumerSet suspended " + this);
        }
        
        // Release the write lock
        maxActiveMessagePrepareWriteLock.unlock();
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "maxActiveMessagePrepareWriteLock.unlock(): " + maxActiveMessagePrepareLock);
      }
      else
      {
        // We only held a read lock, release it now
        maxActiveMessagePrepareReadLock.unlock();
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "maxActiveMessagePrepareReadLock.unlock(): " + maxActiveMessagePrepareLock);
      }
    }

    if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(this, tc, "commitAddActiveMessage");
  }
1304501
@Override
    public void setFlushMode(FlushModeType flushMode)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            Tr.debug(tc, "em.setFlushMode(" + flushMode + ");\n" + toString());

        getEMInvocationInfo(false).setFlushMode(flushMode);
    }
289644
protected ControlAreYouFlushed createControlAreYouFlushed() throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createControlAreYouFlushed");

        ControlAreYouFlushed flushedqMsg = null;

        // Create new message
        try
        {
            flushedqMsg = _cmf.createNewControlAreYouFlushed();
        } catch (MessageCreateFailedException e)
        {
            // FFDC
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.PtoPInputHandler.createControlAreYouFlushed",
                                        "1:1661:1.323",
                                        this);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            {
                SibTr.exception(tc, e);
                SibTr.exit(tc, "createControlAreYouFlushed", e);
            }

            SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002", new Object[] {
                                                                                "com.ibm.ws.sib.processor.impl.PtoPInputHandler",
                                                                                "1:1672:1.323",
                                                                                e });

            throw new SIResourceException(
                            nls.getFormattedMessage(
                                                    "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                    new Object[] {
                                                                  "com.ibm.ws.sib.processor.impl.PtoPInputHandler",
                                                                  "1:1680:1.323",
                                                                  e },
                                                    null),
                            e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createControlAreYouFlushed");

        return flushedqMsg;
    }