Parses the next response in the queue if a data class and a {@link BatchCallback} is specified.

<p>
This method closes the input stream if there are no more individual responses left.
</p>
974797
void parseNextResponse() throws IOException {
    contentId++;

    // Extract the outer headers.
    String line;
    while ((line = readLine()) != null && !line.equals("")) {
      // Do nothing.
    }

    // Extract the status code.
    String statusLine = readLine();
    String[] statusParts = statusLine.split(" ");
    int statusCode = Integer.parseInt(statusParts[1]);

    // Extract and store the inner headers.
    // TODO(rmistry): Handle inner headers that span multiple lines. More details here:
    // http://tools.ietf.org/html/rfc2616#section-2.2
    List<String> headerNames = new ArrayList<String>();
    List<String> headerValues = new ArrayList<String>();
    long contentLength = -1L;
    while ((line = readLine()) != null && !line.equals("")) {
      String[] headerParts = line.split(": ", 2);
      String headerName = headerParts[0];
      String headerValue = headerParts[1];
      headerNames.add(headerName);
      headerValues.add(headerValue);
      if ("Content-Length".equalsIgnoreCase(headerName.trim())) {
          contentLength = Long.parseLong(headerValue);
      }
    }

    InputStream body;
    if (contentLength == -1) {
      // This isn't very efficient, but most respectable servers should set the Content-Length
      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
      while ((line = readRawLine()) != null && !line.startsWith(boundary)) {
        // Convert characters back to bytes:
        buffer.write(line.getBytes("ISO-8859-1"));
      }

      // Remove CRLF that separates body from boundary token
      body = trimCrlf(buffer.toByteArray());

      // Remove CRLF from the boundary token (to match readLine)
      line = trimCrlf(line);
    } else {
      body = new FilterInputStream(ByteStreams.limit(inputStream, contentLength)) {
        @Override
        public void close() {
          // Don't allow the parser to close the underlying stream
        }
      };
    }

    HttpResponse response =
        getFakeResponse(statusCode, body, headerNames, headerValues);

    parseAndCallback(requestInfos.get(contentId - 1), statusCode, response);

    // Consume any bytes that were not consumed by the parser
    while (body.skip(contentLength) > 0 || body.read() != -1) {
    }

    if (contentLength != -1) {
      line = readLine();
    } else {
      // The line was already read
    }
    // Consume any blank lines that follow the response (not included in Content-Length)
    while ((line != null) && (line.length() == 0)) {
      line = readLine();
    }

    checkForFinalBoundary(line);
  }
252450
private void doRead() {
        if (context == null) {
            context = vertx.getOrCreateContext();
        }
        if (state == STATUS_ACTIVE) {
            final Handler<Buffer> dataHandler = this.dataHandler;
            final Handler<Void> closeHandler = this.closeHandler;
            executor.submit(
                    (Runnable) () -> {
                        try {
                            final byte[] bytes = readChunk();

                            if (bytes == null || bytes.length == 0) {
                                // null or 0 means we reach the end of the stream, invoke the close handler.
                                state = STATUS_CLOSED;
                                IOUtils.closeQuietly(in);
                                context.runOnContext(event -> {
                                    if (closeHandler != null) {
                                        closeHandler.handle(null);
                                    }
                                });
                            } else {
                                // We still have data, dispatch it.
                                context.runOnContext(event -> {
                                    dataHandler.handle(Buffer.buffer(bytes));
                                    // The next chunk will be read in another call, and maybe another thread.
                                    // As the data was already given to the data handler, this is fine.
                                    doRead();
                                });
                            }
                        } catch (final Exception e) {
                            // Error detected, invokes the failure handler.
                            state = STATUS_CLOSED;
                            IOUtils.closeQuietly(in);
                            /**
                             * Invokes the failure handler.
                             * @param event irrelevant
                             */
                            context.runOnContext(event -> {
                                if (failureHandler != null) {
                                    failureHandler.handle(e);
                                }
                            });
                        }
                    });
        }
    }
