}
		r = gzr
		pr.closer = gzr
	case bzip2Type:
		ctx, cancel := context.WithCancel(context.Background())
		r = pbzip2.NewReader(ctx, scannedReader, pbzip2.DecompressionOptions(
			pbzip2.BZConcurrency((runtime.GOMAXPROCS(0)+1)/2),
			pbzip2.BZConcurrencyPool(bz2Limiter),
		))
		pr.closer = &nopReadCloser{fn: cancel}
	case zstdType:
		// Set a max window of 64MB. More than reasonable.

// are consistent with each other.
func TestListOnlineDisks(t *testing.T) {
	if runtime.GOOS == globalWindowsOSName {
		t.Skip()
	}

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	obj, disks, err := prepareErasure16(ctx)
	if err != nil {
		t.Fatalf("Prepare Erasure backend failed - %v", err)
	}
	setObjectLayer(obj)
	defer obj.Shutdown(t.Context())

          connectState = ConnectState.TLS_CONNECTED
          ConnectResult(this)
        }
      }
    }

    override fun handleSuccess() = connection

    override fun cancel() {
      events += "plan $id cancel"
      canceled = true
    }

    override fun retry(): FakePlan? {
      check(!retryTaken)
      retryTaken = true
      return retry
    }
  }

		return listener, listenErrs
	}
	listeners = slices.Clip(listeners)

	ctx, cancel := context.WithCancel(ctx)
	listener = &httpListener{
		listeners:   listeners,
		acceptCh:    make(chan acceptResult, len(listeners)),
		opts:        opts,
		ctxDoneCh:   ctx.Done(),
		ctxCanceler: cancel,
	}
	opts.Trace(fmt.Sprintf("opening %d listeners", len(listener.listeners)))
	listener.start()

        // for technical reasons we cannot throw here, even if delegate does cancel transfer
        listener.transferInitiated(event(session, resource, TransferEvent.EventType.INITIATED));

        Thread.sleep(500); // to make sure queue is processed, cancellation applied

        // subsequent call will cancel
        assertThrows(
                TransferCancelledException.class,

public abstract interface class okhttp3/sse/EventSource {
	public abstract fun cancel ()V
	public abstract fun request ()Lokhttp3/Request;
}

public abstract interface class okhttp3/sse/EventSource$Factory {
	public abstract fun newEventSource (Lokhttp3/Request;Lokhttp3/sse/EventSourceListener;)Lokhttp3/sse/EventSource;
}

public abstract class okhttp3/sse/EventSourceListener {
	public fun <init> ()V
	public fun onClosed (Lokhttp3/sse/EventSource;)V

	metaMu := &mc.mu
	rpc := mc.rpc
	cancel := mc.cancel
	defer func() {
		o.debugln(color.Green("saveMetaCacheStream:")+"err:", err)
		if err != nil && !errors.Is(err, io.EOF) {
			go mc.setErr(err.Error())
			cancel()
		}
	}()

	defer cancel()
	// Save continuous updates
	go func() {
		var err error
		ticker := time.NewTicker(10 * time.Second)

            - elastic
          allow-whitelist-orgs-as-admins: true
          trigger-phrase: '.*run\W+elasticsearch-ci/{pr-job}.*'
          github-hooks: true
          status-context: elasticsearch-ci/{pr-job}
          cancel-builds-on-update: true
          excluded-regions:
            - ^docs/.*
          black-list-labels:
            - '>test-mute'
    builders:
      - inject:
          properties-file: '.ci/java-versions.properties'

  protected ForwardingFuture() {}

  @Override
  protected abstract Future<? extends V> delegate();

  @Override
  @CanIgnoreReturnValue
  public boolean cancel(boolean mayInterruptIfRunning) {
    return delegate().cancel(mayInterruptIfRunning);
  }

  @Override
  public boolean isCancelled() {
    return delegate().isCancelled();
  }

  @Override
  public boolean isDone() {

          allow-whitelist-orgs-as-admins: true
          trigger-phrase: '.*run\W+elasticsearch-ci/part-2-fips.*'
          github-hooks: true
          status-context: elasticsearch-ci/part-2-fips
          cancel-builds-on-update: true
          black-list-target-branches:
            - 6.8
          excluded-regions:
            - ^docs/.*
          white-list-labels:
            - 'Team:Security'
          black-list-labels: