Called by the TCP channel when the write has finished.

@param vc
@param wsc
14061
public void complete(VirtualConnection vc, TCPWriteRequestContext wsc) {

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "complete() called: vc=" + vc);
        }
        HttpOutboundServiceContextImpl mySC = (HttpOutboundServiceContextImpl) vc.getStateMap().get(CallbackIDs.CALLBACK_HTTPOSC);
        // LI4335 - handle early reads
        if (mySC.isEarlyRead()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Notifying app channel of write complete");
            }
            mySC.getAppWriteCallback().complete(vc);
            return;
        }
        // if only the headers have been sent, then we need to check some
        // special case handling scenarios
        // 381105 - only start response read after just the headers have been
        // sent and no body... eventually need to get the isPartialBody state
        // implemented to make this simpler.
        if (mySC.isHeadersSentState() && 0 == mySC.getNumBytesWritten()) {
            if (mySC.shouldReadResponseImmediately()) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Sent headers, reading for response");
                }
                mySC.startResponseRead();
                return;
            }
        }

        // if we're here and we need to, notify the channel above that the
        // write has completed, otherwise start the read for the response
        // message
        if (!mySC.isMessageSent()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Calling write complete callback of app channel.");
            }
            mySC.getAppWriteCallback().complete(vc);
        } else {
            if (mySC.shouldReadResponseImmediately()) {
                // we've already done the "read first" call so jump to the
                // regular method now
                mySC.readAsyncResponse();
            } else {
                // initial read for a response
                mySC.startResponseRead();
            }
        }
    }
1005202
protected void nowReconnectedAsync() {

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "Reconnected async for " + this);
        }
        // reset the data buffers first
        if (!resetWriteBuffers()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Resetting buffers (async) failed");
            }
            // otherwise pass the error along to the channel above us, or close
            // the connection if nobody is above
            IOException ioe = new IOException("Failed reconnect");
            if (null != getAppWriteCallback()) {
                getAppWriteCallback().error(getVC(), ioe);
            } else {
                // nobody above us, just close the connection
                getLink().getDeviceLink().close(getVC(), ioe);
            }
            return;
        }

        // now that we've reconnected, we should reset the "broken" flag. First
        // we set it to the default and then recheck the request msg
        setPersistent(true);
        updatePersistence(getRequestImpl());

        // in case we previously read any partial data, clean out the response
        resetRead();

        // attempt to write the data
        VirtualConnection rc = getTSC().getWriteInterface().write(TCPWriteRequestContext.WRITE_ALL_DATA, HttpOSCWriteCallback.getRef(), isForceAsync(), getWriteTimeout());
        if (null != rc) {
            // if we've finished writing part of a request, let the channel
            // above know that it can write more, otherwise start the read
            // for the response
            if (!isMessageSent()) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Calling callback.complete of app channel.");
                }
                getAppWriteCallback().complete(getLink().getVirtualConnection());
            } else {
                if (isReadAheadEnabled()) {
                    // after a reconnect, there is no more read-ahead active
                    this.bReadAheadEnabled = false;
                }
                // force an async read for the response now. Avoid trying to
                // re-use any existing read buffer to skip complexity with
                // adjusting for partial reads before the reconnect.
                setupJITRead(getHttpConfig().getIncomingHdrBufferSize());
                getTSC().getReadInterface().read(1, HttpOSCReadCallback.getRef(), true, getReadTimeout());
            }
        }
    }
1368054
void onWritingThreadFinished(WebSocketFrame closeFrame)
    {
        synchronized (mThreadsLock)
        {
            mWritingThreadFinished = true;
            mClientCloseFrame = closeFrame;

            if (mReadingThreadFinished == false)
            {
                // Wait for the reading thread to finish.
                return;
            }
        }

        // Both the reading thread and the writing thread have finished.
        onThreadsFinished();
    }
239848
@Override
    public void error(VirtualConnection vc, TCPWriteRequestContext wsc, IOException ioe) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "error, vc=" + getVCHash());
        }
        // Nothing else needs to be done here. Report error up the chain.
        this.asyncBytesToWrite = 0L;
        this.callback.error(vc, this, ioe);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "error");
        }
    }
656957
public void closeAfterFlushingPendingWrites(Channel channel, Event event)
	{
		if (channel.isConnected())
		{
			channel.write(event).addListener(ChannelFutureListener.CLOSE);
		}
		else
		{
			System.err.println("Unable to write the Event :" + event
					+ " to socket as channel is ot connected");
		}
	}
