Finish building the current frame: process its payload and pass it to the Stream Processor

@throws ProtocolException
168585
public void processCompleteFrame() throws Http2Exception {
        Frame currentFrame = getCurrentFrame();

        boolean frameSizeError = false;
        try {
            currentFrame.processPayload(this);
        } catch (Http2Exception e) {
            // If we get an error here, it should be safe to assume that this frame doesn't have the expected byte count,
            // which must be treated as an error of type FRAME_SIZE_ERROR.  If we're processing a DATA or PRIORITY frame, then
            // we can treat the error as a stream error rather than a connection error.
            if (!e.isConnectionError()) {
                frameSizeError = true;
            } else {
                // this is a connection error; we need to send a GOAWAY on the connection
                throw e;
            }
        } catch (Exception e) {
            throw new ProtocolException("Error processing the payload for " + currentFrame.getFrameType()
                                        + " frame on stream " + currentFrame.getStreamId());
        }

        // call the stream processor to process this stream. For now, don't return from here until the
        // frame has been fully processed.
        int streamId = currentFrame.getStreamId();
        H2StreamProcessor stream = muxLink.getStream(streamId);

        if (stream == null) {
            if ((streamId != 0) && (streamId % 2 == 0)) {
                if (currentFrame.getFrameType().equals(FrameTypes.PRIORITY)) {
                    // ignore PRIORITY frames in any state
                    return;
                } else if (currentFrame.getFrameType().equals(FrameTypes.RST_STREAM) && streamId < muxLink.getHighestClientStreamId()) {
                    // tolerate RST_STREAM frames that are sent on closed push streams
                    return;
                } else {
                    throw new ProtocolException("Cannot start a stream from the client with an even numbered ID. stream-id: " + streamId);
                }
            } else {
                stream = startNewInboundSession(streamId);
            }
        }
        if (frameSizeError) {
            currentFrame = new FrameRstStream(streamId, 4, (byte) 0, false, FrameDirection.READ);
            ((FrameRstStream) currentFrame).setErrorCode(Constants.FRAME_SIZE_ERROR);
        }

        stream.processNextFrame(currentFrame, Direction.READ_IN);
    }
687702
def finish(self) -> None:
        """"""
        if (
            self._expected_content_remaining is not None
            and self._expected_content_remaining != 0
            and not self.stream.closed()
        ):
            self.stream.close()
            raise httputil.HTTPOutputError(
                "Tried to write %d bytes less than Content-Length"
                % self._expected_content_remaining
            )
        if self._chunking_output:
            if not self.stream.closed():
                self._pending_write = self.stream.write(b"0\r\n\r\n")
                self._pending_write.add_done_callback(self._on_write_complete)
        self._write_finished = True
        # If the app finished the request while we're still reading,
        # divert any remaining data away from the delegate and
        # close the connection when we're done sending our response.
        # Closing the connection is the only way to avoid reading the
        # whole input body.
        if not self._read_finished:
            self._disconnect_on_finish = True
        # No more data is coming, so instruct TCP to send any remaining
        # data immediately instead of waiting for a full packet or ack.
        self.stream.set_nodelay(True)
        if self._pending_write is None:
            self._finish_request(None)
        else:
            future_add_done_callback(self._pending_write, self._finish_request)
808420
public void finish() throws IOException 
    {
        if (TraceComponent.isAnyTracingEnabled()&&logger.isLoggable (Level.FINE)) {  //306998.15
            logger.logp(Level.FINE, CLASS_NAME,"finish", "finish");
        } 
        // PM18453 start
        WebContainerRequestState requestState = WebContainerRequestState.getInstance(false);
        if (requestState != null && (Boolean.valueOf((String)requestState.getAttribute("InputStreamEarlyReadCompleted"))).booleanValue()) {
            if (TraceComponent.isAnyTracingEnabled()&&logger.isLoggable (Level.FINE)) {  //306998.15
                logger.logp(Level.FINE, CLASS_NAME,"finish", "Skip read because data has already been read and destroyed.");
            } 
            total=limit;
        }
        // PM18453 End
        else if (!SRTServletResponse.isSkipInputStreamRead() &&  length != -1 )        {
            // if content length set then skip remaining bytes in message body
            long remaining = limit - total;   // PK79219
            while ( remaining > 0 )
            {
                long n = skip(remaining);   // PM03146
                if ( n == 0 )
                {
                    // begin 280584.3    6021: Cleanup of  defect 280584.2    WAS.webcontainer
                    //logger.logp(Level.SEVERE, CLASS_NAME,"finish", nls.getString("Invalid.Content.Length"));
                    // end 280584.3    6021: Cleanup of  defect 280584.2    WAS.webcontainer
                    throw new IOException(nls.getString("Invalid.Content.Length","Invalid content length"));
                }
                remaining -= n;
            }
        }
    }
