*/
  override fun cancel() {
    if (canceled) return // Already canceled.

    canceled = true
    exchange?.cancel()
    for (plan in plansToCancel) {
      plan.cancel()
    }

    eventListener.canceled(this)
  }

  override fun isCanceled(): Boolean = canceled

  override fun execute(): Response {

      val reader = ServerSentEventReader(body.source(), this)
      try {
        if (!canceled) {
          listener.onOpen(this, response)
          while (!canceled && reader.processNextEvent()) {
          }
        }
      } catch (e: Exception) {
        val exception =
          when {
            canceled -> IOException("canceled", e)
            else -> e
          }
        listener.onFailure(this, exception, response)

    // We may have been asked to cancel while creating the new stream and sending the request
    // headers, but there was still no stream to close.
    if (canceled) {
      stream!!.closeLater(ErrorCode.CANCEL)
      throw IOException("Canceled")
    }
    stream!!.readTimeout().timeout(chain.readTimeoutMillis.toLong(), TimeUnit.MILLISECONDS)

    assertThat(client.canceled).isFalse()

    taskFaker.advanceUntil(ns(RealWebSocket.CANCEL_AFTER_CLOSE_MILLIS - 1))
    assertThat(client.canceled).isFalse()

    taskFaker.advanceUntil(ns(RealWebSocket.CANCEL_AFTER_CLOSE_MILLIS))
    assertThat(client.canceled).isTrue()

    client.processNextFrame() // This won't get a frame, but it will get a closed pipe.

  internal val connectionSpecIndex: Int,
  internal val isTlsFallback: Boolean,
) : RoutePlanner.Plan,
  ExchangeCodec.Carrier {
  /** True if this connect was canceled; typically because it lost a race. */
  @Volatile private var canceled = false

  // These properties are initialized by connect() and never reassigned.

  /** The low-level TCP socket. */
  private var rawSocket: JavaNetSocket? = null

  /**

	for _, err := range errs {
		if IsErrIgnored(err, ignoredErrs...) {
			continue
		}
		// Errors due to context cancellation may be wrapped - group them by context.Canceled.
		if errors.Is(err, context.Canceled) {
			errorCounts[context.Canceled]++
			continue
		}
		errorCounts[err]++
	}

	maxCnt := 0
	for err, count := range errorCounts {
		switch {
		case maxCnt < count:
			maxCnt = count

	TotalSize   int64     `json:"totalSize" msg:"ts"`
	CurrentSize int64     `json:"currentSize" msg:"cs"`

	Complete bool `json:"complete" msg:"cmp"`
	Failed   bool `json:"failed" msg:"fl"`
	Canceled bool `json:"canceled" msg:"cnl"`

	// Internal information.
	QueuedBuckets         []string `json:"-" msg:"bkts"`
	DecommissionedBuckets []string `json:"-" msg:"dbkts"`

	// Last bucket/object decommissioned.

// of errors into a single maximal error with in the list.
func TestReduceErrs(t *testing.T) {
	canceledErrs := make([]error, 0, 5)
	for i := range 5 {
		canceledErrs = append(canceledErrs, fmt.Errorf("error %d: %w", i, context.Canceled))
	}
	// List all of all test cases to validate various cases of reduce errors.
	testCases := []struct {
		errs        []error
		ignoredErrs []error
		err         error
	}{
		// Validate if have reduced properly.

// to the options or to the current bucket ILM expiry rules.
// Caller should close the channel when done.
// The returned function will return the results once there is enough or input is closed,
// or the context is canceled.
func (o *listPathOptions) gatherResults(ctx context.Context, in <-chan metaCacheEntry) func() (metaCacheEntriesSorted, error) {
	resultsDone := make(chan metaCacheEntriesSorted)
	// Copy so we can mutate

				info, err := clnt.PutObject(uploadsCtx, opts.bucketName, tmpObjName, reader, int64(opts.objectSize), popts)
				if err != nil {
					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
						errOnce.Do(func() {
							retError = err.Error()
						})
					}
					uploadsCancel()
					return
				}
				response := time.Since(t)