264831
@Override
	public List<MemorySegment> close() throws IOException {
		if (this.closed) {
			throw new IllegalStateException("Already closed.");
		}
		this.closed = true;

		// re-collect all memory segments
		ArrayList<MemorySegment> list = this.freeMem;
		final MemorySegment current = getCurrentSegment();
		if (current != null) {
			list.add(current);
		}
		clear();

		// close the writer and gather all segments
		final LinkedBlockingQueue<MemorySegment> queue = this.reader.getReturnQueue();
		this.reader.close();

		while (list.size() < this.numSegments) {
			final MemorySegment m = queue.poll();
			if (m == null) {
				// we get null if the queue is empty. that should not be the case if the reader was properly closed.
				throw new RuntimeException("Bug in ChannelReaderInputView: MemorySegments lost.");
			}
			list.add(m);
		}
		return list;
	}
811009
public Future<WriteResponse> write (Batch<String> batch, WriteCallback callback) {
    Timer.Context context = writeTimer.time();
    int returnCode = 0;
    LOG.info ("Dispatching batch " + batch.getId());
    recordsAttempted.mark(batch.getRecords().size());
    try {
      String encoded = encodeBatch(batch);
      returnCode = request (encoded);
      WriteResponse<Integer> response = WRITE_RESPONSE_WRAPPER.wrap(returnCode);
      callback.onSuccess(response);
      bytesWritten.mark(encoded.length());
      recordsSuccess.mark(batch.getRecords().size());
    } catch (Exception e) {
      LOG.error("Dispatching batch " + batch.getId() + " failed :" + e.toString());
      callback.onFailure(e);
      recordsFailed.mark(batch.getRecords().size());
    }

    context.close();
    Future<Integer> future = Futures.immediateFuture(returnCode);
    return new WriteResponseFuture<>(future, WRITE_RESPONSE_WRAPPER);
  }
1259416
public ChannelFuture respond() throws Exception {
        // For chunked response, this method is only called to respond the 1st chunk,
        // next chunks are responded directly by respondXXX

        if (nonChunkedResponseOrFirstChunkSent) throwDoubleResponseError();

        // Run after filter
        if (server.after() != null) {
            server.after().run(request, this);
        }

        ChannelFuture future = channel.writeAndFlush(response);

        // Do not handle keep alive:
        // * If XSendFile or XSendResource is used, because they will handle keep alive in their own way
        // * If the response is chunked, because respondLastChunk will be handle keep alive
        if (//!XSendFile.isHeaderSet(response) &&
            //!XSendResource.isHeaderSet(response) &&
                !HttpUtil.isTransferEncodingChunked(response)) {
            NoRealPipelining.if_keepAliveRequest_then_resumeReading_else_closeOnComplete(request, channel, future);
        }

        nonChunkedResponseOrFirstChunkSent = true;
        if (!HttpUtil.isTransferEncodingChunked(response)) {
            doneResponding = true;
        }

        return future;
    }
465714
def next(self) -> mx.io.DataBatch:
        
        if self.iter_next():
            return self.next_batch
        raise StopIteration
270957
private void closeResponse(ClientResponse response) {
        if (response != null) {
            try {
                response.close();
            } catch (Throwable th) {
                logger.error("Cannot release response resource :", th);
            }
        }
    }
444014
@Override
  public void onCompleted() {
    RpcUtils.streamingRPCAndLog(LOG, new RpcUtils.StreamingRpcCallable<OpenLocalBlockResponse>() {
      @Override
      public OpenLocalBlockResponse call() throws Exception {
        if (mLockId != BlockLockManager.INVALID_LOCK_ID) {
          mWorker.unlockBlock(mLockId);
          mLockId = BlockLockManager.INVALID_LOCK_ID;
        } else if (mRequest != null) {
          LOG.warn("Close a closed block {}.", mRequest.getBlockId());
        }
        return null;
      }

      @Override
      public void exceptionCaught(Throwable e) {
        mResponseObserver.onError(GrpcExceptionUtils.fromThrowable(e));
        mLockId = BlockLockManager.INVALID_LOCK_ID;
      }
    }, "CloseBlock", false, true, mResponseObserver, "Session=%d, Request=%s",
        mSessionId, mRequest);
  }
