}

    @Override
    public synchronized Map<String, Object> read() {
        while (!isFinished.get() && queue.isEmpty()) {
            addDocumentToQueue();
        }

        return queue.poll();
    }

    @Override
    public void close() {
        isFinished.set(true);
        queue.clear();
    }

    /**
     * Sets the scroll size.

        }

        /**
         * Checks if the crawling thread has finished execution.
         *
         * @return true if the thread has completed its crawling operation
         */
        public boolean isFinished() {
            return finished;
        }

        /**
         * Stops the crawling operation gracefully.
         * If a data store is currently active, this method calls

      if (inFinished) {
        dataStream.receiveHeaders(Headers.EMPTY, true)
      }
    }

    override fun headers(
      inFinished: Boolean,
      streamId: Int,
      associatedStreamId: Int,
      headerBlock: List<Header>,
    ) {
      if (pushedStream(streamId)) {
        pushHeadersLater(streamId, headerBlock, inFinished)
        return
      }

    assertThat(data1.streamId).isEqualTo(3)
    assertArrayEquals("abcdefghi".toByteArray(), data1.data)
    assertThat(data1.inFinished).isFalse()
    val headers2 = peer.takeFrame()
    assertThat(headers2.type).isEqualTo(Http2.TYPE_HEADERS)
    assertThat(headers2.inFinished).isTrue()
  }

  @Test fun clientCannotReadTrailersWithoutExhaustingStream() {
    // Write the mocking script.