/* (non-Javadoc)
@see com.ibm.ws.sib.processor.impl.interfaces.DownstreamControl#sendFlushedMessage(com.ibm.ws.sib.utils.SIBUuid12)

Sends an 'I am flushed' message in response to a query from a target
328567
public void sendFlushedMessage(SIBUuid8 ignoreUuid, SIBUuid12 streamID) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "sendFlushedMessage", new Object[]{ignoreUuid, streamID});

    ControlFlushed flushMsg = createControlFlushed(streamID);
    // If the destination in a Link add Link specific properties to message
    if( isLink )
    {
      flushMsg = (ControlFlushed)addLinkProps(flushMsg);
    }
    
    // If the destination is system or temporary then  the 
    // routingDestination into th message
    if( this.isSystemOrTemp )
    {
      flushMsg.setRoutingDestination(routingDestination); 
    }
 
    mpio.sendToMe(routingMEUuid, SIMPConstants.MSG_HIGH_PRIORITY, flushMsg);
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "sendFlushedMessage");
  }
583037
public void forceFlush(SIBUuid12 streamID) throws SIErrorException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "forceFlush", streamID);

        StreamSet streamSet = streamSets.get(streamID);
        streamSet.dereferenceControlAdapter();

        // Send out a flushed message.  If this fails, make sure we get
        // to at least invoke the callback.
        try
        {
            // The Cellule argument is null as it is ignored by our parent handler
            // this flush message should be broadcast downstream.
            // This will also implicitly remove the streamSet
            downControl.sendFlushedMessage(null, streamID);
        } catch (Exception e)
        {
            // FFDC
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.gd.InternalOutputStreamManager.forceFlush",
                                        "1:743:1.48.1.1",
                                        this);

            // Note that it doesn't make much sense to throw an exception here since
            // this is a callback from stream array map.
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "forceFlush", e);
            return;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "forceFlush");
    }
783637
public static Message buildFlush(Zxid lastProposedZxid, ByteBuffer body) {
    ZabMessage.Zxid zxid = toProtoZxid(lastProposedZxid);
    ZabMessage.Flush flush = ZabMessage.Flush.newBuilder()
                                             .setZxid(zxid)
                                             .setBody(ByteString.copyFrom(body))
                                             .build();
    return Message.newBuilder().setType(MessageType.FLUSH).setFlush(flush)
                  .build();
  }
427350
public void flush(final long targetCounter) throws IOException, JournalClosedException {
    // Return if flushed.
    if (targetCounter <= mFlushCounter.get()) {
      return;
    }

    // Submit the ticket for flush thread to process.
    FlushTicket ticket = new FlushTicket(targetCounter);
    try (LockResource lr = new LockResource(mTicketLock)) {
      mTicketList.add(ticket);
    }

    try {
      // Give a permit for flush thread to run.
      mFlushSemaphore.release();

      // Wait on the ticket until completed.
      ticket.waitCompleted();
    } catch (InterruptedException ie) {
      // Interpret interruption as cancellation.
      throw new AlluxioStatusException(Status.CANCELLED.withCause(ie));
    } catch (Throwable e) {
      // Filter, journal specific exception codes.
      if (e instanceof IOException) {
        throw (IOException) e;
      }
      if (e instanceof JournalClosedException) {
        throw (JournalClosedException) e;
      }
      // Not expected. throw internal error.
      throw new AlluxioStatusException(Status.INTERNAL.withCause(e));
    } finally {
      /**
       * Client can only try to reacquire the permit it has given
       * because the permit may or may not have been used by the flush thread.
       */
      mFlushSemaphore.tryAcquire();
    }
  }
843523
public byte[] sendQuerySmResp(OutputStream os, int sequenceNumber,
            String messageId, String finalDate, MessageState messageState,
            byte errorCode) throws PDUStringException, IOException {
        byte[] b = pduComposer.querySmResp(sequenceNumber, messageId,
                finalDate, messageState.value(), errorCode);
        writeAndFlush(os, b);
        return b;
    }
939367
@Override
  protected void flushAndSync(boolean durable) throws IOException {
    if (outputStream == null) {
      throw new IOException("Trying to use aborted output stream!");
    }
    if (doubleBuf.isFlushed()) {
      return;
    }
    // Under the covers, this invokes a method that does not return until
    // we have successfully persisted the buffer to a quorum of bookies.
    doubleBuf.flushTo(outputStream);
  }