622553
def process_frame(self):
        
        input_name = self.inputs[0]
        output_name = self.outputs[0]
        in_frame = self.input_buffer[input_name].get()
        out_frame = self.outframe_pool[output_name].get()
        out_frame.initialise(in_frame)
        if self.transform(in_frame, out_frame):
            self.send(output_name, out_frame)
        else:
            raise StopIteration()
382838
@Override
    @FFDCIgnore(IOException.class)
    public void complete(VirtualConnection vc) {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "complete",  vc);
        }
        if (null == vc) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
                Tr.exit(tc, "complete");
            }
            return;
        }
        //We don't need to do an initial seeding of the buffer here as HTTP Channel has already done that in their
        //complete callback from the TCP Channel. When we call an async read on them they will return indicating
        //there is data to read
        synchronized( this.in.getCompleteLockObj()){

            //This variable was introduced to prevent us from calling into Channel again when there is an outstanding ready
            //Once isReady returns false once, we don't want to change it back until the next call into onDataAvailable
            //This variable prevents isReady from returning true if there is an outstanding read           
            this.in.setAsyncReadOutstanding(false);
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Calling user's ReadListener onDataAvailable : " + this.in.getReadListener());
            }       
            
            SRTServletRequestThreadData.getInstance().init(_requestDataAsyncReadCallbackThread);
            
            //Push the original thread's context onto the current thread, also save off the current thread's context
            boolean localPushedThreadContext = false;
            try {
                if (pushedThreadContext == false) {
                    this.threadContextManager.pushContextData();
                    this.pushedThreadContext = true;
                    localPushedThreadContext = true;
                }

                // Call into the user's ReadListener to indicate there is data available
                try{
                    this.in.getReadListener().onDataAvailable();

                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Returned from user's ReadListener onDataAvailable : " + this.in.getReadListener());
                    } 
                } catch (Throwable onDataAvailableException){
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Exception occurred during ReadListener.onDataAvailable : " + onDataAvailableException + ", " + this.in.getReadListener());
                    }
                    error(vc, onDataAvailableException);
                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
                        Tr.exit(tc, "complete");
                    }
                    return;
                }
            
                if(this.in.getReadListener() != null){ // the stream may have been closed during onDataAvailable
                    // cannot call onAllDataRead()

                    //Determine if the message has been fully read. If so call the user's ReadListener to indicate all data has been read
                    //If the message isn't fully read then issue a forced async read to the channel
                    if(in.isFinished()){
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Message is fully read, calling ReadListener onAllDataRead : " + this.in.getReadListener());
                        }
                        try{
                            this.in.getReadListener().onAllDataRead();
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                Tr.debug(tc, "Returned from user's ReadListener onAllDataRead : " + this.in.getReadListener());
                            } 
                        } catch (Throwable onAllDataReadException){
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                Tr.debug(tc, "Exception occurred during ReadListener.onAllDataRead : " + onAllDataReadException + ", " + this.in.getReadListener());
                            }
                            error(vc, onAllDataReadException);
                            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
                                Tr.exit(tc, "complete");
                            }
                            return;
                        }
                    } else {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Data hasn't been fully read yet. There should be an outstanding read at this point : " + this.in.getReadListener());
                        }
                    
                        // Need to call pre-join since the async read will occur on another
                        // thread.
                        if (onErrorDriven == false) {
                            this.in.prepareAsyncReadListener();
                        }
                    }  
                }
                else{
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "ReadListener is not set , stream must be closed, cannot call onAllDataRead()");
                    }
                }
            } finally {
                // Revert back to the thread's current context
                if (localPushedThreadContext == true) {
                    this.pushedThreadContext = false;
                    this.threadContextManager.popContextData();
                }
            }
        }
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "complete");
        }
    }
1015609
private function process()
    {
        if (empty($this->params)) {
            throw new ResolutionException('Parameters were not given');
        }
        $mode = $this->getMode();

        switch ($mode) {
            case 'list':
                $this->listAction();
                break;
            case 'compile':
                $this->compileAction();
                break;
        }
    }