303815
@Override
    public void close(VirtualConnection inVC, Exception e) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "close, vc=" + getVCHash());
        }

        // Set closed flag so that ready can't be called again in an error condition.
        // This is a protective measure.
        closed = true;

        // Clean up the read and write interfaces as well as the SSL engine.

        // cleanup has logic to avoid writing if stop(0) has been called
        cleanup();

        //If the channel has already processed the close signal, it is too late to try and clean up the individual connection links here.
        //This race condition should not happen if channels above us are well behaved, so not using synchronize logic here, so as not to
        //impact mainline performance.
        if (this.sslChannel.getstop0Called() != true) {
            if (getDeviceLink() != null) {
                getDeviceLink().close(inVC, e);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "close");
        }
    }
411444
public void error(VirtualConnection vc, TCPReadRequestContext rsc, IOException ioe) {

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "error() called: vc=" + vc + " ioe=" + ioe);
        }
        HttpOutboundServiceContextImpl mySC = (HttpOutboundServiceContextImpl) vc.getStateMap().get(CallbackIDs.CALLBACK_HTTPOSC);
        mySC.setPersistent(false);
        mySC.reConnect(vc, ioe);
    }
191948
protected void onStreamBytesRead(RTMPConnection conn, Channel channel, Header source, BytesRead streamBytesRead) {
        conn.receivedBytesRead(streamBytesRead.getBytesRead());
    }
786536
private void sendStatusAndClose(final Channel chan, final HttpResponseStatus status) {
    if (chan.isConnected()) {
      final DefaultHttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
      final ChannelFuture future = chan.write(response);
      future.addListener(ChannelFutureListener.CLOSE);
    }
  }
283504
@Override
    public void afterCompletion(int status) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) { 
            Tr.entry(this, tc, "afterCompletion", status);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
            Tr.debug(this, tc, "Using transaction wrapper@" + Integer.toHexString(this.hashCode()));
        }


        // When afterCompletionCode is called we need to reset the registeredForSync flag.
        registeredForSync = false;

        /*
         * Mark the transaction complete in the wrapper
         */

        // Do NOT catch runtime exceptions here.  Let them flow out
        // of the component.
        mcWrapper.transactionComplete();

        // Check if any connection handles are still in use
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
            if (mcWrapper.getHandleCount() != 0) {
                // Issue warning that Connections have not been closed by
                // the end of the current UOW scope and will be closed
                // by the CM.
                //            Tr.warning(this, tc,"HANDLE_NOT_CLOSED_J2CA0055");
                // removed null check on mcWrapper since it is redundant from the above code (mcWrapper.getUOWCoordinator())
                String pmiName = mcWrapper.gConfigProps.cfName;
                Tr.debug(this, tc, "Information:  handle not closed at end of UOW for resource " + pmiName);
            }
        }

        // Do NOT catch the runtime exception which getConnectionManager
        // might throw. If this is thrown it is an internal bug and needs
        // to be fixed. Allow the rte to flow up to the container.
        boolean shareable = mcWrapper.getConnectionManager().shareable();

        //
        // If the connection is shareable, or it is non-shareable and the
        // handleCount is zero, then release it back to the poolManger,
        // otherwise simply reset the transaction related variables.
        //
        // Note: shareable connections are released at the end of
        // transactions regardless of outstanding handles because we don't
        // support handled being associated to an active
        //  managedConnection outside of a sharing boundary.
        if ((shareable) || (!shareable && mcWrapper.getHandleCount() == 0)) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
                Tr.debug(this, tc, "afterCompletionCode: Releasing the connection to the pool. Shareable = " + shareable);
            }

            try {
                mcWrapper.releaseToPoolManager();
            } catch (Exception e) {
                // Do not rethrow the exception here since the client has
                // successfully completed the transaction.
                // Log the error and go on.

                // need to change the "291" number below
                com.ibm.ws.ffdc.FFDCFilter.processException(e, "com.ibm.ejs.j2c.RRSLocalTransactionWrapper.afterCompletion", "291", this);
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
                    String pmiName = mcWrapper.gConfigProps.cfName;
                    Tr.debug(this, tc, "afterCompletionCode for resource " + pmiName + ":  caught Exception", e);
                }
            }

        } // end ((shareable) || (!shareable && mcWrapper.getHandleCount() == 0))
        else {
            /*
             * The cleanup of the coordinator and enlisted flag should
             * only be done here for non sharable connections. In other cases,
             * the coord is set to null in mcWrapper.releaseToPoolManager().
             * We only need to worry about this here if the handleCount is
             * greater than 0 (in which case the cleanup will be done
             * by the poolManager). Same thing for the tranFailed flag in
             * the MCWrapper.
             */
            enlisted = false;
            //     mcWrapper.setTranFailed(false);
            mcWrapper.setUOWCoordinator(null);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            Tr.exit(this, tc, "afterCompletion");

    }
