Interceptor for server-stream grpc-web request unexpectedly receives message as an AsyncIterable #1647

import { createClient, type Interceptor } from '@connectrpc/connect';
import { createGrpcWebTransport } from '@connectrpc/connect-web';
import { create } from '@bufbuild/protobuf';
// Using the standard Eliza service for reproducibility
import { ElizaService, IntroduceRequestSchema } from './gen/eliza_pb.js';

// util for handling AsyncIterable
async function* handleStreamMessages<T>(
  stream: AsyncIterable<T>,
  handler: (message: T) => T | void,
) {
  for await (const m of stream) {
    const message = handler(m) || m;
    yield message as T;
  }
}

// Interceptor to log request messages
const loggingInterceptor: Interceptor = next => async (req) => {
  console.log('Interceptor triggered for method:', req.method.name);
  console.log('Is this a streaming request?', req.stream);
  console.log('Request message:', req.message);
  let currentRequest = req;
  if (req.stream) {
    // This must be client-streams
    currentRequest = {
      ...req,
      message: handleStreamMessages(
        req.message,
        msg => console.log('Stream outgoing message:', msg),
      ),
    };
  } else {
    console.log('Unary outgoing message', req.message);
  }
  const res = await next(currentRequest);
  if (res.stream) {
    // it's ok for server-stream
    return {
      ...res,
      message: handleStreamMessages(
        res.message,
        msg => console.log('Stream incoming message', msg),
      ),
    };
  }
  console.log('Unary incoming message', req.message);
  return res;
};

// Client setup
const transport = createGrpcWebTransport({
  baseUrl: 'https://demo.connectrpc.com', // Using the official demo server
  interceptors: [loggingInterceptor],
});

const client = createClient(ElizaService, transport);

// Make the server-streaming call
async function callElizaIntroduce() {
  const request = create(IntroduceRequestSchema, { name: 'John' });
  console.log('Client is sending request with payload:', request);

  try {
    const stream = client.introduce(request);
    // eslint-disable-next-line no-restricted-syntax
    for await (const response of stream) {
      console.log('Eliza says:', response.sentence);
    }
  } catch (e) {
    console.error('Error during streaming call:', e);
  }
}

callElizaIntroduce();