diff --git a/packages/grpc-js/src/call-interface.ts b/packages/grpc-js/src/call-interface.ts
index 15035aeac..c0c63b957 100644
--- a/packages/grpc-js/src/call-interface.ts
+++ b/packages/grpc-js/src/call-interface.ts
@@ -37,7 +37,7 @@ export interface StatusObject {
 }
 
 export type PartialStatusObject = Pick<StatusObject, 'code' | 'details'> & {
-  metadata: Metadata | null;
+  metadata?: Metadata | null | undefined;
 };
 
 export const enum WriteFlags {
diff --git a/packages/grpc-js/src/index.ts b/packages/grpc-js/src/index.ts
index 50671b01c..b37f61103 100644
--- a/packages/grpc-js/src/index.ts
+++ b/packages/grpc-js/src/index.ts
@@ -49,6 +49,7 @@ import {
 import { Metadata, MetadataOptions, MetadataValue } from './metadata';
 import {
   Server,
+  ServerOptions,
   UntypedHandleCall,
   UntypedServiceImplementation,
 } from './server';
@@ -226,7 +227,7 @@ export const setLogVerbosity = (verbosity: LogVerbosity): void => {
   logging.setLoggerVerbosity(verbosity);
 };
 
-export { Server };
+export { Server, ServerOptions };
 export { ServerCredentials };
 export { KeyCertPair };
 
@@ -264,6 +265,18 @@ export { addAdminServicesToServer } from './admin';
 
 export { ServiceConfig, LoadBalancingConfig, MethodConfig, RetryPolicy } from './service-config';
 
+export {
+  ServerListener,
+  FullServerListener,
+  ServerListenerBuilder,
+  Responder,
+  FullResponder,
+  ResponderBuilder,
+  ServerInterceptingCallInterface,
+  ServerInterceptingCall,
+  ServerInterceptor
+} from './server-interceptors';
+
 import * as experimental from './experimental';
 export { experimental };
 
diff --git a/packages/grpc-js/src/server-call.ts b/packages/grpc-js/src/server-call.ts
index 107c2e3ef..8fb1de2aa 100644
--- a/packages/grpc-js/src/server-call.ts
+++ b/packages/grpc-js/src/server-call.ts
@@ -16,66 +16,17 @@
  */
 
 import { EventEmitter } from 'events';
-import * as http2 from 'http2';
 import { Duplex, Readable, Writable } from 'stream';
-import * as zlib from 'zlib';
-import { promisify } from 'util';
 
 import {
   Status,
-  DEFAULT_MAX_SEND_MESSAGE_LENGTH,
-  DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
-  LogVerbosity,
 } from './constants';
 import { Deserialize, Serialize } from './make-client';
 import { Metadata } from './metadata';
-import { StreamDecoder } from './stream-decoder';
 import { ObjectReadable, ObjectWritable } from './object-stream';
-import { ChannelOptions } from './channel-options';
-import * as logging from './logging';
 import { StatusObject, PartialStatusObject } from './call-interface';
 import { Deadline } from './deadline';
-import { getErrorCode, getErrorMessage } from './error';
-
-const TRACER_NAME = 'server_call';
-const unzip = promisify(zlib.unzip);
-const inflate = promisify(zlib.inflate);
-
-function trace(text: string): void {
-  logging.trace(LogVerbosity.DEBUG, TRACER_NAME, text);
-}
-
-interface DeadlineUnitIndexSignature {
-  [name: string]: number;
-}
-
-const GRPC_ACCEPT_ENCODING_HEADER = 'grpc-accept-encoding';
-const GRPC_ENCODING_HEADER = 'grpc-encoding';
-const GRPC_MESSAGE_HEADER = 'grpc-message';
-const GRPC_STATUS_HEADER = 'grpc-status';
-const GRPC_TIMEOUT_HEADER = 'grpc-timeout';
-const DEADLINE_REGEX = /(\d{1,8})\s*([HMSmun])/;
-const deadlineUnitsToMs: DeadlineUnitIndexSignature = {
-  H: 3600000,
-  M: 60000,
-  S: 1000,
-  m: 1,
-  u: 0.001,
-  n: 0.000001,
-};
-const defaultCompressionHeaders = {
-  // TODO(cjihrig): Remove these encoding headers from the default response
-  // once compression is integrated.
-  [GRPC_ACCEPT_ENCODING_HEADER]: 'identity,deflate,gzip',
-  [GRPC_ENCODING_HEADER]: 'identity',
-};
-const defaultResponseHeaders = {
-  [http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
-  [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
-};
-const defaultResponseOptions = {
-  waitForTrailers: true,
-} as http2.ServerStreamResponseOptions;
+import { ServerInterceptingCallInterface } from './server-interceptors';
 
 export type ServerStatusResponse = Partial<StatusObject>;
 
@@ -105,6 +56,38 @@ export type ServerDuplexStream<RequestType, ResponseType> = ServerSurfaceCall &
   ObjectReadable<RequestType> &
   ObjectWritable<ResponseType> & { end: (metadata?: Metadata) => void };
 
+export function serverErrorToStatus(error: ServerErrorResponse | ServerStatusResponse, extraTrailers?: Metadata | undefined): PartialStatusObject {
+  const status: PartialStatusObject = {
+    code: Status.UNKNOWN,
+    details: 'message' in error ? error.message : 'Unknown Error',
+    metadata:
+      'metadata' in error && error.metadata !== undefined
+        ? error.metadata
+        : null,
+  };
+
+  if (
+    'code' in error &&
+    typeof error.code === 'number' &&
+    Number.isInteger(error.code)
+  ) {
+    status.code = error.code;
+
+    if ('details' in error && typeof error.details === 'string') {
+      status.details = error.details!;
+    }
+  }
+
+  if (extraTrailers) {
+    if (status.metadata) {
+      status.metadata.merge(extraTrailers);
+    } else {
+      status.metadata = extraTrailers;
+    }
+  }
+  return status;
+}
+
 export class ServerUnaryCallImpl<RequestType, ResponseType>
   extends EventEmitter
   implements ServerUnaryCall<RequestType, ResponseType>
@@ -112,13 +95,13 @@ export class ServerUnaryCallImpl<RequestType, ResponseType>
   cancelled: boolean;
 
   constructor(
-    private call: Http2ServerCallStream<RequestType, ResponseType>,
+    private path: string,
+    private call: ServerInterceptingCallInterface,
     public metadata: Metadata,
     public request: RequestType
   ) {
     super();
     this.cancelled = false;
-    this.call.setupSurfaceCall(this);
   }
 
   getPeer(): string {
@@ -134,7 +117,7 @@ export class ServerUnaryCallImpl<RequestType, ResponseType>
   }
 
   getPath(): string {
-    return this.call.getPath();
+    return this.path;
   }
 }
 
@@ -145,23 +128,16 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
   cancelled: boolean;
 
   constructor(
-    private call: Http2ServerCallStream<RequestType, ResponseType>,
-    public metadata: Metadata,
-    public deserialize: Deserialize<RequestType>,
-    encoding: string
+    private path: string,
+    private call: ServerInterceptingCallInterface,
+    public metadata: Metadata
   ) {
     super({ objectMode: true });
     this.cancelled = false;
-    this.call.setupSurfaceCall(this);
-    this.call.setupReadable(this, encoding);
   }
 
   _read(size: number) {
-    if (!this.call.consumeUnpushedMessages(this)) {
-      return;
-    }
-
-    this.call.resume();
+    this.call.startRead();
   }
 
   getPeer(): string {
@@ -177,7 +153,7 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
   }
 
   getPath(): string {
-    return this.call.getPath();
+    return this.path;
   }
 }
 
@@ -187,20 +163,23 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
 {
   cancelled: boolean;
   private trailingMetadata: Metadata;
+  private pendingStatus: PartialStatusObject = {
+    code: Status.OK,
+    details: 'OK'
+  };
 
   constructor(
-    private call: Http2ServerCallStream<RequestType, ResponseType>,
+    private path: string,
+    private call: ServerInterceptingCallInterface,
     public metadata: Metadata,
-    public serialize: Serialize<ResponseType>,
     public request: RequestType
   ) {
     super({ objectMode: true });
     this.cancelled = false;
     this.trailingMetadata = new Metadata();
-    this.call.setupSurfaceCall(this);
 
     this.on('error', err => {
-      this.call.sendError(err);
+      this.pendingStatus = serverErrorToStatus(err);
       this.end();
     });
   }
@@ -218,7 +197,7 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
   }
 
   getPath(): string {
-    return this.call.getPath();
+    return this.path;
   }
 
   _write(
@@ -227,28 +206,13 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     callback: (...args: any[]) => void
   ) {
-    try {
-      const response = this.call.serializeMessage(chunk);
-
-      if (!this.call.write(response)) {
-        this.call.once('drain', callback);
-        return;
-      }
-    } catch (err) {
-      this.emit('error', {
-        details: getErrorMessage(err),
-        code: Status.INTERNAL,
-      });
-    }
-
-    callback();
+    this.call.sendMessage(chunk, callback);
   }
 
   _final(callback: Function): void {
     this.call.sendStatus({
-      code: Status.OK,
-      details: 'OK',
-      metadata: this.trailingMetadata,
+      ...this.pendingStatus,
+      metadata: this.pendingStatus.metadata ?? this.trailingMetadata,
     });
     callback(null);
   }
@@ -268,27 +232,23 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
   implements ServerDuplexStream<RequestType, ResponseType>
 {
   cancelled: boolean;
-  /* This field appears to be unsued, but it is actually used in _final, which is assiged from
-   * ServerWritableStreamImpl.prototype._final below. */
-  // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-  // @ts-ignore noUnusedLocals
   private trailingMetadata: Metadata;
+  private pendingStatus: PartialStatusObject = {
+    code: Status.OK,
+    details: 'OK'
+  };
 
   constructor(
-    private call: Http2ServerCallStream<RequestType, ResponseType>,
-    public metadata: Metadata,
-    public serialize: Serialize<ResponseType>,
-    public deserialize: Deserialize<RequestType>,
-    encoding: string
+    private path: string,
+    private call: ServerInterceptingCallInterface,
+    public metadata: Metadata
   ) {
     super({ objectMode: true });
     this.cancelled = false;
     this.trailingMetadata = new Metadata();
-    this.call.setupSurfaceCall(this);
-    this.call.setupReadable(this, encoding);
 
     this.on('error', err => {
-      this.call.sendError(err);
+      this.pendingStatus = serverErrorToStatus(err);
       this.end();
     });
   }
@@ -306,7 +266,28 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
   }
 
   getPath(): string {
-    return this.call.getPath();
+    return this.path;
+  }
+
+  _read(size: number) {
+    this.call.startRead();
+  }
+
+  _write(
+    chunk: ResponseType,
+    encoding: string,
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    callback: (...args: any[]) => void
+  ) {
+    this.call.sendMessage(chunk, callback);
+  }
+
+  _final(callback: Function): void {
+    this.call.sendStatus({
+      ...this.pendingStatus,
+      metadata: this.pendingStatus.metadata ?? this.trailingMetadata,
+    });
+    callback(null);
   }
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -319,13 +300,6 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
   }
 }
 
