     _ part: GRPCServerResponsePart<ResponsePayload>,
     promise: EventLoopPromise<Void>?
   ) {
-    self.act(on: self.state.sendResponsePartFromInterceptors(part, promise: promise))
+    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
 
-extension _BaseCallHandler {
-  fileprivate 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 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
 }
 
-extension _BaseCallHandler.State {
+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


    switch part {
    case let .metadata(headers):
      self.act(on: self.state.channelRead(.metadata(headers)))
    case let .message(buffer):
      do {
        let request = try self.requestDeserializer.deserialize(byteBuffer: buffer)
        self.act(on: self.state.channelRead(.message(request)))
      } catch {
        self.errorCaught(context: context, error: error)
      }
    case .end:
      self.act(on: self.state.channelRead(.end))
    }
    // We're the last handler. We don't have anything to forward.
  }


  }

  public func errorCaught(context: ChannelHandlerContext, error: Error) {
    if self.state.errorCaught() {
      self.observeLibraryError(error)
    }
  }

  public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