1234482
public function process()
    {
        switch ($this->inputs['action']) {
            case 'login':
                $this->webLogin();
                break;
            case 'logout':
                $this->webLogout();
                exit($this->showWebLoginForm());
                break;
            case 'getContacts':
                $this->getContacts();
                break;
            case 'updateStatus':
                $this->updateStatus();
                break;
            case 'sendMessage':
                $this->sendMessage();
                break;
            case 'sendBroadcast':
                $this->sendBroadcast();
                break;
            default:
                if (isset($_SESSION['logged_in']) && $_SESSION['logged_in'] == true) {
                    exit($this->showWebForm());
                }
                exit($this->showWebLoginForm());
                break;
        }
    }
992636
protected function getReceivingFrame(): Frame
    {
        $current = $this->getCurrentFrame();

        if ($current->isComplete()) {
            if ($current->isFinal()) {
                throw new PayloadException('Payload cannot receive data: it is already complete');
            } else {
                $this->frames[] = $current = $this->getFrame();
            }
        }

        return $current;
    }
750882
protected void processRequest() {
        final int timeout = getHTTPContext().getReadTimeout();
        final TCPReadCompletedCallback callback = HttpICLReadCallback.getRef();

        // keep looping on processing information until we fully parse the message
        // and hand it off, or until the reads for more data go async
        VirtualConnection rc = null;
        do {
            if (handleNewInformation()) {
                // new information triggered an error message, so we're done
                return;
            }
            // Note: handleNewInformation will allocate the read buffers
            if (!isPartiallyParsed()) {
                // we're done parsing at this point. Note: cannot take any action
                // with information after this call because it may go all the way
                // from the channel above us back to the persist read, must exit
                // this callstack immediately
                handleNewRequest();
                return;
            }
            rc = this.myTSC.getReadInterface().read(1, callback, false, timeout);
        } while (null != rc);
    }
952312
def send_frame(self, frame):
        

        # Get the socket
        sock = self.manager.sock

        if not sock:
            raise ValueError("UDPTendrilManager not running")

        # Send the packet
        try:
            sock.sendto(self._send_streamify(frame), self.remote_addr)
        except Exception:
            # We're a best-effort service anyway, so ignore exceptions
            pass
784084
private void processStreamUpload(final UploadStream req) {
    assert (req.body() == null);
    try {
      RpcResponseCallback callback = new RpcResponseCallback() {
        @Override
        public void onSuccess(ByteBuffer response) {
          respond(new RpcResponse(req.requestId, new NioManagedBuffer(response)));
        }

        @Override
        public void onFailure(Throwable e) {
          respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
        }
      };
      TransportFrameDecoder frameDecoder = (TransportFrameDecoder)
          channel.pipeline().get(TransportFrameDecoder.HANDLER_NAME);
      ByteBuffer meta = req.meta.nioByteBuffer();
      StreamCallbackWithID streamHandler = rpcHandler.receiveStream(reverseClient, meta, callback);
      if (streamHandler == null) {
        throw new NullPointerException("rpcHandler returned a null streamHandler");
      }
      StreamCallbackWithID wrappedCallback = new StreamCallbackWithID() {
        @Override
        public void onData(String streamId, ByteBuffer buf) throws IOException {
          streamHandler.onData(streamId, buf);
        }

        @Override
        public void onComplete(String streamId) throws IOException {
           try {
             streamHandler.onComplete(streamId);
             callback.onSuccess(ByteBuffer.allocate(0));
           } catch (Exception ex) {
             IOException ioExc = new IOException("Failure post-processing complete stream;" +
               " failing this rpc and leaving channel active", ex);
             callback.onFailure(ioExc);
             streamHandler.onFailure(streamId, ioExc);
           }
        }

        @Override
        public void onFailure(String streamId, Throwable cause) throws IOException {
          callback.onFailure(new IOException("Destination failed while reading stream", cause));
          streamHandler.onFailure(streamId, cause);
        }

        @Override
        public String getID() {
          return streamHandler.getID();
        }
      };
      if (req.bodyByteCount > 0) {
        StreamInterceptor<RequestMessage> interceptor = new StreamInterceptor<>(
          this, wrappedCallback.getID(), req.bodyByteCount, wrappedCallback);
        frameDecoder.setInterceptor(interceptor);
      } else {
        wrappedCallback.onComplete(wrappedCallback.getID());
      }
    } catch (Exception e) {
      logger.error("Error while invoking RpcHandler#receive() on RPC id " + req.requestId, e);
      respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
      // We choose to totally fail the channel, rather than trying to recover as we do in other
      // cases.  We don't know how many bytes of the stream the client has already sent for the
      // stream, it's not worth trying to recover.
      channel.pipeline().fireExceptionCaught(e);
    } finally {
      req.meta.release();
    }
  }
