this.source.finished = true
      }
      open = isOpen
      condition.signalAll()
    }
    if (!open) {
      connection.removeStream(id)
    }
  }

  fun receiveRstStream(errorCode: ErrorCode) {
    this.withLock {
      if (this.errorCode == null) {
        this.errorCode = errorCode
        condition.signalAll()
      }
    }
  }

  /**

                if (coordinatorTask != null) {
                  waitingCoordinatorNotified = true
                  currentTask = coordinatorTask
                  taskRunner.condition.signalAll()
                } else {
                  startNextTask()
                }
              }
            }
        }

        override fun coordinatorWait(
          taskRunner: TaskRunner,

    return delegate().awaitUntil(deadline);
  }

  @Override
  public void signal() {
    delegate().signal();
  }

  @Override
  public void signalAll() {
    delegate().signalAll();
  }

  internal fun removeStream(streamId: Int): Http2Stream? {
    this.withLock {
      val stream = streams.remove(streamId)

      // The removed stream may be blocked on a connection-wide window update.
      condition.signalAll()

      return stream
    }
  }

  internal fun updateConnectionFlowControl(read: Long) {
    this.withLock {
      readBytes.update(total = read)