896997
public void finishStage(ResponseBuilder rb) {
    // System.out.println(
    // "collection: " + System.nanoTime() + " - " +
    // Thread.currentThread().getId()
    // + " - " + rb.req.getParams().getBool("isShard", false)
    // + " FINISHSTAGE " + rb.stage + " " + rb.req.getParamString());
    if (rb.req.getParams().getBool(MtasSolrSearchComponent.PARAM_MTAS, false)) {
      if (rb.stage >= ResponseBuilder.STAGE_EXECUTE_QUERY
          && rb.stage < ResponseBuilder.STAGE_GET_FIELDS) {
        ComponentFields mtasFields = getMtasFields(rb);
        if (mtasFields.doCollection) {
          if (rb.stage == ResponseBuilder.STAGE_EXECUTE_QUERY) {
            // mtas response
            NamedList<Object> mtasResponse = null;
            try {
              mtasResponse = (NamedList<Object>) rb.rsp.getValues()
                  .get(MtasSolrSearchComponent.NAME);
            } catch (ClassCastException e) {
              log.debug(e);
              mtasResponse = null;
            }
            if (mtasResponse == null) {
              mtasResponse = new SimpleOrderedMap<>();
              rb.rsp.add(MtasSolrSearchComponent.NAME, mtasResponse);
            }
            ArrayList<Object> mtasCollectionResponses;
            if (mtasResponse.get(NAME) != null
                && mtasResponse.get(NAME) instanceof ArrayList) {
              mtasCollectionResponses = (ArrayList<Object>) mtasResponse
                  .get(NAME);
            } else {
              mtasCollectionResponses = new ArrayList<>();
              mtasResponse.add(NAME, mtasCollectionResponses);
            }
            MtasSolrCollectionResult collectionResult;
            for (ComponentCollection componentCollection : mtasFields.collection) {
              try {
                collectionResult = createMtasSolrCollectionResult(
                    componentCollection, false);
                // Create response
                SimpleOrderedMap<Object> mtasCollectionResponse = new SimpleOrderedMap<>();
                mtasCollectionResponse.add("key", componentCollection.key);
                mtasCollectionResponse.add("data", collectionResult);
                mtasCollectionResponses.add(mtasCollectionResponse);
              } catch (IOException e) {
                log.debug(e);
              }
            }
          }
          // decode shard responses
          for (ShardRequest sreq : rb.finished) {
            if (sreq.params.getBool(MtasSolrSearchComponent.PARAM_MTAS, false)
                && sreq.params.getBool(PARAM_MTAS_COLLECTION, false)) {
              for (ShardResponse shardResponse : sreq.responses) {
                NamedList<Object> solrShardResponse = shardResponse
                    .getSolrResponse().getResponse();
                try {
                  ArrayList<SimpleOrderedMap<Object>> data = (ArrayList<SimpleOrderedMap<Object>>) solrShardResponse
                      .findRecursive(MtasSolrSearchComponent.NAME, NAME);
                  if (data != null) {
                    MtasSolrResultUtil.decode(data);
                    if (rb.stage > ResponseBuilder.STAGE_EXECUTE_QUERY) {
                      ArrayList<SimpleOrderedMap<Object>> filteredData = new ArrayList<>();
                      for (SimpleOrderedMap<Object> dataItem : data) {
                        if (dataItem.get("data") != null && dataItem
                            .get("data") instanceof MtasSolrCollectionResult) {
                          MtasSolrCollectionResult collectionResult = (MtasSolrCollectionResult) dataItem
                              .get("data");
                          if (rb.stage <= MtasSolrSearchComponent.STAGE_COLLECTION_INIT) {
                            if (!collectionResult.action()
                                .equals(ComponentCollection.ACTION_CREATE)
                                && !collectionResult.action()
                                    .equals(ComponentCollection.ACTION_LIST)
                                && !collectionResult.action()
                                    .equals(ComponentCollection.ACTION_CHECK)) {
                              filteredData.add(dataItem);
                            }
                          } else if (rb.stage <= MtasSolrSearchComponent.STAGE_COLLECTION_FINISH) {
                            if (!collectionResult.action()
                                .equals(ComponentCollection.ACTION_POST)
                                && !collectionResult.action()
                                    .equals(ComponentCollection.ACTION_IMPORT)
                                && !collectionResult.action()
                                    .equals(ComponentCollection.ACTION_CHECK)) {
                              filteredData.add(dataItem);
                            }
                          }
                        } else {
                          filteredData.add(dataItem);
                        }
                      }
                      data.clear();
                      data.addAll(filteredData);
                    }
                  }
                } catch (ClassCastException e) {
                  log.debug(e);
                  // shouldn't happen
                }
              }
            }
          }
        }
      }
    }
  }