733870
private void writeResponse(SocketChannel sc, int rc) throws IOException {
        synchronized (responseLock) {
            write(sc, serverUUID + DELIM + rc);
        }
    }
1201790
private void invokeOnComplete() {
        for (; ; ) {
            if (pooledBuffers != null) {
                for (PooledByteBuffer buffer : pooledBuffers) {
                    buffer.close();
                }
                pooledBuffers = null;
            }
            IoCallback callback = this.callback;
            this.buffer = null;
            this.fileChannel = null;
            this.callback = null;
            inCallback = true;
            try {
                callback.onComplete(exchange, this);
            } finally {
                inCallback = false;
            }

            StreamSinkChannel channel = this.channel;
            if (this.buffer != null) {
                long t = Buffers.remaining(buffer);
                final long total = t;
                long written = 0;

                try {
                    do {
                        long res = channel.write(buffer);
                        written += res;
                        if (res == 0) {
                            if(writeListener == null) {
                                initWriteListener();
                            }
                            channel.getWriteSetter().set(writeListener);
                            channel.resumeWrites();
                            return;
                        }
                    } while (written < total);
                    //we loop and invoke onComplete again
                } catch (IOException e) {
                    invokeOnException(callback, e);
                }
            } else if (this.fileChannel != null) {
                if(transferTask == null) {
                    transferTask = new TransferTask();
                }
                if (!transferTask.run(false)) {
                    return;
                }
            } else {
                return;
            }

        }
    }
531265
@Override
    public void ready(VirtualConnection inVC) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "ready: " + this + " " + inVC);
        }
        this.myTSC = (TCPConnectionContext) getDeviceLink().getChannelAccessor();
        HttpInboundServiceContextImpl sc = getHTTPContext();

        sc.init(this.myTSC, this, inVC, getChannel().getHttpConfig());
        if (getChannel().getHttpConfig().getDebugLog().isEnabled(DebugLog.Level.INFO)) {
            getChannel().getHttpConfig().getDebugLog().log(DebugLog.Level.INFO, HttpMessages.MSG_CONN_STARTING, sc);
        }
        processRequest();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "ready");
        }
    }
1038864
public void writeUncommitted(SIMPMessage m) throws SIResourceException
  {

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "writeUncommitted", new Object[] { m });

    TickRange tr = null;

    JsMessage jsMsg = m.getMessage();
   
    long stamp = jsMsg.getGuaranteedValueValueTick();
    long starts = jsMsg.getGuaranteedValueStartTick();
    long ends = jsMsg.getGuaranteedValueEndTick();

    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
    {
      SibTr.debug(tc, "writeUncommitted at: " + stamp + " on Stream " +  stream);
    }

    synchronized (this)
    {      
      if (m.getStreamIsGuess())
      {
        newGuessInStream(stamp);
      }
      msgAdded(stamp);

      // write into stream
      tr = TickRange.newUncommittedTick(stamp);
      tr.startstamp = starts;
      tr.endstamp = ends;
    
      // Save message in the stream while it is Uncommitted
      // It will be replaced by its index in the ItemStream once it becomes a Value
      tr.value = m;

      oststream.writeCombinedRange(tr);
 
      // Now update our lastMsgAdded with the tick of the
      // message we have just added
      lastMsgAdded = stamp;
      
      // SIB0105
      // If this is the first uncommitted messages to be added to the stream, start off the 
      // blocked message health state timer. This will check back in the future to see if the
      // message successfully sent.
      if (blockedStreamAlarm==null)
      {
        blockedStreamAlarm = new BlockedStreamAlarm(getCompletedPrefix());
        am.create(mp.getCustomProperties().getBlockedStreamHealthCheckInterval(), blockedStreamAlarm);
      }

    } // end synchronized

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "writeUncommitted");
  }
528983
public static void event(TraceComponent tc, String msg) {
    if(isTracing()) SibTr.event(tc, msg);

  }
