private fun serviceStreamTask(stream: Stream): FutureTask<Void> {
    return FutureTask<Void> {
      stream.requestBody.use {
        stream.responseBody.use {
          while (true) {
            val action = actions.poll() ?: break
            action(stream)
          }
        }
      }
      return@FutureTask null
    }
  }

      val existingTask = activeTask
      if (existingTask is AwaitIdleTask) {
        return existingTask.latch
      }
      for (futureTask in futureTasks) {
        if (futureTask is AwaitIdleTask) {
          return futureTask.latch
        }
      }

      // Don't delegate to schedule() because that enforces shutdown rules.
      val newTask = AwaitIdleTask()