272455
@Override
    public List<Writable> next() {
        List<String> next = dataIter.next();
        invokeListeners(next);
        List<Writable> ret = new ArrayList<>();
        for (String s : next)
            ret.add(new Text(s));
        return ret;
    }

    /**
     * Check whether there are anymore records
     *
     * @return Whether there are more records
     */
    @Override
    public boolean hasNext() {
        return dataIter.hasNext();
    }

    /**
     * List of label strings
     *
     * @return
     */
    @Override
    public List<String> getLabels() {
        return null;
    }

    /**
     * Reset record reader iterator
     *
     */
    @Override
    public void reset() {
        dataIter = delimitedData.iterator();
    }

    @Override
    public boolean resetSupported() {
        return true;
    }

    /**
     * Load the record from the given DataInputStream
     * Unlike {@link #next()} the internal state of the RecordReader is not modified
     * Implementations of this method should not close the DataInputStream
     *
     * @param uri
     * @param dataInputStream
     * @throws IOException if error occurs during reading from the input stream
     */
    @Override
    public List<Writable> record(URI uri, DataInputStream dataInputStream) throws IOException {
        return null;
    }

    @Override
    public Record nextRecord() {
        return new org.datavec.api.records.impl.Record(next(), null);
    }

    @Override
    public Record loadFromMetaData(RecordMetaData recordMetaData) throws IOException {
        throw new UnsupportedOperationException("Loading from metadata not yet implemented");
    }

    @Override
    public List<Record> loadFromMetaData(List<RecordMetaData> recordMetaDatas) throws IOException {
        throw new UnsupportedOperationException("Loading from metadata not yet implemented");
    }

    /**
     * Closes this stream and releases any system resources associated
     * with it. If the stream is already closed then invoking this
     * method has no effect.
     * <p>
     * <p> As noted in {@link AutoCloseable#close()}, cases where the
     * close may fail require careful attention. It is strongly advised
     * to relinquish the underlying resources and to internally
     * <em>mark</em> the {@code Closeable} as closed, prior to throwing
     * the {@code IOException}.
     *
     * @throws IOException if an I/O error occurs
     */
    @Override
    public void close() throws IOException {

    }

    /**
     * Set the configuration to be used by this object.
     *
     * @param conf
     */
    @Override
    public void setConf(Configuration conf) {
        this.conf = conf;
    }

    /**
     * Return the configuration used by this object.
     */
    @Override
    public Configuration getConf() {
        return conf;
    }
499456
@Override
  public void processEvent(QueryBatch batch) {
    try {
      batch.getClient().newDocumentManager().delete( batch.getItems() );
    } catch (Throwable t) {
      for ( BatchFailureListener<Batch<String>> listener : failureListeners ) {
        try {
          listener.processFailure(batch, t);
        } catch (Throwable t2) {
          logger.error("Exception thrown by an onBatchFailure listener", t2);
        }
      }
      for ( BatchFailureListener<QueryBatch> queryBatchFailureListener : queryBatchFailureListeners ) {
        try {
          queryBatchFailureListener.processFailure(batch, t);
        } catch (Throwable t2) {
          logger.error("Exception thrown by an onFailure listener", t2);
        }
      }
    }
  }