1308236
private void processCompleteHeaders(boolean isPush) throws CompressionException, ProtocolException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "processCompleteHeaders entry: stream " + myID);
        }
        if (this.headerBlock != null) {
            //Leave this to maintain current functionality, but passing a buffer
            //down to the channel is not necessary. H2 Headers will be stored in the H2
            //inbound link wrap.
            WsByteBufferPoolManager bufManager = HttpDispatcher.getBufferManager();
            WsByteBuffer buf = bufManager.allocate(getByteCount(headerBlock));

            int firstBlockLength = headerBlock.get(0).length;
            for (byte[] byteArray : headerBlock) {
                buf.put(byteArray);
            }
            buf.flip();
            headerBlock = null;

            //Get ready to decode headers
            boolean isFirstLineComplete = false;
            HashMap<String, String> pseudoHeaders = new HashMap<String, String>();
            ArrayList<H2HeaderField> headers = new ArrayList<H2HeaderField>();
            H2HeaderField current = null;
            boolean isFirstHeaderBlock;
            boolean isFirstHeader = true;
            boolean processTrailerHeaders = headersCompleted;

            //Decode headers until we reach the end of the buffer
            while (buf.hasRemaining()) {
                isFirstHeaderBlock = buf.position() < firstBlockLength;
                try {
                    current = (H2Headers.decodeHeader(buf, this.muxLink.getReadTable(), isFirstHeader && isFirstHeaderBlock,
                                                      processTrailerHeaders && !isPush, this.muxLink.getLocalConnectionSettings()));
                } catch (Http2Exception e) {
                    buf.release();
                    throw e;
                }
                if (current == null) {
                    // processed a dynamic table size update; go to the next header
                    continue;
                }
                isFirstHeader = false;
                if (!isFirstLineComplete) {
                    //Is this a Pseudo-Header?
                    if (current.getName().startsWith(":")) {
                        //Verify if this is a valid pseudo-header. In the case that this pseudo-header had
                        //already been decoded (two pseudo-headers with same name), throw a decompression
                        //exception and invalidate the table.
                        if (pseudoHeaders.get(current.getName()) != null) {
                            this.muxLink.getReadTable().setDynamicTableValidity(false);
                            buf.release();
                            throw new CompressionException("Invalid pseudo-header for decompression context: " + current.toString());
                        }
                        pseudoHeaders.put(current.getName(), current.getValue());
                    }
                    //If it doesn't start with colon, it is a regular header
                    //so mark this as the beginning of the parsing headers
                    //state and try to set the decoded first line.
                    else {
                        isFirstLineComplete = true;
                        if (H2Headers.getContentLengthValue(current) > -1) {
                            expectedContentLength = H2Headers.getContentLengthValue(current);
                        }
                        headers.add(current);
                    }
                } else {
                    //If header starts with ':' throw error
                    if (current.getName().startsWith(":")) {
                        this.muxLink.getReadTable().setDynamicTableValidity(false);
                        buf.release();
                        throw new CompressionException("Invalid pseudo-header decoded: all pseudo-headers must appear " +
                                                       "in the header block before regular header fields.");
                    }
                    if (H2Headers.getContentLengthValue(current) > -1) {
                        expectedContentLength = H2Headers.getContentLengthValue(current);
                    }
                    headers.add(current);
                }
            }
            buf.release();
            // only set headers on the link once
            if ((isPush || !processTrailerHeaders) && h2HttpInboundLinkWrap.getHeadersLength() == 0) {
                if (!isValidH2Request(pseudoHeaders)) {
                    muxLink.setContinuationExpected(false);
                    muxLink.setWriteContinuationExpected(false);
                    ProtocolException e = new ProtocolException("An invalid request was received on stream-id: " + myID);
                    e.setConnectionError(false); // mark this as a stream error so we'll generate an RST_STREAM
                    throw e;
                }
                if (pseudoHeaders.get(HpackConstants.AUTHORITY) != null) {
                    muxLink.setAuthority(pseudoHeaders.get(HpackConstants.AUTHORITY));
                }
                //Add all decoded pseudo-headers and headers
                this.h2HttpInboundLinkWrap.setReadHeaders(headers);
                this.h2HttpInboundLinkWrap.setReadPseudoHeaders(pseudoHeaders);
            }
        } else {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "processCompleteHeaders no header block set: stream " + myID);
            }
        }
    }
