Once a response has been read and parsed asynchronously, this method will
decide what to do next, either starting another read or notifying the
application channel of the parsed message.
18153
protected void handleParsedMessage() {

        InterChannelCallback cb = null;
        // use the read callback for early reads or temp messages requested
        // LI4335 - handle early reads too
        if (this.bEarlyReads || this.bTempResponsesUsed) {
            cb = getAppReadCallback();
        } else {
            cb = getAppWriteCallback();
        }
        VirtualConnection vc = null;
        do {
            this.numResponsesReceived++;
            if (!getResponseImpl().isTemporaryStatusCode()) {
                // a final response message was received
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Notifying app channel of final response.");
                }
                cb.complete(getVC());
                return;
            }

            // received a temporary response
            int code = getResponseImpl().getStatusCodeAsInt();
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Received response (#" + numberResponsesReceived() + "): " + code);
            }
            if (numberResponsesReceived() > getHttpConfig().getLimitOnNumberOfResponses()) {
                // too many temp responses
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Too many async temp responses received.");
                }
                cb.error(getVC(), new IOException("Max temp responses received: " + numberResponsesReceived()));
                return;
            }
            if (this.bTempResponsesUsed) {
                // call the app channel with this new response
                cb.complete(getVC());
                return;
            }
            if (getRequestImpl().isExpect100Continue()) {
                if (100 != code) {
                    // if the Expectation failed, then this connection is finished, so
                    // no need to reset any values.
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Expect 100-continue failed with " + code);
                    }
                    setPersistent(false);
                    cb.error(getVC(), new ExpectationFailedException(code + " " + getResponseImpl().getReasonPhrase()));
                    return;
                } else if (1 == numberResponsesReceived() && isHeadersSentState()) {
                    // first 100-continue only, return to app channel immed but
                    // only if sendHeaders was used (meaning they want 100-continue)
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Notifying channel of first 100-continue");
                    }
                    cb.complete(getVC());
                    return;
                }
            }
            // otherwise, this was a valid temporary response but the app channel
            // doesn't want to see it
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Ignoring temporary response...");
            }
            resetRead();
            if (setupReadBuffers(getHttpConfig().getIncomingHdrBufferSize(), false)) {
                // data is already available
                vc = getVC();
            } else {
                // read for the next bit of data
                vc = getTSC().getReadInterface().read(1, HttpOSCReadCallback.getRef(), false, getReadTimeout());
            }
            if (null != vc) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Attempting a parse of response data.");
                }
                vc = parseResponseMessageAsync();
            }
        } while (null != vc);
    }
525177
private void sendReply(Response<ByteString> response) {
    if (!replied.compareAndSet(false, true)) {
      LOGGER.warn("Already replied to ongoing request {} - spurious response {}", request, response);
    } else {
      final HttpServletResponse httpResponse = (HttpServletResponse) asyncContext.getResponse();

      final StatusType status = response.status();
      httpResponse.setStatus(status.code(), status.reasonPhrase());

      response.headerEntries().forEach(entry ->
          httpResponse.addHeader(entry.getKey(), entry.getValue()));

      response.payload().ifPresent(payload -> {
        try {
          payload.write(httpResponse.getOutputStream());
        } catch (IOException e) {
          LOGGER.warn("Failed to write response", e);
        }
      });

      asyncContext.complete();

      logger.accept(this, Optional.of(response));
    }
  }
1202957
public void startRead() {
		SynchronizationPoint<NoException> sp;
		synchronized (this) {
			if (!writer) {
				// nobody is writing, we can read
				readers++;
				return;
			}
			// if someone is writing, we need to wait
			if (readerWaiting == null)
				readerWaiting = new SynchronizationPoint<NoException>();
			sp = readerWaiting;
			readers++;
		}
		sp.block(0);
	}
409666
void onReadingThreadStarted()
    {
        boolean bothStarted = false;

        synchronized (mThreadsLock)
        {
            mReadingThreadStarted = true;

            if (mWritingThreadStarted)
            {
                // Both the reading thread and the writing thread have started.
                bothStarted = true;
            }
        }

        // Call onConnected() method of listeners if not called yet.
        callOnConnectedIfNotYet();

        // If both the reading thread and the writing thread have started.
        if (bothStarted)
        {
            onThreadsStarted();
        }
    }
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);
    }
1215401
public void connectedReceived() {
        if (null != listenerReference) {
            final Listener listener = listenerReference.get();
            if (null != listener) {
                new Handler(Looper.getMainLooper()).post(new Runnable() {
                    public void run() {
                        if (isActive()) {
                            listener.onConnected(RespokeCall.this);
                        }
                    }
                });
            }
        }
    }