1293816
public WriterFlushResult withFlushResult(WriterFlushResult flushResult) {
        this.flushedBytes.addAndGet(flushResult.flushedBytes.get());
        this.mergedBytes.addAndGet(flushResult.mergedBytes.get());
        this.flushedAttributes.addAndGet(flushResult.flushedAttributes.get());
        return this;
    }
645210
public void processFlushQuery(ControlAreYouFlushed flushQuery) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processFlushQuery", new Object[] { flushQuery });

    SIBUuid12 streamID = flushQuery.getGuaranteedStreamUUID();

    try
    {
      //synchronize to give a consistent view of the flushed state
      synchronized(this)
      {
        SIBUuid8 requestor = flushQuery.getGuaranteedSourceMessagingEngineUUID();
        if(isFlushed(streamID))
        {
          // It's flushed.  Send a message saying as much.
          downControl.sendFlushedMessage(requestor, streamID);
        }
        else
        {
          // Not flushed.  Send a message saying as much.
          downControl.sendNotFlushedMessage(requestor, streamID, flushQuery.getRequestID());
        }
      }
    }
    catch (SIResourceException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.gd.SourceStreamManager.processFlushQuery",
        "1:1333:1.102",
        this);

      SibTr.exception(tc, e);

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

      throw e;
    }

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

  }
394260
@Override
    public void sendAckMessage(
                               SIBUuid8 upstream,
                               SIBUuid12 destUuid,
                               SIBUuid8 busUuid,
                               long ackPrefix,
                               int priority,
                               Reliability reliability,
                               SIBUuid12 stream,
                               boolean consolidate) throws SIResourceException, SIErrorException
    {

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

        if (upstream == null)
        {
            upstream = _originStreamMap.get(stream);
        }

        ControlAck ackMsg = createControlAckMessage(priority, reliability, stream);

        ackMsg.setAckPrefix(ackPrefix);

        if ((consolidate) &&
            (_internalInputStreamManager.hasStream(stream, priority, reliability)))
        {
            try
            {
                processAck(ackMsg);
            } catch (SIResourceException e)
            {
                // FFDC
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.PubSubInputHandler.sendAckMessage",
                                            "1:600:1.329.1.1",
                                            this);
            }
        }
        else
        {
            // Send this to MPIO
            // Send Ack messages at the priority of the original message +1
            _mpio.sendToMe(
                           upstream,
                           priority + 1,
                           ackMsg);

        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendAckMessage");
    }
1291890
@Override
    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 });

        switch (event)
        {
            case MessageEvents.POST_COMMIT_REMOVE:
            case MessageEvents.POST_ROLLBACK_REMOVE:
            case MessageEvents.UNLOCKED:
                // Reset ordered tran id
                if (currentTranId != null)
                {
                    currentTranId = null;
                    break;
                }
            case MessageEvents.POST_COMMIT_ADD:
            case MessageEvents.POST_ROLLBACK_ADD:

                deleteIfPossible(true);
                break;

            default: {
                final SIErrorException e = new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0001",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.store.itemstreams.SubscriptionItemStream",
                                                                      "1:241:1.54" },
                                                        null));

                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.store.itemstreams.SubscriptionItemStream.messageEventOccurred",
                                            "1:247:1.54",
                                            this);

                SibTr.exception(tc, e);
                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.store.itemstreams.SubscriptionItemStream",
                                          "1:254:1.54" });

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

                throw e;

            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "messageEventOccurred");
    }
163896
public int flush(final LogSender sender) throws IOException, HttpException {

		int numSent = 0;
		int maxToSend = queue.size();

		if (0 < maxToSend) {
			AppIdentity appIdentity = appIdentityService.getAppIdentity();

			while (numSent < maxToSend) {

				// get the next batch of messages
				int batchSize = Math.min(maxToSend - numSent, MAX_BATCH);

				List<LogMsg> batch = new ArrayList<LogMsg>(batchSize);

				for (int i = 0; i < batchSize; ++i) {
					batch.add(queue.remove());
				}

				// build the log message group
				LogMsgGroup group = createLogMessageGroup(batch, platform, logger, envDetail, appIdentity);

				// send the batch to Stackify
				int httpStatus = sender.send(group);

				// if the batch failed to transmit, return the appropriate transmission status
				if (httpStatus != HttpURLConnection.HTTP_OK) {
					throw new HttpException(httpStatus);
				}

				// next iteration
				numSent += batchSize;
			}
		}

		return numSent;
	}
