throw failure.get();
                    }
                } catch (UserInterruptException e) {
                    // Ignore
                    // return CANCELED;
                } catch (EndOfFileException e) {
                    return OK;
                } catch (SystemRegistryImpl.UnknownCommandException e) {
                    context.writer.accept(context.invokerRequest

        ).setHeader("content-type", "text/event-stream")
        .build(),
    )
    listener.enqueueCancel() // Will cancel in onOpen().
    newEventSource()
    listener.assertOpen()
    listener.assertFailure("canceled")
  }

  @Test
  fun badContentType() {
    server.enqueue(
      MockResponse
        .Builder()
        .body(
          """
          |data: hey
          |
          |

    override val call: Call,
    val ioe: IOException,
  ) : CallEvent() {
    override fun closes(event: CallEvent): Boolean = event is CallStart && call == event.call
  }

  data class Canceled(
    override val timestampNs: Long,
    override val call: Call,
  ) : CallEvent()

  data class RequestHeadersStart(
    override val timestampNs: Long,
    override val call: Call,
  ) : CallEvent()

		r.writeCond = sync.NewCond(&r.mu)
	}
	return r
}

// WithCancel sets a context to cancel the ring buffer.
// When the context is canceled, the ring buffer will be closed with the context error.
// A goroutine will be started and run until the provided context is canceled.
func (r *RingBuffer) WithCancel(ctx context.Context) *RingBuffer {
	go func() {
		<-ctx.Done()
		r.CloseWithError(ctx.Err())
	}()
	return r
}

=== ":material-language-kotlin: Kotlin"
    ```kotlin
      private val executor = Executors.newScheduledThreadPool(1)
      private val client = OkHttpClient()

      fun run() {

   * inbound data and send an early 400 before the request body completes.
   *
   * This test sends a slow request that is canceled by the server. It expects the response to still
   * be readable after the request stream is canceled.
   */
  @Disabled
  @Test
  fun serverCancelsRequestBodyAndSendsResponseBody() {
    client =
      client
        .newBuilder()

    }

    @Override
    public V get(long timeout, TimeUnit unit) {
      throw new AssertionFailedError();
    }

    @Override
    public boolean isCancelled() {
      throw new AssertionFailedError();
    }

    @Override
    public boolean isDone() {
      /*
       * If isDone is true during the call to listenInPoolThread,

func (c *kesConn) Status(ctx context.Context) (map[string]madmin.ItemState, error) {
	if len(c.client.Endpoints) == 1 {
		if _, err := c.client.Status(ctx); err != nil {
			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
				return nil, err
			}
			if errors.Is(err, kes.ErrNotAllowed) {
				return nil, ErrPermission
			}

			return map[string]madmin.ItemState{

    if (!newRouteSelector.hasNext()) throw IOException("exhausted all routes")
    val newRouteSelection = newRouteSelector.next()
    routeSelection = newRouteSelection

    if (isCanceled()) throw IOException("Canceled")

    return planConnectToRoute(newRouteSelection.next(), newRouteSelection.routes)
  }

  /**
   * Returns a plan to reuse a pooled connection, or null if the pool doesn't have a connection for

				if unlock, ok := val.(chan struct{}); ok && unlock != nil {
					// Wait until unfrozen.
					select {
					case <-unlock:
					case <-r.Context().Done():
						// if client canceled we don't need to wait here forever.
						return
					}
				}
			}
		}

		globalHTTPStats.addRequestsInQueue(1)
		pool := globalAPIConfig.getRequestsPool()
		if pool == nil {