127792
def _handle_offset_response(self, response):
        
        # Got a response, clear our outstanding request deferred
        self._request_d = None

        # Successful request, reset our retry delay, count, etc
        self.retry_delay = self.retry_init_delay
        self._fetch_attempt_count = 1

        response = response[0]
        if hasattr(response, 'offsets'):
            # It's a response to an OffsetRequest
            self._fetch_offset = response.offsets[0]
        else:
            # It's a response to an OffsetFetchRequest
            # Make sure we got a valid offset back. Kafka uses -1 to indicate
            # no committed offset was retrieved
            if response.offset == OFFSET_NOT_COMMITTED:
                self._fetch_offset = OFFSET_EARLIEST
            else:
                self._fetch_offset = response.offset + 1
                self._last_committed_offset = response.offset
        self._do_fetch()
1063348
def wait_response(self):
        ''''''
        responses = self.read()
        while not responses:
            responses = self.read()
        return responses
777997
@Override
    public void onReceive(Object message) {
        if (message instanceof NewMessage) {
            NewMessage newMessage = (NewMessage) message;
            onNewMessage(newMessage.getPeer(), newMessage.getSender(), newMessage.getSortDate(),
                    newMessage.getContentDescription(), newMessage.getHasCurrentUserMention(),
                    newMessage.getUnreadMessagesCount(), newMessage.getUnreadDialogsCount());
        } else if (message instanceof MessagesRead) {
            MessagesRead read = (MessagesRead) message;
            onMessagesRead(read.getPeer(), read.getFromDate());
        } else if (message instanceof PauseNotifications) {
            onNotificationsPaused();
        } else if (message instanceof ResumeNotifications) {
            onNotificationsResumed();
        } else {
            super.onReceive(message);
        }
    }
271209
private void readAsync() throws IOException {
    stateChangeLock.lock();
    final byte[] arr = readAheadBuffer.array();
    try {
      if (endOfStream || readInProgress) {
        return;
      }
      checkReadException();
      readAheadBuffer.position(0);
      readAheadBuffer.flip();
      readInProgress = true;
    } finally {
      stateChangeLock.unlock();
    }
    executorService.execute(() -> {
      stateChangeLock.lock();
      try {
        if (isClosed) {
          readInProgress = false;
          return;
        }
        // Flip this so that the close method will not close the underlying input stream when we
        // are reading.
        isReading = true;
      } finally {
        stateChangeLock.unlock();
      }

      // Please note that it is safe to release the lock and read into the read ahead buffer
      // because either of following two conditions will hold - 1. The active buffer has
      // data available to read so the reader will not read from the read ahead buffer.
      // 2. This is the first time read is called or the active buffer is exhausted,
      // in that case the reader waits for this async read to complete.
      // So there is no race condition in both the situations.
      int read = 0;
      int off = 0, len = arr.length;
      Throwable exception = null;
      try {
        // try to fill the read ahead buffer.
        // if a reader is waiting, possibly return early.
        do {
          read = underlyingInputStream.read(arr, off, len);
          if (read <= 0) break;
          off += read;
          len -= read;
        } while (len > 0 && !isWaiting.get());
      } catch (Throwable ex) {
        exception = ex;
        if (ex instanceof Error) {
          // `readException` may not be reported to the user. Rethrow Error to make sure at least
          // The user can see Error in UncaughtExceptionHandler.
          throw (Error) ex;
        }
      } finally {
        stateChangeLock.lock();
        readAheadBuffer.limit(off);
        if (read < 0 || (exception instanceof EOFException)) {
          endOfStream = true;
        } else if (exception != null) {
          readAborted = true;
          readException = exception;
        }
        readInProgress = false;
        signalAsyncReadComplete();
        stateChangeLock.unlock();
        closeUnderlyingInputStreamIfNecessary();
      }
    });
  }
1135011
private VirtualConnection startEarlyRead(InterChannelCallback cb, boolean forceQueue) {
        // disallow rewrites once we start mixing the request and temp responses
        getLink().disallowRewrites();
        setAppReadCallback(cb);

        // check for an existing final response
        if (headersParsed() && !getResponseImpl().isTemporaryStatusCode()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "earlyRead: Final response already received.");
            }
            if (forceQueue) {
                cb.complete(getVC());
                return null;
            }
            return getVC();
        }

        // otherwise, if a message is already parsed, then the assumption is
        // that the app channel above has already looked at it and is now asking
        // for the next, so clear out the previous message and start a read
        if (headersParsed()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "earlyRead: Message exists, isOwner: " + isResponseOwner());
            }
            resetMsgParsedState();
            if (!isResponseOwner()) {
                // null our reference as not being owner means another SC has this
                // message already (i.e. the proxy passed it to the inbound
                // side) and then create the new object
                setMyResponse(null);
                getResponseImpl();
            } else {
                // otherwise clear the existing response
                getResponseImpl().clear();
            }
        }
        // start the read/parse cycle for the next response message
        VirtualConnection vc = parseResponseMessageAsync();
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "earlyRead: return vc=" + vc);
        }
        if (null != vc && forceQueue) {
            cb.complete(getVC());
            return null;
        }
        return vc;
    }
