request: Request,
    duplex: Boolean,
  ): Sink {
    this.isDuplex = duplex
    val contentLength = request.body!!.contentLength()
    call.eventListener.requestBodyStart(call)
    val rawRequestBody = codec.createRequestBody(request, contentLength)
    return RequestBodySink(
      delegate = rawRequestBody,
      contentLength = contentLength,
      isSocket = false,
    )
  }

  @Throws(IOException::class)

    sizeMapper: (T) -> Int,
  ): T {
    val contentLength = contentLength()
    if (contentLength > Int.MAX_VALUE) {
      throw IOException("Cannot buffer entire body for content length: $contentLength")
    }

    val bytes = source().use(consumer)
    val size = sizeMapper(bytes)
    if (contentLength != -1L && contentLength != size.toLong()) {

		}
	}
}

// Provides a fully populated http request instance, fails otherwise.
func mustNewRequest(method string, urlStr string, contentLength int64, body io.ReadSeeker, t *testing.T) *http.Request {
	req, err := newTestRequest(method, urlStr, contentLength, body)
	if err != nil {
		t.Fatalf("Unable to initialize new http request %s", err)
	}
	return req
}

  fun stringClosesUnderlyingSource() {
    val closed = AtomicBoolean()
    val body: ResponseBody =
      object : ResponseBody() {
        override fun contentType(): MediaType? = null

        override fun contentLength(): Long = 5

        override fun source(): BufferedSource {
          val source = Buffer().writeUtf8("hello")
          return object : ForwardingSource(source) {
            @Throws(IOException::class)

      }

      // We can't measure the body's size without the sizes of its components.
      val contentLength = body.contentLength()
      if (contentLength == -1L && countBytes) {
        byteCountBuffer!!.clear()
        return -1L
      }

      sink.write(CRLF)

      if (countBytes) {
        byteCount += contentLength
      } else {
        body.writeTo(sink)
      }

      sink.write(CRLF)
    }

      }

      response.isChunked -> {
        newChunkedSource(response.request.url)
      }

      else -> {
        val contentLength = response.headersContentLength()
        if (contentLength != -1L) {
          newFixedLengthSource(response.request.url, contentLength)
        } else {
          newUnknownLengthSource(response.request.url)
        }
      }
    }

                  }
                }
                println(bytesWritten)
                if (contentLength != -1L) {
                  println("${100 * bytesWritten / contentLength}% done")
                }
              }
            }
          }

          val file = File("docs/images/logo-square.png")
          val requestBody: RequestBody = file.asRequestBody(MEDIA_TYPE_PNG)

        throw e
      }

      val tookMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs)

      val responseBody = response.body!!
      val contentLength = responseBody.contentLength()
      val bodySize = if (contentLength != -1L) "$contentLength-byte" else "unknown-length"
      logger.log(
        buildString {
          append("<-- ${response.code}")

            expectedByteCount = contentLength,
            socket = socket,
          ).buffer()
      requestBodySink.use {
        when {
          peek.doNotReadRequestBody -> {
            hasBody = false // Ignore the body completely.
          }

          contentLength != -1L -> {
            hasBody = contentLength > 0L || HttpMethod.permitsRequestBody(request.method)

     * Fields to fetch for access result list without data.
     */
    private static final String[] ACCESS_RESULT_FIELDS = { "parentUrl", "method", "mimeType", "sessionId", "url", "executionTime",
            "createTime", "contentLength", "lastModified", "ruleId", "httpStatusCode", "status" };

    /**
     * Creates a new instance of OpenSearchDataService.
     * @param crawlerConfig The crawler configuration.
     */