-ServerDuplexStreamImpl.prototype._read =
-  ServerReadableStreamImpl.prototype._read;
-ServerDuplexStreamImpl.prototype._write =
-  ServerWritableStreamImpl.prototype._write;
-ServerDuplexStreamImpl.prototype._final =
-  ServerWritableStreamImpl.prototype._final;
-
 // Unary response callback signature.
 export type sendUnaryData<ResponseType> = (
   error: ServerErrorResponse | ServerStatusResponse | null,
@@ -401,597 +375,3 @@ export type Handler<RequestType, ResponseType> =
   | BidiStreamingHandler<RequestType, ResponseType>;
 
 export type HandlerType = 'bidi' | 'clientStream' | 'serverStream' | 'unary';
-
-// Internal class that wraps the HTTP2 request.
-export class Http2ServerCallStream<
-  RequestType,
-  ResponseType
-> extends EventEmitter {
-  cancelled = false;
-  deadlineTimer: NodeJS.Timeout | null = null;
-  private statusSent = false;
-  private deadline: Deadline = Infinity;
-  private wantTrailers = false;
-  private metadataSent = false;
-  private canPush = false;
-  private isPushPending = false;
-  private bufferedMessages: Array<Buffer | null> = [];
-  private messagesToPush: Array<RequestType | null> = [];
-  private maxSendMessageSize: number = DEFAULT_MAX_SEND_MESSAGE_LENGTH;
-  private maxReceiveMessageSize: number = DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH;
-
-  constructor(
-    private stream: http2.ServerHttp2Stream,
-    private handler: Handler<RequestType, ResponseType>,
-    options: ChannelOptions
-  ) {
-    super();
-
-    this.stream.once('error', (err: ServerErrorResponse) => {
-      /* We need an error handler to avoid uncaught error event exceptions, but
-       * there is nothing we can reasonably do here. Any error event should
-       * have a corresponding close event, which handles emitting the cancelled
-       * event. And the stream is now in a bad state, so we can't reasonably
-       * expect to be able to send an error over it. */
-    });
-
-    this.stream.once('close', () => {
-      trace(
-        'Request to method ' +
-          this.handler?.path +
-          ' stream closed with rstCode ' +
-          this.stream.rstCode
-      );
-
-      if (!this.statusSent) {
-        this.cancelled = true;
-        this.emit('cancelled', 'cancelled');
-        this.emit('streamEnd', false);
-        this.sendStatus({
-          code: Status.CANCELLED,
-          details: 'Cancelled by client',
-          metadata: null,
-        });
-        if (this.deadlineTimer) clearTimeout(this.deadlineTimer);
-      }
-    });
-
-    this.stream.on('drain', () => {
-      this.emit('drain');
-    });
-
-    if ('grpc.max_send_message_length' in options) {
-      this.maxSendMessageSize = options['grpc.max_send_message_length']!;
-    }
-    if ('grpc.max_receive_message_length' in options) {
-      this.maxReceiveMessageSize = options['grpc.max_receive_message_length']!;
-    }
-  }
-
-  private checkCancelled(): boolean {
-    /* In some cases the stream can become destroyed before the close event
-     * fires. That creates a race condition that this check works around */
-    if (this.stream.destroyed || this.stream.closed) {
-      this.cancelled = true;
-    }
-    return this.cancelled;
-  }
-
-  private getDecompressedMessage(
-    message: Buffer,
-    encoding: string
-  ): Buffer | Promise<Buffer> {
-    if (encoding === 'deflate') {
-      return inflate(message.subarray(5));
-    } else if (encoding === 'gzip') {
-      return unzip(message.subarray(5));
-    } else if (encoding === 'identity') {
-      return message.subarray(5);
-    }
-
-    return Promise.reject({
-      code: Status.UNIMPLEMENTED,
-      details: `Received message compressed with unsupported encoding "${encoding}"`,
-    });
-  }
-
-  sendMetadata(customMetadata?: Metadata) {
-    if (this.checkCancelled()) {
-      return;
-    }
-
-    if (this.metadataSent) {
-      return;
-    }
-
-    this.metadataSent = true;
-    const custom = customMetadata ? customMetadata.toHttp2Headers() : null;
-    // TODO(cjihrig): Include compression headers.
-    const headers = {
-      ...defaultResponseHeaders,
-      ...defaultCompressionHeaders,
-      ...custom,
-    };
-    this.stream.respond(headers, defaultResponseOptions);
-  }
-
-  receiveMetadata(headers: http2.IncomingHttpHeaders) {
-    const metadata = Metadata.fromHttp2Headers(headers);
-
-    if (logging.isTracerEnabled(TRACER_NAME)) {
-      trace(
-        'Request to ' +
-          this.handler.path +
-          ' received headers ' +
-          JSON.stringify(metadata.toJSON())
-      );
-    }
-
-    // TODO(cjihrig): Receive compression metadata.
-
-    const timeoutHeader = metadata.get(GRPC_TIMEOUT_HEADER);
-
-    if (timeoutHeader.length > 0) {
-      const match = timeoutHeader[0].toString().match(DEADLINE_REGEX);
-
-      if (match === null) {
-        const err = new Error('Invalid deadline') as ServerErrorResponse;
-        err.code = Status.OUT_OF_RANGE;
-        this.sendError(err);
-        return metadata;
-      }
-
-      const timeout = (+match[1] * deadlineUnitsToMs[match[2]]) | 0;
-
-      const now = new Date();
-      this.deadline = now.setMilliseconds(now.getMilliseconds() + timeout);
-      this.deadlineTimer = setTimeout(handleExpiredDeadline, timeout, this);
-      metadata.remove(GRPC_TIMEOUT_HEADER);
-    }
-
-    // Remove several headers that should not be propagated to the application
-    metadata.remove(http2.constants.HTTP2_HEADER_ACCEPT_ENCODING);
-    metadata.remove(http2.constants.HTTP2_HEADER_TE);
-    metadata.remove(http2.constants.HTTP2_HEADER_CONTENT_TYPE);
-    metadata.remove('grpc-accept-encoding');
-
-    return metadata;
-  }
-
-  receiveUnaryMessage(encoding: string): Promise<RequestType> {
-    return new Promise((resolve, reject) => {
-      const { stream } = this;
-
-      let receivedLength = 0;
-
-      // eslint-disable-next-line @typescript-eslint/no-this-alias
-      const call = this;
-      const body: Buffer[] = [];
-      const limit = this.maxReceiveMessageSize;
-
-      this.stream.on('data', onData);
-      this.stream.on('end', onEnd);
-      this.stream.on('error', onEnd);
-
-      function onData(chunk: Buffer) {
-        receivedLength += chunk.byteLength;
-
-        if (limit !== -1 && receivedLength > limit) {
-          stream.removeListener('data', onData);
-          stream.removeListener('end', onEnd);
-          stream.removeListener('error', onEnd);
-
-          reject({
-            code: Status.RESOURCE_EXHAUSTED,
-            details: `Received message larger than max (${receivedLength} vs. ${limit})`,
-          });
-          return;
-        }
-
-        body.push(chunk);
-      }
-
-      function onEnd(err?: Error) {
-        stream.removeListener('data', onData);
-        stream.removeListener('end', onEnd);
-        stream.removeListener('error', onEnd);
-
-        if (err !== undefined) {
-          reject({ code: Status.INTERNAL, details: err.message });
-          return;
-        }
-
-        if (receivedLength === 0) {
-          reject({
-            code: Status.INTERNAL,
-            details: 'received empty unary message',
-          });
-          return;
-        }
-
-        call.emit('receiveMessage');
-
-        const requestBytes = Buffer.concat(body, receivedLength);
-        const compressed = requestBytes.readUInt8(0) === 1;
-        const compressedMessageEncoding = compressed ? encoding : 'identity';
-        const decompressedMessage = call.getDecompressedMessage(
-          requestBytes,
-          compressedMessageEncoding
-        );
-
-        if (Buffer.isBuffer(decompressedMessage)) {
-          resolve(
-            call.deserializeMessageWithInternalError(decompressedMessage)
-          );
-          return;
-        }
-
-        decompressedMessage.then(
-          decompressed =>
-            resolve(call.deserializeMessageWithInternalError(decompressed)),
-          (err: any) =>
-            reject(
-              err.code
-                ? err
-                : {
-                    code: Status.INTERNAL,
-                    details: `Received "grpc-encoding" header "${encoding}" but ${encoding} decompression failed`,
-                  }
-            )
-        );
-      }
-    });
-  }
-
-  private async deserializeMessageWithInternalError(buffer: Buffer) {
-    try {
-      return this.deserializeMessage(buffer);
-    } catch (err) {
-      throw {
-        details: getErrorMessage(err),
-        code: Status.INTERNAL,
-      };
-    }
-  }
-
-  serializeMessage(value: ResponseType) {
-    const messageBuffer = this.handler.serialize(value);
-
-    // TODO(cjihrig): Call compression aware serializeMessage().
-    const byteLength = messageBuffer.byteLength;
-    const output = Buffer.allocUnsafe(byteLength + 5);
-    output.writeUInt8(0, 0);
-    output.writeUInt32BE(byteLength, 1);
-    messageBuffer.copy(output, 5);
-    return output;
-  }
-
-  deserializeMessage(bytes: Buffer) {
-    return this.handler.deserialize(bytes);
-  }
-
-  async sendUnaryMessage(
-    err: ServerErrorResponse | ServerStatusResponse | null,
-    value?: ResponseType | null,
-    metadata?: Metadata | null,
-    flags?: number
-  ) {
-    if (this.checkCancelled()) {
-      return;
-    }
-
-    if (metadata === undefined) {
-      metadata = null;
-    }
-
-    if (err) {
-      if (!Object.prototype.hasOwnProperty.call(err, 'metadata') && metadata) {
-        err.metadata = metadata;
-      }
-      this.sendError(err);
-      return;
-    }
-
-    try {
-      const response = this.serializeMessage(value!);
-
-      this.write(response);
-      this.sendStatus({ code: Status.OK, details: 'OK', metadata });
-    } catch (err) {
-      this.sendError({
-        details: getErrorMessage(err),
-        code: Status.INTERNAL,
-      });
-    }
-  }
-
-  sendStatus(statusObj: PartialStatusObject) {
-    this.emit('callEnd', statusObj.code);
-    this.emit('streamEnd', statusObj.code === Status.OK);
-    if (this.checkCancelled()) {
-      return;
-    }
-
-    trace(
-      'Request to method ' +
-        this.handler?.path +
-        ' ended with status code: ' +
-        Status[statusObj.code] +
-        ' details: ' +
-        statusObj.details
-    );
-
-    if (this.deadlineTimer) clearTimeout(this.deadlineTimer);
-
-    if (this.stream.headersSent) {
-      if (!this.wantTrailers) {
-        this.wantTrailers = true;
-        this.stream.once('wantTrailers', () => {
-          const trailersToSend = {
-            [GRPC_STATUS_HEADER]: statusObj.code,
-            [GRPC_MESSAGE_HEADER]: encodeURI(statusObj.details),
-            ...statusObj.metadata?.toHttp2Headers(),
-          };
-
-          this.stream.sendTrailers(trailersToSend);
-          this.statusSent = true;
-        });
-        this.stream.end();
-      }
-    } else {
-      // Trailers-only response
-      const trailersToSend = {
-        [GRPC_STATUS_HEADER]: statusObj.code,
-        [GRPC_MESSAGE_HEADER]: encodeURI(statusObj.details),
-        ...defaultResponseHeaders,
-        ...statusObj.metadata?.toHttp2Headers(),
-      };
-      this.stream.respond(trailersToSend, { endStream: true });
-      this.statusSent = true;
-    }
-  }
-
-  sendError(error: ServerErrorResponse | ServerStatusResponse) {
-    const status: PartialStatusObject = {
-      code: Status.UNKNOWN,
-      details: 'message' in error ? error.message : 'Unknown Error',
-      metadata:
-        'metadata' in error && error.metadata !== undefined
-          ? error.metadata
-          : null,
-    };
-
-    if (
-      'code' in error &&
-      typeof error.code === 'number' &&
-      Number.isInteger(error.code)
-    ) {
-      status.code = error.code;
-
-      if ('details' in error && typeof error.details === 'string') {
-        status.details = error.details!;
-      }
-    }
-
-    this.sendStatus(status);
-  }
-
-  write(chunk: Buffer) {
-    if (this.checkCancelled()) {
-      return;
-    }
-
-    if (
-      this.maxSendMessageSize !== -1 &&
-      chunk.length > this.maxSendMessageSize
-    ) {
-      this.sendError({
-        code: Status.RESOURCE_EXHAUSTED,
-        details: `Sent message larger than max (${chunk.length} vs. ${this.maxSendMessageSize})`,
-      });
-      return;
-    }
-
-    this.sendMetadata();
-    this.emit('sendMessage');
-    return this.stream.write(chunk);
-  }
-
-  resume() {
-    this.stream.resume();
-  }
-
-  setupSurfaceCall(call: ServerSurfaceCall) {
-    this.once('cancelled', reason => {
-      call.cancelled = true;
-      call.emit('cancelled', reason);
-    });
-
-    this.once('callEnd', status => call.emit('callEnd', status));
-  }
-
-  setupReadable(
-    readable:
-      | ServerReadableStream<RequestType, ResponseType>
-      | ServerDuplexStream<RequestType, ResponseType>,
-    encoding: string
-  ) {
-    const decoder = new StreamDecoder();
-
-    let readsDone = false;
-
-    let pendingMessageProcessing = false;
-
-    let pushedEnd = false;
-
-    const maybePushEnd = async () => {
-      if (!pushedEnd && readsDone && !pendingMessageProcessing) {
-        pushedEnd = true;
-        await this.pushOrBufferMessage(readable, null);
-      }
-    };
-
-    this.stream.on('data', async (data: Buffer) => {
-      const messages = decoder.write(data);
-
-      pendingMessageProcessing = true;
-      this.stream.pause();
-      for (const message of messages) {
-        if (
-          this.maxReceiveMessageSize !== -1 &&
-          message.length > this.maxReceiveMessageSize
-        ) {
-          this.sendError({
-            code: Status.RESOURCE_EXHAUSTED,
-            details: `Received message larger than max (${message.length} vs. ${this.maxReceiveMessageSize})`,
-          });
-          return;
-        }
-        this.emit('receiveMessage');
-
-        const compressed = message.readUInt8(0) === 1;
-        const compressedMessageEncoding = compressed ? encoding : 'identity';
-        const decompressedMessage = await this.getDecompressedMessage(
-          message,
-          compressedMessageEncoding
-        );
-
-        // Encountered an error with decompression; it'll already have been propogated back
-        // Just return early
-        if (!decompressedMessage) return;
-
-        await this.pushOrBufferMessage(readable, decompressedMessage);
-      }
-      pendingMessageProcessing = false;
-      this.stream.resume();
-      await maybePushEnd();
-    });
-
-    this.stream.once('end', async () => {
-      readsDone = true;
-      await maybePushEnd();
-    });
-  }
-
-  consumeUnpushedMessages(
-    readable:
-      | ServerReadableStream<RequestType, ResponseType>
-      | ServerDuplexStream<RequestType, ResponseType>
-  ): boolean {
-    this.canPush = true;
-
-    while (this.messagesToPush.length > 0) {
-      const nextMessage = this.messagesToPush.shift();
-      const canPush = readable.push(nextMessage);
-
-      if (nextMessage === null || canPush === false) {
-        this.canPush = false;
-        break;
-      }
-    }
-
-    return this.canPush;
-  }
-
-  private async pushOrBufferMessage(
-    readable:
-      | ServerReadableStream<RequestType, ResponseType>
-      | ServerDuplexStream<RequestType, ResponseType>,
-    messageBytes: Buffer | null
-  ): Promise<void> {
-    if (this.isPushPending) {
-      this.bufferedMessages.push(messageBytes);
-    } else {
-      await this.pushMessage(readable, messageBytes);
-    }
-  }
-
-  private async pushMessage(
-    readable:
-      | ServerReadableStream<RequestType, ResponseType>
-      | ServerDuplexStream<RequestType, ResponseType>,
-    messageBytes: Buffer | null
-  ) {
-    if (messageBytes === null) {
-      trace('Received end of stream');
-      if (this.canPush) {
-        readable.push(null);
-      } else {
-        this.messagesToPush.push(null);
-      }
-
-      return;
-    }
-
-    trace('Received message of length ' + messageBytes.length);
-
-    this.isPushPending = true;
-
-    try {
-      const deserialized = await this.deserializeMessage(messageBytes);
-
-      if (this.canPush) {
-        if (!readable.push(deserialized)) {
-          this.canPush = false;
-          this.stream.pause();
-        }
-      } else {
-        this.messagesToPush.push(deserialized);
-      }
-    } catch (error) {
-      // Ignore any remaining messages when errors occur.
-      this.bufferedMessages.length = 0;
-      let code = getErrorCode(error);
-      if (code === null || code < Status.OK || code > Status.UNAUTHENTICATED) {
-        code = Status.INTERNAL;
-      }
-
-      readable.emit('error', {
-        details: getErrorMessage(error),
-        code: code,
-      });
-    }
-
-    this.isPushPending = false;
-
-    if (this.bufferedMessages.length > 0) {
-      await this.pushMessage(
-        readable,
-        this.bufferedMessages.shift() as Buffer | null
-      );
-    }
-  }
-
-  getPeer(): string {
-    const socket = this.stream.session?.socket;
-    if (socket?.remoteAddress) {
-      if (socket.remotePort) {
-        return `${socket.remoteAddress}:${socket.remotePort}`;
-      } else {
-        return socket.remoteAddress;
-      }
-    } else {
-      return 'unknown';
-    }
-  }
-
-  getDeadline(): Deadline {
-    return this.deadline;
-  }
-
-  getPath(): string {
-    return this.handler.path;
-  }
-}
-
-/* eslint-disable @typescript-eslint/no-explicit-any */
-type UntypedServerCall = Http2ServerCallStream<any, any>;
-
-function handleExpiredDeadline(call: UntypedServerCall) {
-  const err = new Error('Deadline exceeded') as ServerErrorResponse;
-  err.code = Status.DEADLINE_EXCEEDED;
-
-  call.sendError(err);
-  call.cancelled = true;
-  call.emit('cancelled', 'deadline');
-}
diff --git a/packages/grpc-js/src/server-interceptors.ts b/packages/grpc-js/src/server-interceptors.ts
new file mode 100644
index 000000000..c03f3028c
--- /dev/null
+++ b/packages/grpc-js/src/server-interceptors.ts
@@ -0,0 +1,886 @@
+/*
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import { PartialStatusObject} from "./call-interface";
+import { ServerMethodDefinition } from "./make-client";
+import { Metadata } from "./metadata";
+import { ChannelOptions } from "./channel-options";
+import { Handler, ServerErrorResponse } from "./server-call";
+import { Deadline } from "./deadline";
+import { DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH, DEFAULT_MAX_SEND_MESSAGE_LENGTH, LogVerbosity, Status } from "./constants";
+import * as http2 from 'http2';
+import { getErrorMessage } from "./error";
+import * as zlib from 'zlib';
+import { promisify } from "util";
+import { StreamDecoder } from "./stream-decoder";
+import { CallEventTracker } from "./transport";
+import * as logging from './logging';
+
+const unzip = promisify(zlib.unzip);
+const inflate = promisify(zlib.inflate);
+
+const TRACER_NAME = 'server_call';
+
+function trace(text: string) {
+  logging.trace(LogVerbosity.DEBUG, TRACER_NAME, text);
+}
+
+export interface ServerMetadataListener {
+  (metadata: Metadata, next: (metadata: Metadata) => void): void;
+}
+
+export interface ServerMessageListener {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  (message: any, next: (message: any) => void): void;
+}
+
+export interface ServerHalfCloseListener {
+  (next: () => void): void;
+}
+
+export interface ServerCancelListener {
+  (): void;
+}
+
+export interface FullServerListener {
+  onReceiveMetadata: ServerMetadataListener;
+  onReceiveMessage: ServerMessageListener;
+  onReceiveHalfClose: ServerHalfCloseListener;
+  onCancel: ServerCancelListener;
+}
+
+export type ServerListener = Partial<FullServerListener>;
+
+export class ServerListenerBuilder {
+  private metadata: ServerMetadataListener | undefined = undefined;
+  private message: ServerMessageListener | undefined = undefined;
+  private halfClose: ServerHalfCloseListener | undefined = undefined;
+  private cancel: ServerCancelListener | undefined = undefined;
+
+  withOnReceiveMetadata(onReceiveMetadata: ServerMetadataListener): this {
+    this.metadata = onReceiveMetadata;
+    return this;
+  }
+
+  withOnReceiveMessage(onReceiveMessage: ServerMessageListener): this {
+    this.message = onReceiveMessage;
+    return this;
+  }
+
+  withOnReceiveHalfClose(onReceiveHalfClose: ServerHalfCloseListener): this {
+    this.halfClose = onReceiveHalfClose;
+    return this;
+  }
+
+  withOnCancel(onCancel: ServerCancelListener): this {
+    this.cancel = onCancel;
+    return this;
+  }
+
+  build(): ServerListener {
+    return {
+      onReceiveMetadata: this.metadata,
+      onReceiveMessage: this.message,
+      onReceiveHalfClose: this.halfClose,
+      onCancel: this.cancel
+    };
+  }
+}
+
+export interface InterceptingServerListener {
+  onReceiveMetadata(metadata: Metadata): void;
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  onReceiveMessage(message: any): void;
+  onReceiveHalfClose(): void;
+  onCancel(): void;
+}
+
+export function isInterceptingServerListener(listener: ServerListener | InterceptingServerListener): listener is InterceptingServerListener {
+  return listener.onReceiveMetadata !== undefined && listener.onReceiveMetadata.length === 1;
+}
+
+class InterceptingServerListenerImpl implements InterceptingServerListener {
+  /**
+   * Once the call is cancelled, ignore all other events.
+   */
+  private cancelled: boolean = false;
+  private processingMetadata: boolean = false;
+  private hasPendingMessage: boolean = false;
+  private pendingMessage: any = null;
+  private processingMessage: boolean = false;
+  private hasPendingHalfClose: boolean = false;
+
+  constructor(private listener: FullServerListener, private nextListener: InterceptingServerListener) {}
+
+  private processPendingMessage() {
+    if (this.hasPendingMessage) {
+      this.nextListener.onReceiveMessage(this.pendingMessage);
+      this.pendingMessage = null;
+      this.hasPendingMessage = false;
+    }
+  }
+
+  private processPendingHalfClose() {
+    if (this.hasPendingHalfClose) {
+      this.nextListener.onReceiveHalfClose();
+      this.hasPendingHalfClose = false;
+    }
+  }
+
+  onReceiveMetadata(metadata: Metadata): void {
+    if (this.cancelled) {
+      return;
+    }
+    this.processingMetadata = true;
+    this.listener.onReceiveMetadata(metadata, interceptedMetadata => {
+      this.processingMetadata = false;
+      if (this.cancelled) {
+        return;
+      }
+      this.nextListener.onReceiveMetadata(interceptedMetadata);
+      this.processPendingMessage();
+      this.processPendingHalfClose();
+    });
+  }
+  onReceiveMessage(message: any): void {
+    if (this.cancelled) {
+      return;
+    }
+    this.processingMessage = true;
+    this.listener.onReceiveMessage(message, msg => {
+      this.processingMessage = false;
+      if (this.cancelled) {
+        return;
+      }
+      if (this.processingMetadata) {
+        this.pendingMessage = msg;
+        this.hasPendingMessage = true;
+      } else {
+        this.nextListener.onReceiveMessage(msg);
+        this.processPendingHalfClose();
+      }
+    });
+  }
+  onReceiveHalfClose(): void {
+    if (this.cancelled) {
+      return;
+    }
+    this.listener.onReceiveHalfClose(() => {
+      if (this.cancelled) {
+        return;
+      }
+      if (this.processingMetadata || this.processingMessage) {
+        this.hasPendingHalfClose = true;
+      } else {
+        this.nextListener.onReceiveHalfClose();
+      }
+    });
+  }
+  onCancel(): void {
+    this.cancelled = true;
+    this.listener.onCancel();
+    this.nextListener.onCancel();
+  }
+
+}
+
+export interface StartResponder {
+  (next: (listener?: ServerListener) => void): void;
+}
+
+export interface MetadataResponder {
+  (metadata: Metadata, next: (metadata: Metadata) => void): void;
+}
+
+export interface MessageResponder {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  (message: any, next: (message: any) => void): void;
+}
+
+export interface StatusResponder {
+  (status: PartialStatusObject, next: (status: PartialStatusObject) => void): void;
+}
+
+export interface FullResponder {
+  start: StartResponder;
+  sendMetadata: MetadataResponder;
+  sendMessage: MessageResponder;
+  sendStatus: StatusResponder;
+}
+
+export type Responder = Partial<FullResponder>;
+
+export class ResponderBuilder {
+  private start: StartResponder | undefined = undefined;
+  private metadata: MetadataResponder | undefined = undefined;
+  private message: MessageResponder | undefined = undefined;
+  private status: StatusResponder | undefined = undefined;
+
+  withStart(start: StartResponder): this {
+    this.start = start;
+    return this;
+  }
+
+  withSendMetadata(sendMetadata: MetadataResponder): this {
+    this.metadata = sendMetadata;
+    return this;
+  }
+
+  withSendMessage(sendMessage: MessageResponder): this {
+    this.message = sendMessage;
+    return this;
+  }
+
+  withSendStatus(sendStatus: StatusResponder): this {
+    this.status = sendStatus;
+    return this;
+  }
+
+  build(): Responder {
+    return {
+      start: this.start,
+      sendMetadata: this.metadata,
+      sendMessage: this.message,
+      sendStatus: this.status
+    };
+  }
+}
+
+const defaultServerListener: FullServerListener = {
+  onReceiveMetadata: (metadata, next) => {
+    next(metadata);
+  },
+  onReceiveMessage: (message, next) => {
+    next(message);
+  },
+  onReceiveHalfClose: next => {
+    next();
+  },
+  onCancel: () => {}
+};
+
+const defaultResponder: FullResponder = {
+  start: (next) => {
+    next();
+  },
+  sendMetadata: (metadata, next) => {
+    next(metadata);
+  },
+  sendMessage: (message, next) => {
+    next(message);
+  },
+  sendStatus: (status, next) => {
+    next(status);
+  }
+};
+
+export interface ServerInterceptingCallInterface {
+  /**
+   * Register the listener to handle inbound events.
+   */
+  start(listener: InterceptingServerListener): void;
+  /**
+   * Send response metadata.
+   */
+  sendMetadata(metadata: Metadata): void;
+  /**
+   * Send a response message.
+   */
+  sendMessage(message: any, callback: () => void): void;
+  /**
+   * End the call by sending this status.
+   */
+  sendStatus(status: PartialStatusObject): void;
+  /**
+   * Start a single read, eventually triggering either listener.onReceiveMessage or listener.onReceiveHalfClose.
+   */
+  startRead(): void;
+  /**
+   * Return the peer address of the client making the request, if known, or "unknown" otherwise
+   */
+  getPeer(): string;
+  /**
+   * Return the call deadline set by the client. The value is Infinity if there is no deadline.
+   */
+  getDeadline(): Deadline;
+}
+
+export class ServerInterceptingCall implements ServerInterceptingCallInterface {
+  private responder: FullResponder;
+  private processingMetadata: boolean = false;
+  private processingMessage: boolean = false;
+  private pendingMessage: any = null;
+  private pendingMessageCallback: (() => void) | null = null;
+  private pendingStatus: PartialStatusObject | null = null;
+  constructor(private nextCall: ServerInterceptingCallInterface, responder?: Responder) {
+    this.responder = {...defaultResponder, ...responder};
+  }
+
+  private processPendingMessage() {
+    if (this.pendingMessageCallback) {
+      this.nextCall.sendMessage(this.pendingMessage, this.pendingMessageCallback);
+      this.pendingMessage = null;
+      this.pendingMessageCallback = null;
+    }
+  }
+
+  private processPendingStatus() {
+    if (this.pendingStatus) {
+      this.nextCall.sendStatus(this.pendingStatus);
+      this.pendingStatus = null;
+    }
+  }
+
+  start(listener: InterceptingServerListener): void {
+    this.responder.start(interceptedListener => {
+      const fullInterceptedListener: FullServerListener = {...defaultServerListener, ...interceptedListener};
+      const finalInterceptingListener = new InterceptingServerListenerImpl(fullInterceptedListener, listener);
+      this.nextCall.start(finalInterceptingListener);
+    });
+  }
+  sendMetadata(metadata: Metadata): void {
+    this.processingMetadata = true;
+    this.responder.sendMetadata(metadata, interceptedMetadata => {
+      this.processingMetadata = false;
+      this.nextCall.sendMetadata(interceptedMetadata);
+      this.processPendingMessage();
+      this.processPendingStatus();
+    });
+  }
+  sendMessage(message: any, callback: () => void): void {
+    this.processingMessage = true;
+    this.responder.sendMessage(message, interceptedMessage => {
+      this.processingMessage = false;
+      if (this.processingMetadata) {
+        this.pendingMessage = interceptedMessage;
+        this.pendingMessageCallback = callback;
+      } else {
+        this.nextCall.sendMessage(interceptedMessage, callback);
+      }
+    });
+  }
+  sendStatus(status: PartialStatusObject): void {
+    this.responder.sendStatus(status, interceptedStatus => {
+      if (this.processingMetadata || this.processingMessage) {
+        this.pendingStatus = interceptedStatus;
+      } else {
+        this.nextCall.sendStatus(interceptedStatus);
+      }
+    });
+  }
+  startRead(): void {
+    this.nextCall.startRead();
+  }
+  getPeer(): string {
+    return this.nextCall.getPeer();
+  }
+  getDeadline(): Deadline {
+    return this.nextCall.getDeadline();
+  }
+}
+
+export interface ServerInterceptor {
+  (methodDescriptor: ServerMethodDefinition<any, any>, call: ServerInterceptingCallInterface): ServerInterceptingCall;
+}
+
+interface DeadlineUnitIndexSignature {
+  [name: string]: number;
+}
+
+const GRPC_ACCEPT_ENCODING_HEADER = 'grpc-accept-encoding';
+const GRPC_ENCODING_HEADER = 'grpc-encoding';
+const GRPC_MESSAGE_HEADER = 'grpc-message';
+const GRPC_STATUS_HEADER = 'grpc-status';
+const GRPC_TIMEOUT_HEADER = 'grpc-timeout';
+const DEADLINE_REGEX = /(\d{1,8})\s*([HMSmun])/;
+const deadlineUnitsToMs: DeadlineUnitIndexSignature = {
+  H: 3600000,
+  M: 60000,
+  S: 1000,
+  m: 1,
+  u: 0.001,
+  n: 0.000001,
+};
+
+const defaultCompressionHeaders = {
+  // TODO(cjihrig): Remove these encoding headers from the default response
+  // once compression is integrated.
+  [GRPC_ACCEPT_ENCODING_HEADER]: 'identity,deflate,gzip',
+  [GRPC_ENCODING_HEADER]: 'identity',
+};
+const defaultResponseHeaders = {
+  [http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
+  [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
+};
+const defaultResponseOptions = {
+  waitForTrailers: true,
+} as http2.ServerStreamResponseOptions;
+
+type ReadQueueEntryType = 'COMPRESSED' | 'READABLE' | 'HALF_CLOSE';
+
+interface ReadQueueEntry {
+  type: ReadQueueEntryType;
+  compressedMessage: Buffer | null;
+  parsedMessage: any;
+}
+
+export class BaseServerInterceptingCall implements ServerInterceptingCallInterface {
+  private listener: InterceptingServerListener | null = null;
+  private metadata: Metadata;
+  private deadlineTimer: NodeJS.Timeout | null = null;
+  private deadline: Deadline = Infinity;
+  private maxSendMessageSize: number = DEFAULT_MAX_SEND_MESSAGE_LENGTH;
+  private maxReceiveMessageSize: number = DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH;
+  private cancelled = false;
+  private metadataSent = false;
+  private wantTrailers = false;
+  private cancelNotified = false;
+  private incomingEncoding: string = 'identity';
+  private decoder = new StreamDecoder();
+  private readQueue: ReadQueueEntry[] = [];
+  private isReadPending = false;
+  private receivedHalfClose = false;
+  private streamEnded = false;
+
+  constructor(
+    private readonly stream: http2.ServerHttp2Stream,
+    headers: http2.IncomingHttpHeaders,
+    private readonly callEventTracker: CallEventTracker | null,
+    private readonly handler: Handler<any, any>,
+    options: ChannelOptions
+  ) {
+    this.stream.once('error', (err: ServerErrorResponse) => {
+      /* We need an error handler to avoid uncaught error event exceptions, but
+       * there is nothing we can reasonably do here. Any error event should
+       * have a corresponding close event, which handles emitting the cancelled
+       * event. And the stream is now in a bad state, so we can't reasonably
+       * expect to be able to send an error over it. */
+    });
+
+    this.stream.once('close', () => {
+      trace(
+        'Request to method ' +
+          this.handler?.path +
+          ' stream closed with rstCode ' +
+          this.stream.rstCode
+      );
+
+      if (this.callEventTracker && !this.streamEnded) {
+        this.streamEnded = true;
+        this.callEventTracker.onStreamEnd(false);
+        this.callEventTracker.onCallEnd({
+          code: Status.CANCELLED,
+          details: 'Stream closed before sending status',
+          metadata: null
+        });
+      }
+
+      this.notifyOnCancel();
+    });
+
+    this.stream.on('data', (data: Buffer) => {
+      this.handleDataFrame(data);
+    });
+    this.stream.pause();
+
+    this.stream.on('end', () => {
+      this.handleEndEvent();
+    });
+
+    if ('grpc.max_send_message_length' in options) {
+      this.maxSendMessageSize = options['grpc.max_send_message_length']!;
+    }
+    if ('grpc.max_receive_message_length' in options) {
+      this.maxReceiveMessageSize = options['grpc.max_receive_message_length']!;
+    }
+
+    const metadata = Metadata.fromHttp2Headers(headers);
+
+    if (logging.isTracerEnabled(TRACER_NAME)) {
+      trace(
+        'Request to ' +
+          this.handler.path +
+          ' received headers ' +
+          JSON.stringify(metadata.toJSON())
+      );
+    }
+
+    const timeoutHeader = metadata.get(GRPC_TIMEOUT_HEADER);
+
+    if (timeoutHeader.length > 0) {
+      this.handleTimeoutHeader(timeoutHeader[0] as string);
+    }
+
+    const encodingHeader = metadata.get(GRPC_ENCODING_HEADER);
+
+    if (encodingHeader.length > 0) {
+      this.incomingEncoding = encodingHeader[0] as string;
+    }
+
+    // Remove several headers that should not be propagated to the application
+    metadata.remove(GRPC_TIMEOUT_HEADER);
+    metadata.remove(GRPC_ENCODING_HEADER);
+    metadata.remove(GRPC_ACCEPT_ENCODING_HEADER);
+    metadata.remove(http2.constants.HTTP2_HEADER_ACCEPT_ENCODING);
+    metadata.remove(http2.constants.HTTP2_HEADER_TE);
+    metadata.remove(http2.constants.HTTP2_HEADER_CONTENT_TYPE);
+    this.metadata = metadata;
+  }
+
+  private handleTimeoutHeader(timeoutHeader: string) {
+    const match = timeoutHeader.toString().match(DEADLINE_REGEX);
+
+    if (match === null) {
+      const status: PartialStatusObject = {
+        code: Status.INTERNAL,
+        details: `Invalid ${GRPC_TIMEOUT_HEADER} value "${timeoutHeader}"`,
+        metadata: null
+      };
+      // Wait for the constructor to complete before sending the error.
+      process.nextTick(() => {
+        this.sendStatus(status);
+      });
+      return;
+    }
+
+    const timeout = (+match[1] * deadlineUnitsToMs[match[2]]) | 0;
+
+    const now = new Date();
+    this.deadline = now.setMilliseconds(now.getMilliseconds() + timeout);
+    this.deadlineTimer = setTimeout(() => {
+      const status: PartialStatusObject = {
+        code: Status.DEADLINE_EXCEEDED,
+        details: 'Deadline exceeded',
+        metadata: null
+      };
+      this.sendStatus(status);
+    }, timeout);
+
+  }
+
+  private checkCancelled(): boolean {
+    /* In some cases the stream can become destroyed before the close event
+     * fires. That creates a race condition that this check works around */
+    if (!this.cancelled && (this.stream.destroyed || this.stream.closed)) {
+      this.notifyOnCancel();
+      this.cancelled = true;
+    }
+    return this.cancelled;
+  }
+  private notifyOnCancel() {
+    if (this.cancelNotified) {
+      return;
+    }
+    this.cancelNotified = true;
+    this.cancelled = true;
+    process.nextTick(() => {
+      this.listener?.onCancel();
+    });
+    if (this.deadlineTimer) {
+      clearTimeout(this.deadlineTimer);
+    }
+    // Flush incoming data frames
+    this.stream.resume();
+  }
+
+  /**
+   * A server handler can start sending messages without explicitly sending
+   * metadata. In that case, we need to send headers before sending any
+   * messages. This function does that if necessary.
+   */
+  private maybeSendMetadata() {
+    if (!this.metadataSent) {
+      this.sendMetadata(new Metadata());
+    }
+  }
+
+  /**
+   * Serialize a message to a length-delimited byte string.
+   * @param value
+   * @returns
+   */
+  private serializeMessage(value: any) {
+    const messageBuffer = this.handler.serialize(value);
+    const byteLength = messageBuffer.byteLength;
+    const output = Buffer.allocUnsafe(byteLength + 5);
+    /* Note: response compression is currently not supported, so this
+     * compressed bit is always 0. */
+    output.writeUInt8(0, 0);
+    output.writeUInt32BE(byteLength, 1);
+    messageBuffer.copy(output, 5);
+    return output;
+  }
+
+  private decompressMessage(
+    message: Buffer,
+    encoding: string
+  ): Buffer | Promise<Buffer> {
+    switch (encoding) {
+      case 'deflate':
+        return inflate(message.subarray(5));
+      case 'gzip':
+        return unzip(message.subarray(5));
+      case 'identity':
+        return message.subarray(5);
+      default:
+        return Promise.reject({
+          code: Status.UNIMPLEMENTED,
+          details: `Received message compressed with unsupported encoding "${encoding}"`,
+        });
+    }
+  }
+
+  private async decompressAndMaybePush(queueEntry: ReadQueueEntry) {
+    if (queueEntry.type !== 'COMPRESSED') {
+      throw new Error(`Invalid queue entry type: ${queueEntry.type}`);
+    }
+
+    const compressed = queueEntry.compressedMessage!.readUInt8(0) === 1;
+    const compressedMessageEncoding = compressed ? this.incomingEncoding : 'identity';
+    const decompressedMessage = await this.decompressMessage(queueEntry.compressedMessage!, compressedMessageEncoding);
+    try {
+      queueEntry.parsedMessage = this.handler.deserialize(decompressedMessage);
+    } catch (err) {
+      this.sendStatus({
+        code: Status.INTERNAL,
+        details: `Error deserializing request: ${(err as Error).message}`
+      });
+      return;
+    }
+    queueEntry.type = 'READABLE';
+    this.maybePushNextMessage();
+  }
+
+  private maybePushNextMessage() {
+    if (this.listener && this.isReadPending && this.readQueue.length > 0 && this.readQueue[0].type !== 'COMPRESSED') {
+      this.isReadPending = false;
+      const nextQueueEntry = this.readQueue.shift()!;
+      if (nextQueueEntry.type === 'READABLE') {
+        this.listener.onReceiveMessage(nextQueueEntry.parsedMessage);
+      } else {
+        // nextQueueEntry.type === 'HALF_CLOSE'
+        this.listener.onReceiveHalfClose();
+      }
+    }
+  }
+
+  private handleDataFrame(data: Buffer) {
+    if (this.checkCancelled()) {
+      return;
+    }
+    trace('Request to ' + this.handler.path + ' received data frame of size ' + data.length);
+    const rawMessages = this.decoder.write(data);
+
+    for (const messageBytes of rawMessages) {
+      this.stream.pause();
+      if (this.maxReceiveMessageSize !== -1 && messageBytes.length - 5 > this.maxReceiveMessageSize) {
+        this.sendStatus({
+          code: Status.RESOURCE_EXHAUSTED,
+          details: `Received message larger than max (${messageBytes.length - 5} vs. ${this.maxReceiveMessageSize})`,
+          metadata: null
+        });
+        return;
+      }
+      const queueEntry: ReadQueueEntry = {
+        type: 'COMPRESSED',
+        compressedMessage: messageBytes,
+        parsedMessage: null
+      };
+      this.readQueue.push(queueEntry);
+      this.decompressAndMaybePush(queueEntry);
+      this.callEventTracker?.addMessageReceived();
+    }
+  }
+  private handleEndEvent() {
+    this.readQueue.push({
+      type: 'HALF_CLOSE',
+      compressedMessage: null,
+      parsedMessage: null
+    });
+    this.receivedHalfClose = true;
+    this.maybePushNextMessage();
+  }
+  start(listener: InterceptingServerListener): void {
+    trace('Request to ' + this.handler.path + ' start called');
+    if (this.checkCancelled()) {
+      return;
+    }
+    this.listener = listener;
+    listener.onReceiveMetadata(this.metadata);
+  }
+  sendMetadata(metadata: Metadata): void {
+    if (this.checkCancelled()) {
+      return;
+    }
+
+    if (this.metadataSent) {
+      return;
+    }
+
+    this.metadataSent = true;
+    const custom = metadata ? metadata.toHttp2Headers() : null;
+    const headers = {
+      ...defaultResponseHeaders,
+      ...defaultCompressionHeaders,
+      ...custom,
+    };
+    this.stream.respond(headers, defaultResponseOptions);
+  }
+  sendMessage(message: any, callback: () => void): void {
+    if (this.checkCancelled()) {
+      return;
+    }
+    let response: Buffer;
+    try {
+      response = this.serializeMessage(message);
+    } catch (e) {
+      this.sendStatus({
+        code: Status.INTERNAL,
+        details: `Error serializing response: ${getErrorMessage(e)}`,
+        metadata: null
+      });
+      return;
+    }
+
+    if (
+      this.maxSendMessageSize !== -1 &&
+      response.length - 5 > this.maxSendMessageSize
+    ) {
+      this.sendStatus({
+        code: Status.RESOURCE_EXHAUSTED,
+        details: `Sent message larger than max (${response.length} vs. ${this.maxSendMessageSize})`,
+        metadata: null
+      });
+      return;
+    }
+    this.maybeSendMetadata();
+    trace('Request to ' + this.handler.path + ' sent data frame of size ' + response.length);
+    this.stream.write(response, error => {
+      if (error) {
+        this.sendStatus({
+          code: Status.INTERNAL,
+          details: `Error writing message: ${getErrorMessage(error)}`,
+          metadata: null
+        });
+        return;
+      }
+      this.callEventTracker?.addMessageSent();
+      callback();
+    });
+  }
+  sendStatus(status: PartialStatusObject): void {
+    if (this.checkCancelled()) {
+      return;
+    }
+    this.notifyOnCancel();
+
+    trace(
+      'Request to method ' +
+        this.handler?.path +
+        ' ended with status code: ' +
+        Status[status.code] +
+        ' details: ' +
+        status.details
+    );
+
+    if (this.stream.headersSent) {
+      if (!this.wantTrailers) {
+        this.wantTrailers = true;
+        this.stream.once('wantTrailers', () => {
+          if (this.callEventTracker && !this.streamEnded) {
+            this.streamEnded = true;
+            this.callEventTracker.onStreamEnd(true);
+            this.callEventTracker.onCallEnd(status);
+          }
+          const trailersToSend = {
+            [GRPC_STATUS_HEADER]: status.code,
+            [GRPC_MESSAGE_HEADER]: encodeURI(status.details),
+            ...status.metadata?.toHttp2Headers(),
+          };
+
+          this.stream.sendTrailers(trailersToSend);
+        });
+        this.stream.end();
+      }
+    } else {
+      if (this.callEventTracker && !this.streamEnded) {
+        this.streamEnded = true;
+        this.callEventTracker.onStreamEnd(true);
+        this.callEventTracker.onCallEnd(status);
+      }
+      // Trailers-only response
+      const trailersToSend = {
+        [GRPC_STATUS_HEADER]: status.code,
+        [GRPC_MESSAGE_HEADER]: encodeURI(status.details),
+        ...defaultResponseHeaders,
+        ...status.metadata?.toHttp2Headers(),
+      };
+      this.stream.respond(trailersToSend, { endStream: true });
+    }
+  }
+  startRead(): void {
+    trace('Request to ' + this.handler.path + ' startRead called');
+    if (this.checkCancelled()) {
+      return;
+    }
+    this.isReadPending = true;
+    if (this.readQueue.length === 0) {
+      if (!this.receivedHalfClose) {
+        this.stream.resume();
+      }
+    } else {
+      this.maybePushNextMessage();
+    }
+  }
+  getPeer(): string {
+    const socket = this.stream.session?.socket;
+    if (socket?.remoteAddress) {
+      if (socket.remotePort) {
+        return `${socket.remoteAddress}:${socket.remotePort}`;
+      } else {
+        return socket.remoteAddress;
+      }
+    } else {
+      return 'unknown';
+    }
+  }
+  getDeadline(): Deadline {
+    return this.deadline;
+  }
+}
+
+export function getServerInterceptingCall(
+  interceptors: ServerInterceptor[],
+  stream: http2.ServerHttp2Stream,
+  headers: http2.IncomingHttpHeaders,
+  callEventTracker: CallEventTracker | null,
+  handler: Handler<any, any>,
+  options: ChannelOptions
+) {
+
+  const methodDefinition: ServerMethodDefinition<any, any> = {
+    path: handler.path,
+    requestStream: handler.type === 'clientStream' || handler.type === 'bidi',
+    responseStream: handler.type === 'serverStream' || handler.type === 'bidi',
+    requestDeserialize: handler.deserialize,
+    responseSerialize: handler.serialize
+  }
+  const baseCall = new BaseServerInterceptingCall(stream, headers, callEventTracker, handler, options);
+  return interceptors.reduce((call: ServerInterceptingCallInterface, interceptor: ServerInterceptor) => {
+    return interceptor(methodDefinition, call);
+  }, baseCall);
+}
diff --git a/packages/grpc-js/src/server.ts b/packages/grpc-js/src/server.ts
index adc7f299f..31851b832 100644
--- a/packages/grpc-js/src/server.ts
+++ b/packages/grpc-js/src/server.ts
@@ -28,20 +28,18 @@ import {
   HandleCall,
   Handler,
   HandlerType,
-  Http2ServerCallStream,
   sendUnaryData,
   ServerDuplexStream,
   ServerDuplexStreamImpl,
   ServerReadableStream,
-  ServerReadableStreamImpl,
   ServerStreamingHandler,
   ServerUnaryCall,
-  ServerUnaryCallImpl,
   ServerWritableStream,
   ServerWritableStreamImpl,
   UnaryHandler,
   ServerErrorResponse,
   ServerStatusResponse,
+  serverErrorToStatus,
 } from './server-call';
 import { ServerCredentials } from './server-credentials';
 import { ChannelOptions } from './channel-options';
@@ -72,6 +70,9 @@ import {
   unregisterChannelzRef,
 } from './channelz';
 import { CipherNameAndProtocol, TLSSocket } from 'tls';
+import { ServerInterceptingCallInterface, ServerInterceptor, getServerInterceptingCall } from './server-interceptors';
+import { PartialStatusObject } from './call-interface';
+import { CallEventTracker } from './transport';
 
 const UNLIMITED_CONNECTION_AGE_MS = ~(1 << 31);
 const KEEPALIVE_MAX_TIME_MS = ~(1 << 31);
@@ -109,7 +110,7 @@ function deprecate(message: string) {
 
 function getUnimplementedStatusResponse(
   methodName: string
-): Partial<ServiceError> {
+): PartialStatusObject {
   return {
     code: Status.UNIMPLEMENTED,
     details: `The server does not implement the method ${methodName}`,
@@ -219,6 +220,10 @@ interface Http2ServerInfo {
   sessions: Set<http2.ServerHttp2Session>;
 }
 
+export interface ServerOptions extends ChannelOptions {
+  interceptors?: ServerInterceptor[]
+}
+
 export class Server {
   private boundPorts: Map<string, BoundPort>= new Map();
   private http2Servers: Map<AnyHttp2Server, Http2ServerInfo> = new Map();
@@ -234,7 +239,7 @@ export class Server {
    */
   private started = false;
   private shutdown = false;
-  private options: ChannelOptions;
+  private options: ServerOptions;
   private serverAddressString = 'null';
 
   // Channelz Info
@@ -251,13 +256,15 @@ export class Server {
   private readonly keepaliveTimeMs: number;
   private readonly keepaliveTimeoutMs: number;
 
+  private readonly interceptors: ServerInterceptor[];
+
   /**
    * Options that will be used to construct all Http2Server instances for this
    * Server.
    */
   private commonServerOptions: http2.ServerOptions;
 
-  constructor(options?: ChannelOptions) {
+  constructor(options?: ServerOptions) {
     this.options = options ?? {};
     if (this.options['grpc.enable_channelz'] === 0) {
       this.channelzEnabled = false;
@@ -296,6 +303,7 @@ export class Server {
         maxConcurrentStreams: this.options['grpc.max_concurrent_streams'],
       };
     }
+    this.interceptors = this.options.interceptors ?? [];
     this.trace('Server constructed');
   }
 
@@ -1072,23 +1080,24 @@ export class Server {
     return handler;
   }
 
-  private _respondWithError<T extends Partial<ServiceError>>(
-    err: T,
+  private _respondWithError(
+    err: PartialStatusObject,
     stream: http2.ServerHttp2Stream,
     channelzSessionInfo: ChannelzSessionInfo | null = null
   ) {
-    const call = new Http2ServerCallStream(stream, null!, this.options);
-
-    if (err.code === undefined) {
-      err.code = Status.INTERNAL;
-    }
+    const trailersToSend = {
+      'grpc-status': err.code ?? Status.INTERNAL,
+      'grpc-message': err.details,
+      [http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
+      [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
+      ...err.metadata?.toHttp2Headers()
+    };
+    stream.respond(trailersToSend, {endStream: true});
 
     if (this.channelzEnabled) {
       this.callTracker.addCallFailed();
       channelzSessionInfo?.streamTracker.addCallFailed();
     }
-
-    call.sendError(err);
   }
 
   private _channelzHandler(
@@ -1120,39 +1129,44 @@ export class Server {
       return;
     }
 
-    const call = new Http2ServerCallStream(stream, handler, this.options);
-
-    call.once('callEnd', (code: Status) => {
-      if (code === Status.OK) {
-        this.callTracker.addCallSucceeded();
-      } else {
-        this.callTracker.addCallFailed();
-      }
-    });
-
-    if (channelzSessionInfo) {
-      call.once('streamEnd', (success: boolean) => {
-        if (success) {
-          channelzSessionInfo.streamTracker.addCallSucceeded();
+    let callEventTracker: CallEventTracker = {
+      addMessageSent: () => {
+        if (channelzSessionInfo) {
+          channelzSessionInfo.messagesSent += 1;
+          channelzSessionInfo.lastMessageSentTimestamp = new Date();
+        }
+      },
+      addMessageReceived: () => {
+        if (channelzSessionInfo) {
+          channelzSessionInfo.messagesReceived += 1;
+          channelzSessionInfo.lastMessageReceivedTimestamp = new Date();
+        }
+      },
+      onCallEnd: status => {
+        if (status.code === Status.OK) {
+          this.callTracker.addCallSucceeded();
         } else {
-          channelzSessionInfo.streamTracker.addCallFailed();
+          this.callTracker.addCallFailed();
         }
-      });
-      call.on('sendMessage', () => {
-        channelzSessionInfo.messagesSent += 1;
-        channelzSessionInfo.lastMessageSentTimestamp = new Date();
-      });
-      call.on('receiveMessage', () => {
-        channelzSessionInfo.messagesReceived += 1;
-        channelzSessionInfo.lastMessageReceivedTimestamp = new Date();
-      });
+      },
+      onStreamEnd: success => {
+        if (channelzSessionInfo) {
+          if (success) {
+            channelzSessionInfo.streamTracker.addCallSucceeded();
+          } else {
+            channelzSessionInfo.streamTracker.addCallFailed();
+          }
+        }
+      }
     }
 
-    if (!this._runHandlerForCall(call, handler, headers)) {
+    const call = getServerInterceptingCall(this.interceptors, stream, headers, callEventTracker, handler, this.options);
+
+    if (!this._runHandlerForCall(call, handler)) {
       this.callTracker.addCallFailed();
       channelzSessionInfo?.streamTracker.addCallFailed();
 
-      call.sendError({
+      call.sendStatus({
         code: Status.INTERNAL,
         details: `Unknown handler type: ${handler.type}`,
       });
@@ -1179,9 +1193,10 @@ export class Server {
       return;
     }
 
-    const call = new Http2ServerCallStream(stream, handler, this.options);
-    if (!this._runHandlerForCall(call, handler, headers)) {
-      call.sendError({
+    const call = getServerInterceptingCall(this.interceptors, stream, headers, null, handler, this.options);
+
+    if (!this._runHandlerForCall(call, handler)) {
+      call.sendStatus({
         code: Status.INTERNAL,
         details: `Unknown handler type: ${handler.type}`,
       });
@@ -1189,38 +1204,27 @@ export class Server {
   }
 
   private _runHandlerForCall(
-    call: Http2ServerCallStream<any, any>,
-    handler: Handler<any, any>,
-    headers: http2.IncomingHttpHeaders
+    call: ServerInterceptingCallInterface,
+    handler: Handler<any, any>
   ): boolean {
-    const metadata = call.receiveMetadata(headers);
-    const encoding =
-      (metadata.get('grpc-encoding')[0] as string | undefined) ?? 'identity';
-    metadata.remove('grpc-encoding');
 
     const { type } = handler;
     if (type === 'unary') {
-      handleUnary(call, handler as UntypedUnaryHandler, metadata, encoding);
+      handleUnary(call, handler as UntypedUnaryHandler);
     } else if (type === 'clientStream') {
       handleClientStreaming(
         call,
-        handler as UntypedClientStreamingHandler,
-        metadata,
-        encoding
+        handler as UntypedClientStreamingHandler
       );
     } else if (type === 'serverStream') {
       handleServerStreaming(
         call,
-        handler as UntypedServerStreamingHandler,
-        metadata,
-        encoding
+        handler as UntypedServerStreamingHandler
       );
     } else if (type === 'bidi') {
       handleBidiStreaming(
         call,
-        handler as UntypedBidiStreamingHandler,
-        metadata,
-        encoding
+        handler as UntypedBidiStreamingHandler
       );
     } else {
       return false;
@@ -1365,52 +1369,83 @@ export class Server {
 }
 
 async function handleUnary<RequestType, ResponseType>(
-  call: Http2ServerCallStream<RequestType, ResponseType>,
-  handler: UnaryHandler<RequestType, ResponseType>,
-  metadata: Metadata,
-  encoding: string
+  call: ServerInterceptingCallInterface,
+  handler: UnaryHandler<RequestType, ResponseType>
 ): Promise<void> {
-  try {
-    const request = await call.receiveUnaryMessage(encoding);
+  let stream: ServerUnaryCall<RequestType, ResponseType>;
 
-    if (request === undefined || call.cancelled) {
+  function respond(
+    err: ServerErrorResponse | ServerStatusResponse | null,
+    value?: ResponseType | null,
+    trailer?: Metadata,
+    flags?: number
+  ) {
+    if (err) {
+      call.sendStatus(serverErrorToStatus(err, trailer));
       return;
     }
+    call.sendMessage(value, () => {
+      call.sendStatus({
+        code: Status.OK,
+        details: 'OK',
+        metadata: trailer ?? null
+      });
+    });
+  }
 
-    const emitter = new ServerUnaryCallImpl<RequestType, ResponseType>(
-      call,
-      metadata,
-      request
-    );
-
-    handler.func(
-      emitter,
-      (
-        err: ServerErrorResponse | ServerStatusResponse | null,
-        value?: ResponseType | null,
-        trailer?: Metadata,
-        flags?: number
-      ) => {
-        call.sendUnaryMessage(err, value, trailer, flags);
+  let requestMetadata: Metadata;
+  let requestMessage: RequestType | null = null;
+  call.start({
+    onReceiveMetadata(metadata) {
+      requestMetadata = metadata;
+      call.startRead();
+    },
+    onReceiveMessage(message) {
+      if (requestMessage) {
+        call.sendStatus({
+          code: Status.UNIMPLEMENTED,
+          details: `Received a second request message for server streaming method ${handler.path}`,
+          metadata: null
+        });
+        return;
       }
-    );
-  } catch (err) {
-    call.sendError(err as ServerErrorResponse);
-  }
+      requestMessage = message;
+      call.startRead();
+    },
+    onReceiveHalfClose() {
+      if (!requestMessage) {
+        call.sendStatus({
+          code: Status.UNIMPLEMENTED,
+          details: `Received no request message for server streaming method ${handler.path}`,
+          metadata: null
+        });
+        return;
+      }
+      stream = new ServerWritableStreamImpl(handler.path, call, requestMetadata, requestMessage);
+      try {
+        handler.func(stream, respond);
+      } catch (err) {
+        call.sendStatus({
+          code: Status.UNKNOWN,
+          details: `Server method handler threw error ${(err as Error).message}`,
+          metadata: null
+        });
+      }
+    },
+    onCancel() {
+      if (stream) {
+        stream.cancelled = true;
+        stream.emit('cancelled', 'cancelled');
+      }
+    },
+  });
 }
 
 function handleClientStreaming<RequestType, ResponseType>(
-  call: Http2ServerCallStream<RequestType, ResponseType>,
-  handler: ClientStreamingHandler<RequestType, ResponseType>,
-  metadata: Metadata,
-  encoding: string
+  call: ServerInterceptingCallInterface,
+  handler: ClientStreamingHandler<RequestType, ResponseType>
 ): void {
-  const stream = new ServerReadableStreamImpl<RequestType, ResponseType>(
-    call,
-    metadata,
-    handler.deserialize,
-    encoding
-  );
+  let stream: ServerReadableStream<RequestType, ResponseType>;
 
   function respond(
     err: ServerErrorResponse | ServerStatusResponse | null,
@@ -1418,61 +1453,134 @@ function handleClientStreaming<RequestType, ResponseType>(
     trailer?: Metadata,
     flags?: number
   ) {
-    stream.destroy();
-    call.sendUnaryMessage(err, value, trailer, flags);
-  }
-
-  if (call.cancelled) {
-    return;
-  }
-
-  stream.on('error', respond);
-  handler.func(stream, respond);
-}
-
-async function handleServerStreaming<RequestType, ResponseType>(
-  call: Http2ServerCallStream<RequestType, ResponseType>,
-  handler: ServerStreamingHandler<RequestType, ResponseType>,
-  metadata: Metadata,
-  encoding: string
-): Promise<void> {
-  try {
-    const request = await call.receiveUnaryMessage(encoding);
-
-    if (request === undefined || call.cancelled) {
+    if (err) {
+      call.sendStatus(serverErrorToStatus(err, trailer));
       return;
     }
+    call.sendMessage(value, () => {
+      call.sendStatus({
+        code: Status.OK,
+        details: 'OK',
+        metadata: trailer ?? null
+      });
+    });
+  }
 
-    const stream = new ServerWritableStreamImpl<RequestType, ResponseType>(
-      call,
-      metadata,
-      handler.serialize,
-      request
-    );
+  call.start({
+    onReceiveMetadata(metadata) {
+      stream = new ServerDuplexStreamImpl(handler.path, call, metadata);
+      try {
+        handler.func(stream, respond);
+      } catch (err) {
+        call.sendStatus({
+          code: Status.UNKNOWN,
+          details: `Server method handler threw error ${(err as Error).message}`,
+          metadata: null
+        });
+      }
+    },
+    onReceiveMessage(message) {
+      stream.push(message);
+    },
+    onReceiveHalfClose() {
+      stream.push(null);
+    },
+    onCancel() {
+      if (stream) {
+        stream.cancelled = true;
+        stream.emit('cancelled', 'cancelled');
+        stream.destroy();
+      }
+    },
+  });
+}
 
-    handler.func(stream);
-  } catch (err) {
-    call.sendError(err as ServerErrorResponse);
-  }
+function handleServerStreaming<RequestType, ResponseType>(
+  call: ServerInterceptingCallInterface,
+  handler: ServerStreamingHandler<RequestType, ResponseType>
+): void {
+  let stream: ServerWritableStream<RequestType, ResponseType>;
+
+  let requestMetadata: Metadata;
+  let requestMessage: RequestType | null = null;
+  call.start({
+    onReceiveMetadata(metadata) {
+      requestMetadata = metadata;
+      call.startRead();
+    },
+    onReceiveMessage(message) {
+      if (requestMessage) {
+        call.sendStatus({
+          code: Status.UNIMPLEMENTED,
+          details: `Received a second request message for server streaming method ${handler.path}`,
+          metadata: null
+        });
+        return;
+      }
+      requestMessage = message;
+      call.startRead();
+    },
+    onReceiveHalfClose() {
+      if (!requestMessage) {
+        call.sendStatus({
+          code: Status.UNIMPLEMENTED,
+          details: `Received no request message for server streaming method ${handler.path}`,
+          metadata: null
+        });
+        return;
+      }
+      stream = new ServerWritableStreamImpl(handler.path, call, requestMetadata, requestMessage);
+      try {
+        handler.func(stream);
+      } catch (err) {
+        call.sendStatus({
+          code: Status.UNKNOWN,
+          details: `Server method handler threw error ${(err as Error).message}`,
+          metadata: null
+        });
+      }
+    },
+    onCancel() {
+      if (stream) {
+        stream.cancelled = true;
+        stream.emit('cancelled', 'cancelled');
+        stream.destroy();
+      }
+    },
+  });
 }
 
 function handleBidiStreaming<RequestType, ResponseType>(
-  call: Http2ServerCallStream<RequestType, ResponseType>,
-  handler: BidiStreamingHandler<RequestType, ResponseType>,
-  metadata: Metadata,
-  encoding: string
+  call: ServerInterceptingCallInterface,
+  handler: BidiStreamingHandler<RequestType, ResponseType>
 ): void {
-  const stream = new ServerDuplexStreamImpl<RequestType, ResponseType>(
-    call,
-    metadata,
-    handler.serialize,
-    handler.deserialize,
-    encoding
-  );
-
-  if (call.cancelled) {
-    return;
-  }
-
-  handler.func(stream);
+  let stream: ServerDuplexStream<RequestType, ResponseType>;
+
+  call.start({
+    onReceiveMetadata(metadata) {
+      stream = new ServerDuplexStreamImpl(handler.path, call, metadata);
+      try {
+        handler.func(stream);
+      } catch (err) {
+        call.sendStatus({
+          code: Status.UNKNOWN,
+          details: `Server method handler threw error ${(err as Error).message}`,
+          metadata: null
+        });
+      }
+    },
+    onReceiveMessage(message) {
+      stream.push(message);
+    },
+    onReceiveHalfClose() {
+      stream.push(null);
+    },
+    onCancel() {
+      if (stream) {
+        stream.cancelled = true;
+        stream.emit('cancelled', 'cancelled');
+        stream.destroy();
+      }
+    },
+  });
 }
diff --git a/packages/grpc-js/src/transport.ts b/packages/grpc-js/src/transport.ts
index 39ca69383..b21581375 100644
--- a/packages/grpc-js/src/transport.ts
+++ b/packages/grpc-js/src/transport.ts
@@ -23,7 +23,7 @@ import {
   PeerCertificate,
   TLSSocket,
 } from 'tls';
-import { StatusObject } from './call-interface';
+import { PartialStatusObject } from './call-interface';
 import { ChannelCredentials } from './channel-credentials';
 import { ChannelOptions } from './channel-options';
 import {
@@ -72,7 +72,7 @@ const KEEPALIVE_TIMEOUT_MS = 20000;
 export interface CallEventTracker {
   addMessageSent(): void;
   addMessageReceived(): void;
-  onCallEnd(status: StatusObject): void;
+  onCallEnd(status: PartialStatusObject): void;
   onStreamEnd(success: boolean): void;
 }
 
diff --git a/packages/grpc-js/test/common.ts b/packages/grpc-js/test/common.ts
index 20352fb2f..f6aeed103 100644
--- a/packages/grpc-js/test/common.ts
+++ b/packages/grpc-js/test/common.ts
@@ -72,7 +72,7 @@ const serviceImpl = {
 export class TestServer {
   private server: grpc.Server;
   public port: number | null = null;
-  constructor(public useTls: boolean, options?: grpc.ChannelOptions) {
+  constructor(public useTls: boolean, options?: grpc.ServerOptions) {
     this.server = new grpc.Server(options);
     this.server.addService(echoService.service, serviceImpl);
   }
@@ -92,7 +92,6 @@ export class TestServer {
           return;
         }
         this.port = port;
-        this.server.start();
         resolve();
       });
     });
@@ -130,6 +129,10 @@ export class TestClient {
     this.client.echo({}, callback);
   }
 
+  sendRequestWithMetadata(metadata: grpc.Metadata, callback: (error?: grpc.ServiceError) => void) {
+    this.client.echo({}, metadata, callback);
+  }
+
   getChannelState() {
     return this.client.getChannel().getConnectivityState(false);
   }
diff --git a/packages/grpc-js/test/test-channelz.ts b/packages/grpc-js/test/test-channelz.ts
index e740c4329..5c3dc3d7d 100644
--- a/packages/grpc-js/test/test-channelz.ts
+++ b/packages/grpc-js/test/test-channelz.ts
@@ -495,12 +495,12 @@ describe('Channelz', () => {
                                 assert.strictEqual(
                                   +serverSocketResult.socket.data
                                     .streams_succeeded,
-                                  0
+                                  1
                                 );
                                 assert.strictEqual(
                                   +serverSocketResult.socket.data
                                     .streams_failed,
-                                  1
+                                  0
                                 );
                                 assert.strictEqual(
                                   +serverSocketResult.socket.data
diff --git a/packages/grpc-js/test/test-server-deadlines.ts b/packages/grpc-js/test/test-server-deadlines.ts
index 2a966e664..43e39808a 100644
--- a/packages/grpc-js/test/test-server-deadlines.ts
+++ b/packages/grpc-js/test/test-server-deadlines.ts
@@ -110,8 +110,8 @@ describe('Server deadlines', () => {
       {},
       (error: any, response: any) => {
         assert(error);
-        assert.strictEqual(error.code, grpc.status.OUT_OF_RANGE);
-        assert.strictEqual(error.details, 'Invalid deadline');
+        assert.strictEqual(error.code, grpc.status.INTERNAL);
+        assert.match(error.details, /^Invalid grpc-timeout value/);
         done();
       }
     );
diff --git a/packages/grpc-js/test/test-server-interceptors.ts b/packages/grpc-js/test/test-server-interceptors.ts
new file mode 100644
index 000000000..b6b5b55f2
--- /dev/null
+++ b/packages/grpc-js/test/test-server-interceptors.ts
@@ -0,0 +1,296 @@
+/*
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import * as assert from 'assert';
+import * as path from 'path';
+import * as grpc from '../src';
+import { TestClient, loadProtoFile } from './common';
+
+const protoFile = path.join(__dirname, 'fixtures', 'echo_service.proto');
+const echoService = loadProtoFile(protoFile)
+  .EchoService as grpc.ServiceClientConstructor;
+
+const AUTH_HEADER_KEY = 'auth';
+const AUTH_HEADER_ALLOWED_VALUE = 'allowed';
+const testAuthInterceptor: grpc.ServerInterceptor = (methodDescriptor, call) => {
+  return new grpc.ServerInterceptingCall(call, {
+    start: next => {
+      const authListener: grpc.ServerListener = {
+        onReceiveMetadata: (metadata, mdNext) => {
+          if (metadata.get(AUTH_HEADER_KEY)?.[0] !== AUTH_HEADER_ALLOWED_VALUE) {
+            call.sendStatus({
+              code: grpc.status.UNAUTHENTICATED,
+              details: 'Auth metadata not correct'
+            });
+          } else {
+            mdNext(metadata);
+          }
+        }
+      };
+      next(authListener);
+    }
+  });
+};
+
+let eventCounts = {
+  receiveMetadata: 0,
+  receiveMessage: 0,
+  receiveHalfClose: 0,
+  sendMetadata: 0,
+  sendMessage: 0,
+  sendStatus: 0
+};
+
+function resetEventCounts() {
+  eventCounts = {
+    receiveMetadata: 0,
+    receiveMessage: 0,
+    receiveHalfClose: 0,
+    sendMetadata: 0,
+    sendMessage: 0,
+    sendStatus: 0
+  };
+}
+
+/**
+ * Test interceptor to verify that interceptors see each expected event by
+ * counting each kind of event.
+ * @param methodDescription
+ * @param call
+ */
+const testLoggingInterceptor: grpc.ServerInterceptor = (methodDescription, call) => {
+  return new grpc.ServerInterceptingCall(call, {
+    start: next => {
+      next({
+        onReceiveMetadata: (metadata, mdNext) => {
+          eventCounts.receiveMetadata += 1;
+          mdNext(metadata);
+        },
+        onReceiveMessage: (message, messageNext) => {
+          eventCounts.receiveMessage += 1;
+          messageNext(message);
+        },
+        onReceiveHalfClose: hcNext => {
+          eventCounts.receiveHalfClose += 1;
+          hcNext();
+        }
+      });
+    },
+    sendMetadata: (metadata, mdNext) => {
+      eventCounts.sendMetadata += 1;
+      mdNext(metadata);
+    },
+    sendMessage: (message, messageNext) => {
+      eventCounts.sendMessage += 1;
+      messageNext(message);
+    },
+    sendStatus: (status, statusNext) => {
+      eventCounts.sendStatus += 1;
+      statusNext(status);
+    }
+  });
+};
+
+const testHeaderInjectionInterceptor: grpc.ServerInterceptor = (methodDescriptor, call) => {
+  return new grpc.ServerInterceptingCall(call, {
+    start: next => {
+      const authListener: grpc.ServerListener = {
+        onReceiveMetadata: (metadata, mdNext) => {
+          metadata.set('injected-header', 'present');
+          mdNext(metadata);
+        }
+      };
+      next(authListener);
+    }
+  });
+};
+
+describe('Server interceptors', () => {
+  describe('Auth-type interceptor', () => {
+    let server: grpc.Server;
+    let client: TestClient;
+    /* Tests that an interceptor can entirely prevent the handler from being
+     * invoked, based on the contents of the metadata. */
+    before(done => {
+      server = new grpc.Server({interceptors: [testAuthInterceptor]});
+      server.addService(echoService.service, {
+        echo: (
+          call: grpc.ServerUnaryCall<any, any>,
+          callback: grpc.sendUnaryData<any>
+        ) => {
+          // A test will fail if a request makes it to the handler without the correct auth header
+          assert.strictEqual(call.metadata.get(AUTH_HEADER_KEY)?.[0], AUTH_HEADER_ALLOWED_VALUE);
+          callback(null, call.request);
+        },
+      });
+      server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
+        assert.ifError(error);
+        client = new TestClient(port, false);
+        done();
+      });
+    });
+    after(done => {
+      client.close();
+      server.tryShutdown(done);
+    });
+    it('Should accept a request with the expected header', done => {
+      const requestMetadata = new grpc.Metadata();
+      requestMetadata.set(AUTH_HEADER_KEY, AUTH_HEADER_ALLOWED_VALUE);
+      client.sendRequestWithMetadata(requestMetadata, done);
+    });
+    it('Should reject a request without the expected header', done => {
+      const requestMetadata = new grpc.Metadata();
+      requestMetadata.set(AUTH_HEADER_KEY, 'not allowed');
+      client.sendRequestWithMetadata(requestMetadata, error => {
+        assert.strictEqual(error?.code, grpc.status.UNAUTHENTICATED);
+        done();
+      });
+    });
+  });
+  describe('Logging-type interceptor', () => {
+    let server: grpc.Server;
+    let client: TestClient;
+    before(done => {
+      server = new grpc.Server({interceptors: [testLoggingInterceptor]});
+      server.addService(echoService.service, {
+        echo: (
+          call: grpc.ServerUnaryCall<any, any>,
+          callback: grpc.sendUnaryData<any>
+        ) => {
+          call.sendMetadata(new grpc.Metadata());
+          callback(null, call.request);
+        },
+      });
+      server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
+        assert.ifError(error);
+        client = new TestClient(port, false);
+        done();
+      });
+    });
+    after(done => {
+      client.close();
+      server.tryShutdown(done);
+    });
+    beforeEach(() => {
+      resetEventCounts();
+    });
+    it('Should see every event once', done => {
+      client.sendRequest(error => {
+        assert.ifError(error);
+        assert.deepStrictEqual(eventCounts, {
+          receiveMetadata: 1,
+          receiveMessage: 1,
+          receiveHalfClose: 1,
+          sendMetadata: 1,
+          sendMessage: 1,
+          sendStatus: 1
+        });
+        done();
+      });
+    });
+  });
+  describe('Header injection interceptor', () => {
+    let server: grpc.Server;
+    let client: TestClient;
+    before(done => {
+      server = new grpc.Server({interceptors: [testHeaderInjectionInterceptor]});
+      server.addService(echoService.service, {
+        echo: (
+          call: grpc.ServerUnaryCall<any, any>,
+          callback: grpc.sendUnaryData<any>
+        ) => {
+          assert.strictEqual(call.metadata.get('injected-header')?.[0], 'present');
+          callback(null, call.request);
+        },
+      });
+      server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
+        assert.ifError(error);
+        client = new TestClient(port, false);
+        done();
+      });
+    });
+    after(done => {
+      client.close();
+      server.tryShutdown(done);
+    });
+    it('Should inject the header for the handler to see', done => {
+      client.sendRequest(done);
+    });
+  });
+  describe('Multiple interceptors', () => {
+    let server: grpc.Server;
+    let client: TestClient;
+    before(done => {
+      server = new grpc.Server({interceptors: [testAuthInterceptor, testLoggingInterceptor, testHeaderInjectionInterceptor]});
+      server.addService(echoService.service, {
+        echo: (
+          call: grpc.ServerUnaryCall<any, any>,
+          callback: grpc.sendUnaryData<any>
+        ) => {
+          assert.strictEqual(call.metadata.get(AUTH_HEADER_KEY)?.[0], AUTH_HEADER_ALLOWED_VALUE);
+          assert.strictEqual(call.metadata.get('injected-header')?.[0], 'present');
+          call.sendMetadata(new grpc.Metadata());
+          callback(null, call.request);
+        },
+      });
+      server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
+        assert.ifError(error);
+        client = new TestClient(port, false);
+        done();
+      });
+    });
+    after(done => {
+      client.close();
+      server.tryShutdown(done);
+    });
+    beforeEach(() => {
+      resetEventCounts();
+    });
+    it('Should not log requests rejected by auth', done => {
+      const requestMetadata = new grpc.Metadata();
+      requestMetadata.set(AUTH_HEADER_KEY, 'not allowed');
+      client.sendRequestWithMetadata(requestMetadata, error => {
+        assert.strictEqual(error?.code, grpc.status.UNAUTHENTICATED);
+        assert.deepStrictEqual(eventCounts, {
+          receiveMetadata: 0,
+          receiveMessage: 0,
+          receiveHalfClose: 0,
+          sendMetadata: 0,
+          sendMessage: 0,
+          sendStatus: 0
+        });
+        done();
+      });
+    });
+    it('Should log requests accepted by auth', done => {
+      const requestMetadata = new grpc.Metadata();
+      requestMetadata.set(AUTH_HEADER_KEY, AUTH_HEADER_ALLOWED_VALUE);
+      client.sendRequestWithMetadata(requestMetadata, error => {
+        assert.ifError(error);
+        assert.deepStrictEqual(eventCounts, {
+          receiveMetadata: 1,
+          receiveMessage: 1,
+          receiveHalfClose: 1,
+          sendMetadata: 1,
+          sendMessage: 1,
+          sendStatus: 1
+        });
+        done();
+      });
+    });
+  });
+});