1047191
protected void sendTransacted(Destination destination, Serializable message) {
        send(destination, message, null, null, true);
    }
365153
def flush(self):
        
        # Flush buffer
        line = self._buffer.getvalue()
        self._buffer = StringIO()

        # Send the message
        content = {"session_id": self._session, "text": line}
        self._herald.fire(self._peer, beans.Message(MSG_CLIENT_PRINT, content))
447917
@Override
  protected void flush(Buffer data, boolean isEnd)
  {
    if (isClosed()) {
      throw new IllegalStateException();
    }

    _request.outProxy().write(_request, data, isEnd);
    
    /*
    boolean isHeader = ! isCommitted();
    toCommitted();
    
    MessageResponseHttp2 message;
    
    TempBuffer next = head;
    TempBuffer tBuf = null;
    
    if (head != null) {
      int headLength = head.length();
    
      if (headLength > 0) {
        tBuf = head;
      
        next = TempBuffer.allocate();
      }
    }
    
    message = new MessageResponseHttp2(_request, tBuf, isHeader, isEnd);
    _request.getOutHttp().offer(message);
    
    return next;
    */
  }
347411
@Override
    public void messageSent(final NextFilter nextFilter,
            final IoSession session, final WriteRequest writeRequest)
            throws Exception {
        if (writeRequest.getMessage() != null
                && writeRequest.getMessage() instanceof ProxyHandshakeIoBuffer) {
            // Ignore buffers used in handshaking
            return;
        }

        nextFilter.messageSent(session, writeRequest);
    }
1238723
public void addFlush() {
        // There is no need to process all entries if there was already a flush before and no new messages
        // where added in the meantime.
        //
        // See https://github.com/netty/netty/issues/2577
        Entry entry = unflushedEntry;
        if (entry != null) {
            if (flushedEntry == null) {
                // there is no flushedEntry yet, so start with the entry
                flushedEntry = entry;
            }
            do {
                flushed ++;
                if (!entry.promise.setUncancellable()) {
                    // Was cancelled so make sure we free up memory and notify about the freed bytes
                    int pending = entry.cancel();
                    decrementPendingOutboundBytes(pending, false, true);
                }
                entry = entry.next;
            } while (entry != null);

            // All flushed so reset unflushedEntry
            unflushedEntry = null;
        }
    }
1252857
private void handleNewStreamID(AbstractMessage aMessage, MessageItem msgItem) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "handleNewStreamID", new Object[] { aMessage, msgItem });

    SIBUuid12 streamID = aMessage.getGuaranteedStreamUUID();

    // Synchronize to resolve race between multiple messages on same
    // stream.  Can't actually happen until multihop, e.g. messages on
    // same stream arrive simultaneously from multiple cellules.
    synchronized (flushMap)
    {
      // Otherwise, create a new flush query record and proceed
      FlushQueryRecord entry = flushMap.get(streamID);

      if ( (entry != null) && (msgItem != null) )
      {
        // Since the entry already exists, somebody else already
        // started the timer so all we have to do is cache the
        // message for replay later.
        entry.append(msgItem);
      }
      else
      {
        // Otherwise, new entry:
        // 1. Create a request ID and flush record, store it in the map.
        // 2. Create and send an "are you flushed".
        // 3. Start an alarm to repeat "are you flushed" a set number of
        //    times.

        SIBUuid8 sourceMEUuid = aMessage.getGuaranteedSourceMessagingEngineUUID();

        SIBUuid12 destID = aMessage.getGuaranteedTargetDestinationDefinitionUUID();
        SIBUuid8  busID  = aMessage.getGuaranteedCrossBusSourceBusUUID();

        // Create and store the request record
        long reqID = messageProcessor.nextTick();
        entry = new FlushQueryRecord(sourceMEUuid, destID, busID,
                                     msgItem, reqID);
        flushMap.put(streamID, entry);

        // Create and send the query
        // TODO: create a proper srcID here
        upControl.sendAreYouFlushedMessage(sourceMEUuid, destID, busID, reqID, streamID);

        // Start the alarm.  The context for the alarm is the stream id, which
        // we can use to look up the FlushQueryRecord if the alarm expires.
        entry.resend = am.create(GDConfig.FLUSH_QUERY_INTERVAL, this, streamID);
      }
    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "handleNewStreamID");
  }