108476
@Override protected synchronized void done() {
        // dispose of canvas frames on the Event Dispatcher Thread...
        if (monitorWindows != null) {
            for (int i = 0; i < monitorWindows.length; i++) {
                if (monitorWindows[i] != null) {
                    monitorWindows[i].dispose();
                    monitorWindows[i] = null;
                }
            }
        }
        if (projectorFrame != null) {
            projectorFrame.dispose();
            projectorFrame = null;
            projectorConverter = null;
        }

        // force release of native memory
        System.gc();
        Pointer.deallocateReferences();
    }
952126
@Override
	public void run() {

		// Read buffer
		ByteBuffer readBuffer = ByteBuffer.allocate(Math.min(maxPacketSize, 1024 * 1024));
		byte[] readArray = readBuffer.array();

		// Processing variables
		Iterator<SelectionKey> keys;
		SocketChannel channel;
		SelectionKey key;

		byte[] bytes;
		byte[] remaining;
		int processed;
		int pos;

		// Loop
		while (true) {

			// Waiting for sockets
			int n;
			try {
				n = selector.select(3000);
			} catch (NullPointerException nullPointer) {
				continue;
			} catch (Exception anyError) {
				break;
			}
			if (n < 1) {
				continue;
			}
			keys = selector.selectedKeys().iterator();
			while (keys.hasNext()) {
				key = keys.next();
				if (key == null) {
					continue;
				}
				if (!key.isValid()) {
					keys.remove();
					continue;
				}
				if (key.isAcceptable()) {

					// Accept channel
					try {

						// Register socket
						channel = serverChannel.accept();
						channel.configureBlocking(false);

						channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
						channel.setOption(StandardSocketOptions.TCP_NODELAY, true);
						channel.setOption(StandardSocketOptions.SO_LINGER, -1);

						channel.register(selector, SelectionKey.OP_READ);

						// Debug
						if (debug) {
							logger.info("Client channel opened from " + channel.getRemoteAddress() + ".");
						}

					} catch (Exception cause) {
						close(key, cause);
					} finally {
						keys.remove();
					}

				} else if (key.isReadable()) {
					try {

						// Read bytes into the read buffer
						channel = (SocketChannel) key.channel();
						readBuffer.clear();
						n = channel.read(readBuffer);
						if (n < 0) {
							throw new EOFException();
						}
						if (n == 0) {
							keys.remove();
							continue;
						}

						// Debug
						if (debug) {
							logger.info(n + " bytes received from " + channel.getRemoteAddress() + ".");
						}

						// Get the incoming bytes
						bytes = new byte[n];
						System.arraycopy(readArray, 0, bytes, 0, n);

						// Add remaining bytes
						remaining = (byte[]) key.attachment();
						if (remaining != null) {
							byte[] tmp = new byte[remaining.length + bytes.length];
							System.arraycopy(remaining, 0, tmp, 0, remaining.length);
							System.arraycopy(bytes, 0, tmp, remaining.length, bytes.length);
							bytes = tmp;
							key.attach(null);
						}

						// Split data
						pos = 0;
						while ((processed = processPacket(bytes, pos)) > 0) {
							pos += processed;
						}

						// Has remaining?
						if (pos < bytes.length) {
							if (pos == 0) {
								remaining = bytes;
							} else {
								remaining = new byte[bytes.length - pos];
								System.arraycopy(bytes, pos, remaining, 0, remaining.length);
							}
							key.attach(remaining);
						}

					} catch (EOFException eof) {

						// Peer closed
						close(key, null);

					} catch (Exception cause) {
						close(key, cause);
					} finally {
						keys.remove();
					}
				}
			}
		}
	}
782994
void process(ByteChannel sock, AutoBuffer ab) throws IOException {
        int requestType = ab.get1();
        switch (requestType) {
            case CREATE_FRAME:
                ExternalFrameWriterBackend.handleWriteToChunk(sock, ab);
                break;
            case DOWNLOAD_FRAME:
                ExternalFrameReaderBackend.handleReadingFromChunk(sock, ab);
                break;
        }
    }
1037618
public void finish() throws IOException {
		int lastBlockSize = (int)(byteCount % blockSize);
		if(lastBlockSize!=0) {
			while(lastBlockSize<blockSize) {
				out.write(padding);
				byteCount++;
				lastBlockSize++;
			}
		}
		out.flush();
	}
395232
private function finalizeOutputBuffer()
    {
        // Clean out the output buffer to the response, and send the built-up response to the client
        if ($this->isBuffering) {
            $this->response->appendBody(ob_get_contents());
            $this->isBuffering = false;

            ob_end_clean();
        }
    }