345369
def _dequeue_batch(self) -> Optional[Batch]:
        
        if self._enqueueing_thread is None:
            raise ValueError('StreamWrapper `{}` with buffer of size `{}` was used outside with-resource environment.'
                             .format(self._name, self._buffer_size))
        if not self._enqueueing_thread.is_alive() and self._queue.empty():
            self._start_thread()
        while True:
            try:
                batch = self._queue.get(timeout=2)
                self._queue.task_done()
                break
            except Empty:
                if not self._enqueueing_thread.is_alive():
                    try:
                        # the enqueueing thread may just finished properly so lets check the queue eagerly
                        batch = self._queue.get_nowait()
                        self._queue.task_done()
                        break
                    except Empty:
                        # so we failed to retrieve a batch and the enqueueing thread is dead
                        # there is no hope, something must went wrong
                        raise ChildProcessError('Enqueueing thread ended unexpectedly.')
        return batch
729298
public void batchCommitted()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "batchCommitted");
      
    if (batchSendList.size() != 0)
    {
      try
      {
        sendMsgs(batchSendList, false);
      }
      catch (SIResourceException e)
      {
        //FFDC
        FFDCFilter.processException(
          e,
          "com.ibm.ws.sib.processor.gd.SourceStream.batchCommitted",
          "1:2590:1.138",
          this);

        SibTr.exception(tc, e);
      }
    }
    
    // Empty list ready for use by next batch
    batchList.clear();
    batchSendList.clear();
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "batchCommitted");
  }
568578
@Override
    public void close() {
        if (!closed.compareAndSet(false, true))
            return;

        //is synchronized with itself and _next calls,
        //so no synchronization on inner reader is needed
        try {
            synchronized (inputStream) {
                inputStream.close();
            }
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }
1181928
void onSealExclusive(Batch batch, long elapsedTimeMillis)
  {
    batchFillingTimeCounter.add((int) Math.max(elapsedTimeMillis, 0));
    if (elapsedTimeMillis > 0) {
      // If elapsedTimeMillis is 0 or negative, it's likely because System.currentTimeMillis() is not monotonic, so not
      // accounting this time for determining batch sending timeout.
      lastFillTimeMillis = elapsedTimeMillis;
    }
    addBatchToEmitQueue(batch);
    wakeUpEmittingThread();
    if (!isTerminated()) {
      int nextBatchNumber = EmittedBatchCounter.nextBatchNumber(batch.batchNumber);
      if (!concurrentBatch.compareAndSet(batch, new Batch(this, acquireBuffer(), nextBatchNumber))) {
        // If compareAndSet failed, the service is closed concurrently.
        Preconditions.checkState(isTerminated());
      }
    }
  }
605738
private void completeResponses() {
    // Iterate through queued responses and complete as many as possible.
    ResponseCallback response = responseCallbacks.get(responseSequence + 1);
    while (response != null) {
      // If the response was completed, remove the response callback from the response queue,
      // increment the response sequence number, and check the next response.
      if (completeResponse(response.response, response.callback)) {
        responseCallbacks.remove(++responseSequence);
        response = responseCallbacks.get(responseSequence + 1);
      } else {
        break;
      }
    }

    // Once we've completed as many responses as possible, if no more operations are outstanding
    // and events remain in the event queue, complete the events.
    if (requestSequence == responseSequence) {
      EventCallback eventCallback = eventCallbacks.poll();
      while (eventCallback != null) {
        log.trace("Completing {}", eventCallback.request);
        eventCallback.run();
        eventIndex = eventCallback.request.eventIndex();
        eventCallback = eventCallbacks.poll();
      }
    }
  }
1199512
@Override
  public synchronized void close() {
    // try to write everything queued / buffered
    for (BatchPoints points : batchQueue) {
      WriteResult result = tryToWrite(points);
      if (result.outcome != WriteResultOutcome.WRITTEN) {
        exceptionHandler.accept(points.getPoints(), result.throwable);
      }
    }
  }