       // Avoid CoW-ing 'state'.
       self = .idle
 
-      let filter: StreamState.Filter
+      let allow: Bool
 
       switch part {
       case .metadata:
-        filter = state.channelStreamState.sendHeaders()
+        allow = state.channelStreamState.sendHeaders()
         self = .active(state)
       case .message:
-        filter = state.channelStreamState.sendMessage()
+        allow = state.channelStreamState.sendMessage()
         self = .active(state)
       case .end:
-        filter = state.channelStreamState.sendEnd()
+        allow = state.channelStreamState.sendEnd()
         // We're sending end, we're no longer active.
         self = .closed
       }
 
-      switch filter {
-      case .allow:
-        return .writeResponsePartToChannel(state.context, part, promise: promise)
-      case .drop:
-        return .completePromise(promise, .failure(GRPCError.AlreadyComplete()))
-      }
+      return allow
 
     case .closed:
-      // We're already closed, fail any promise.
-      return .completePromise(promise, .failure(GRPCError.AlreadyComplete()))
+      // We're already closed.
+      return false
     }
   }
 
   /// A request part has traversed the interceptor pipeline, now send it to the observer.
-  internal mutating func receiveRequestPartFromInterceptors(
-    _ part: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>
-  ) -> Action {
+  /// - Returns: True if the part should be forwarded to the observer, false otherwise.
+  internal mutating func receiveRequestPartFromInterceptors(_ part: RPCStreamPart) -> Bool {
     switch self {
     case .idle:
       preconditionFailure("Invalid state: the handler isn't in the pipeline yet")

    case requestClosedResponseOpen
    case requestClosedResponseClosed

    enum Filter {
      case allow
      case drop
    }

    mutating func receiveHeaders() -> Filter {
      switch self {
      case .requestIdleResponseIdle:
        self = .requestOpenResponseIdle
        return .allow

      case .requestOpenResponseIdle,
           .requestOpenResponseOpen,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    func receiveMessage() -> Filter {
      switch self {
      case .requestOpenResponseIdle,
           .requestOpenResponseOpen:
        return .allow

      case .requestIdleResponseIdle,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    mutating func receiveEnd() -> Filter {
      switch self {
      case .requestOpenResponseIdle:
        self = .requestClosedResponseIdle
        return .allow

      case .requestOpenResponseOpen:
        self = .requestClosedResponseOpen
        return .allow

      case .requestIdleResponseIdle,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    mutating func sendHeaders() -> Filter {
      switch self {
      case .requestOpenResponseIdle:
        self = .requestOpenResponseOpen
        return .allow

      case .requestClosedResponseIdle:
        self = .requestClosedResponseOpen
        return .allow

      case .requestIdleResponseIdle,
           .requestOpenResponseOpen,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    func sendMessage() -> Filter {
      switch self {
      case .requestOpenResponseOpen,
           .requestClosedResponseOpen:
        return .allow

      case .requestIdleResponseIdle,
           .requestOpenResponseIdle,
           .requestClosedResponseIdle,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    mutating func sendEnd() -> Filter {
      switch self {
      case .requestIdleResponseIdle:
        return .drop

      case .requestOpenResponseIdle,
           .requestOpenResponseOpen,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen:
        self = .requestClosedResponseClosed
        return .allow

      case .requestClosedResponseClosed:
        return .drop
      }
    }
  }

  fileprivate struct ActiveState {
    var context: ChannelHandlerContext

    /// The stream state between the 'Channel' and interceptor pipeline.
    var channelStreamState: StreamState

    /// The stream state between the interceptor pipeline and event observer.
    var observerStreamState: StreamState

    init(context: ChannelHandlerContext) {
      self.context = context
      self.channelStreamState = .requestIdleResponseIdle
      self.observerStreamState = .requestIdleResponseIdle
    }
  }
}

extension _BaseCallHandler.State {
  fileprivate enum Action {
    /// Do nothing.
    case none

    /// Receive the request part in the interceptor pipeline.
    case receiveRequestPartInInterceptors(GRPCServerRequestPart<_BaseCallHandler.RequestPayload>)

    /// Receive the request part in the observer.
    case receiveRequestPartInObserver(GRPCServerRequestPart<_BaseCallHandler.RequestPayload>)

    /// Receive an error in the observer.
    case receiveLibraryErrorInObserver(Error)

    /// Send a response part to the interceptor pipeline.
    case sendResponsePartToInterceptors(
      GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
      EventLoopPromise<Void>?
    )

    /// Write the response part to the `Channel`.
    case writeResponsePartToChannel(
      ChannelHandlerContext,
      GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
      promise: EventLoopPromise<Void>?
    )

    /// Complete the promise with the result.
    case completePromise(EventLoopPromise<Void>?, Result<Void, Error>)

    /// Perform multiple actions.
    indirect case multiple([Action])
  }
}

extension _BaseCallHandler.State {
  /// The handler was added to the `ChannelPipeline`: this is the only way to move from the `.idle`
  /// state. We only expect this to be called once.
  internal mutating func handlerAdded(context: ChannelHandlerContext) -> Action {
    switch self {
    case .idle:
      // This is the only way we can become active.
      self = .active(.init(context: context))
      return .none

    case .active:
      preconditionFailure("Invalid state: already active")

    case .closed:
      return .none
    }
  }

  /// Received an error from the `Channel`.
  internal mutating func errorCaught(_ error: Error) -> Action {
    switch self {
    case .active:
      return .receiveLibraryErrorInObserver(error)

    case .idle, .closed:
      return .none
    }
  }

  /// Receive a request part from the `Channel`. If we're active we just forward these through the
  /// pipeline. We validate at the other end.
  internal mutating func channelRead(
    _ requestPart: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>
  ) -> Action {
    switch self {
    case .idle:
      preconditionFailure("Invalid state: the handler isn't in the pipeline yet")

    case var .active(state):
      // Avoid CoW-ing state.
      self = .idle

      let filter: StreamState.Filter
      let part: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>

      switch requestPart {
      case let .metadata(headers):
        filter = state.channelStreamState.receiveHeaders()
        part = .metadata(headers)
      case let .message(message):
        filter = state.channelStreamState.receiveMessage()
        part = .message(message)
      case .end:
        filter = state.channelStreamState.receiveEnd()
        part = .end
      }

      // Restore state.
      self = .active(state)

      switch filter {
      case .allow:
        return .receiveRequestPartInInterceptors(part)
      case .drop:
        return .none
      }

    case .closed:
      return .none
    }
  }

  /// Send a response part from the observer to the interceptors.
  internal mutating func sendResponsePartFromObserver(
    _ part: GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
    promise: EventLoopPromise<Void>?
  ) -> Action {
    switch self {
    case .idle:
      preconditionFailure("Invalid state: the handler isn't in the pipeline yet")

    case var .active(state):
      // Avoid CoW-ing 'state'.
      self = .idle

      let filter: StreamState.Filter

      switch part {
      case .metadata:
        filter = state.observerStreamState.sendHeaders()
      case .message:
        filter = state.observerStreamState.sendMessage()
      case .end:
        filter = state.observerStreamState.sendEnd()
      }

      // Restore the state.
      self = .active(state)

      switch filter {
      case .allow:
        return .sendResponsePartToInterceptors(part, promise)
      case .drop:
        return .completePromise(promise, .failure(GRPCError.AlreadyComplete()))
      }

    case .closed:
      return .completePromise(promise, .failure(GRPCError.AlreadyComplete()))
    }
  }

  /// Send a response part from the interceptors to the `Channel`.
  internal mutating func sendResponsePartFromInterceptors(
    _ part: GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
    promise: EventLoopPromise<Void>?
  ) -> Action {
    switch self {
    case .idle:
      preconditionFailure("Invalid state: can't send response on idle call")

    _ part: GRPCServerResponsePart<ResponsePayload>,
    promise: EventLoopPromise<Void>?
  ) {
    let forward: Bool

    switch part {
    case .metadata:
      forward = self.state.sendResponsePartFromInterceptors(.metadata)
    case .message:
      forward = self.state.sendResponsePartFromInterceptors(.message)
    case .end:
      forward = self.state.sendResponsePartFromInterceptors(.end)
    }

    if forward, let context = self.context {
      self.writeResponsePartToChannel(context: context, part: part, promise: promise)
    } else {
      promise?.fail(GRPCError.AlreadyComplete())
    }
  }
}

// MARK: - State

private enum State {
  /// Idle. We're waiting to be added to a pipeline.
  case idle

  /// We're in a pipeline and receiving from the client.
  case active(ActiveState)

  /// We're done. This state is terminal, all actions are ignored.
  case closed
}

private enum RPCStreamPart {
  case metadata
  case message
  case end
}

extension State {
  /// The state of the request and response streams.
  ///
  /// We track the stream state twice: between the 'Channel' and interceptor pipeline, and between