1020718
def _on_response_message(self, sequence, topic, message):
        

        try:
            conn_key = self._find_connection(topic)
            context = self.conns.get_context(conn_key)
        except ArgumentError:
            self._logger.warn("Dropping message that does not correspond with a known connection, message=%s", message)
            return

        if 'client' in message and message['client'] != self.name:
            self._logger.debug("Dropping message that is for another client %s, we are %s", message['client'], self.name)

        if messages.DisconnectionResponse.matches(message):
            self.conns.finish_disconnection(conn_key, message['success'], message.get('failure_reason', None))
        elif messages.OpenInterfaceResponse.matches(message):
            self.conns.finish_operation(conn_key, message['success'], message.get('failure_reason', None))
        elif messages.RPCResponse.matches(message):
            rpc_message = messages.RPCResponse.verify(message)
            self.conns.finish_operation(conn_key, rpc_message['success'], rpc_message.get('failure_reason', None), rpc_message.get('status', None), rpc_message.get('payload', None))
        elif messages.ProgressNotification.matches(message):
            progress_callback = context.get('progress_callback', None)
            if progress_callback is not None:
                progress_callback(message['done_count'], message['total_count'])
        elif messages.ScriptResponse.matches(message):
            if 'progress_callback' in context:
                del context['progress_callback']

            self.conns.finish_operation(conn_key, message['success'], message.get('failure_reason', None))
        elif messages.DisconnectionNotification.matches(message):
            try:
                conn_key = self._find_connection(topic)
                conn_id = self.conns.get_connection_id(conn_key)
            except ArgumentError:
                self._logger.warn("Dropping disconnect notification that does not correspond with a known connection, topic=%s", topic)
                return

            self.conns.unexpected_disconnect(conn_key)
            self._trigger_callback('on_disconnect', self.id, conn_id)
        else:
            self._logger.warn("Invalid response message received, message=%s", message)
218311
@Override
    protected void readResponse(HttpState state, HttpConnection conn)
    throws IOException, HttpException {
        LOG.trace("enter HttpMethodBase.readResponse(HttpState, HttpConnection)");
        
        boolean isUpgrade = false;
        
		while (getStatusLine() == null) {
            readStatusLine(state, conn);
            processStatusLine(state, conn);
            readResponseHeaders(state, conn);
            processResponseHeaders(state, conn);
            
            int status = this.statusLine.getStatusCode();
            if (status == 101) {
            	LOG.debug("Retrieved HTTP status code '101 Switching Protocols'. Keep connection open!");

            	// This means the requester has asked the server to switch protocols
            	// and the server is acknowledging that it will do so
            	// e.g.: upgrade to websocket
            	
				if (conn instanceof ZapHttpConnection) {
	            	isUpgrade = true;
	            	// avoid connection release of HttpClient library 
	            	conn.setHttpConnectionManager(null);
	            }
            } else if ((status >= 100) && (status < 200)) {
                if (LOG.isInfoEnabled()) {
                    LOG.info("Discarding unexpected response: " + this.statusLine.toString()); 
                }
                this.statusLine = null;
            }
        }

    	// get socket and input stream out of HttpClient
		if (conn instanceof ZapHttpConnection) {
			ZapHttpConnection zapConn = (ZapHttpConnection) conn;
	    	upgradedSocket = zapConn.getSocket();
			inputStream = zapConn.getResponseInputStream();
		}
		
		if (!isUpgrade) {
			// read & process rest of response
			// only if connection should not be kept
	        readResponseBody(state, conn);        
	        processResponseBody(state, conn);
		}
    }
638270
public boolean nextInput() throws IOException
    {
        if (!options.contains(READ) && !inflater.finished())
        {
            throw new IllegalStateException("should be called after read returns -1");
        }
        if (!compBuf.hasRemaining())
        {
            compBuf.clear();
            int rc = channel.read(compBuf);
            if (rc == -1)
            {
                return false;
            }
            compBuf.flip();
        }
        readHeader();
        inflater.reset();
        crc32.reset();
        return true;
    }
1269708
def on_message_received(self, msg):
        
        if self.is_stopped:
            raise RuntimeError("reader has already been stopped")
        else:
            self.buffer.put(msg)
826666
def handle_read(self, *args, **kwargs):
        
        self.read_events()
        self.process_events()
        if self.handle_read_callback is not None:
            self.handle_read_callback(self)
979143
function(result) {
        // start the previous reader listing again
        Espruino.Core.Serial.startListening(prevReader);
        // forward the original text to the previous reader
        if(prevReader) prevReader(receivedData);
        // run the callback
        callback(result);
      }