510516
public void init(VirtualConnection inVC, HttpOutboundChannel channel) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "Init on link: " + this + " " + inVC);
        }
        super.init(inVC);
        this.myChannel = channel;
        this.bIsActive = true;
        setEnableReconnect(this.myChannel.getHttpConfig().allowsRetries());
        setAllowReconnect(true);
    }
106989
@Override
    public void complete(com.ibm.wsspi.channelfw.VirtualConnection vc, com.ibm.wsspi.tcpchannel.TCPWriteRequestContext wsc) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())  
            Tr.debug(tc, "complete called : " + vc + "WriteListener enabled: " + this._wl);

        synchronized(_upgradeOut) {

            if (null == vc) {
                return;
            }

            if (_upgradeOut.getBufferHelper().isOutputStream_closed()) {

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    Tr.debug(tc, "complete , outputStream closed ignoring complete : " + vc);

                return;
            }

            // clean up any request state we had on this thread.
            WebContainerRequestState reqState = WebContainerRequestState.getInstance(true);
            reqState.init();
            try{    
                _upgradeOut.getBufferHelper().writeRemainingToBuffers();                 

            } catch (Exception e) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
                    Tr.debug(tc, "exception during writeBuffers" + e.toString());
                }
                this.error(vc, e);
                return;
            }


            // rest must be written now ,
            // this case will only be true if write which went aysnc was
            // from println, now we have to write crlf
            if ( _upgradeOut.getBufferHelper().isInternalReady() && _upgradeOut.getBufferHelper().write_crlf_pending) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    Tr.debug(tc, "write CRLF bytes  , WriteListener enabled: " + this._wl);
                try {
                    reqState.setAttribute("com.ibm.ws.webcontainer.upgrade.WriteAllowedonThisThread", true);
                    reqState.setAttribute("com.ibm.ws.webcontainer.upgrade.CRLFWriteinPorgress", true);
                    _upgradeOut.getBufferHelper().writeCRLFIfNeeded();
                } catch (IOException e) {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) 
                        Tr.debug(tc, " Exception during write CRLF bytes: " + e);

                    _upgradeOut.getBufferHelper().write_crlf_pending = false;
                    this.error(vc, e);
                    return;
                }
            }

            if( _upgradeOut.getBufferHelper().isInternalReady()){
                WebContainerRequestState.getInstance(true).setAttribute("com.ibm.ws.webcontainer.upgrade.WriteAllowedonThisThread", true);

                _upgradeOut.getBufferHelper().setReadyForApp(true);

                // if close was in progress
                if(_upgradeOut.getBufferHelper().isOutputStream_close_initiated_but_not_Flush_ready()){
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){
                        Tr.debug(tc, "complete , In process of closing outputStream, no more data is left to write, call close");
                    }
                    try {
                        if(_upCon.isOutputStream_CloseStartedFromWC()){
                            //at this point inputstream and handler are already closed
                            _upCon.closeOutputandConnection();
                        }
                        else{
                            _upgradeOut.close();
                        }
                    } catch (Exception e) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                            Tr.debug(tc, "complete , stack exception" + e.toString());

                        this.error(vc, e);
                    }

                }// no close in progress now check if onWP needs to be called.
                else{
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
                        Tr.debug(tc, "WriteListener enabled: " + this._wl + " ,status_not_ready_checked -->" + _upgradeOut.getBufferHelper().status_not_ready_checked);
                    }
                    if (_upgradeOut.getBufferHelper().status_not_ready_checked) {
                        try{
                            _upgradeOut.getBufferHelper().status_not_ready_checked = false;

                            // Push the original thread's context onto the current thread,
                            // also save off the current thread's context
                            _contextManager.pushContextData();
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
                                Tr.debug(tc, "Calling user's WriteListener onWritePossible");
                            }
                            // Call into the user's WriteListener to indicate more data cane
                            // be written
                            _wl.onWritePossible();
                        } catch (Exception e) {
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
                                Tr.debug(tc, "stack exception" + e.toString());
                            }
                            this.error(vc, e);
                        } finally {
                            // Revert back to the thread's current context
                            _contextManager.popContextData();
                        }
                    }
                    else{
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
                            Tr.debug(tc, "WriteListener enabled: " 
                                            + this._wl + " .onWritePossible will be skipped as isReady has not been checked since write went async.");
                        }
                    }
                }
            }
        }
    }