From f5508d9801dd52c1bc9929d0c5b669d25ad2fc3f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C3=A1lint=20B=C3=B6de?= <balint.bode@gmail.com>
Date: Tue, 29 Jun 2021 23:56:12 +0200
Subject: [PATCH] Fix a typo on the inner invoker parameters (#492)

---
 lib/src/client/client.dart                    |  2 +-
 .../client_tests/client_interceptor_test.dart | 47 ++++++++++++++++++-
 2 files changed, 47 insertions(+), 2 deletions(-)

diff --git a/lib/src/client/client.dart b/lib/src/client/client.dart
index 8814064..1ba6d00 100644
--- a/lib/src/client/client.dart
+++ b/lib/src/client/client.dart
@@ -60,7 +60,7 @@ regenerate these stubs using  protobuf compiler plugin version 19.2.0 or newer.
   ResponseStream<R> $createStreamingCall<Q, R>(
       ClientMethod<Q, R> method, Stream<Q> requests,
       {CallOptions? options}) {
-    var invoker = (method, request, options) =>
+    var invoker = (method, requests, options) =>
         ResponseStream<R>(_channel.createCall<Q, R>(method, requests, options));
 
     for (final interceptor in _interceptors.reversed) {
diff --git a/test/client_tests/client_interceptor_test.dart b/test/client_tests/client_interceptor_test.dart
index 92a25b5..0fbaf59 100644
--- a/test/client_tests/client_interceptor_test.dart
+++ b/test/client_tests/client_interceptor_test.dart
@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:grpc/grpc.dart';
 import 'package:grpc/src/client/interceptor.dart';
 import 'package:http2/transport.dart';
@@ -44,7 +46,11 @@ class FakeInterceptor implements ClientInterceptor {
       ClientStreamingInvoker<Q, R> invoker) {
     _invocations.add(InterceptorInvocation(_id, _unary, ++_streaming));
 
-    return invoker(method, requests, _inject(options));
+    final requestStream = _id > 10
+        ? requests.cast<int>().map((req) => req * _id).cast<Q>()
+        : requests;
+
+    return invoker(method, requestStream, _inject(options));
   }
 
   CallOptions _inject(CallOptions options) {
@@ -206,4 +212,43 @@ void main() {
     harness.tearDown();
     FakeInterceptor.tearDown();
   });
+
+  test('streaming interceptors with request stream operation', () async {
+    const ids = [21, 73, 37];
+
+    final harness = ClientHarness()
+      ..interceptors = ids.map((e) => FakeInterceptor(e))
+      ..setUp();
+
+    const requests = [1, 15, 7];
+    const multiplier = 21 * 73 * 37;
+    const responses = [1 * multiplier, 15 * multiplier, 7 * multiplier];
+
+    var index = 0;
+
+    void handleRequest(StreamMessage message) {
+      final data = validateDataMessage(message);
+      final request = mockDecode(data.data);
+      if (index++ == 0) {
+        harness.sendResponseHeader();
+      }
+      harness.sendResponseValue(request);
+    }
+
+    void handleDone() {
+      harness.sendResponseTrailer();
+    }
+
+    await harness.runTest(
+      clientCall:
+          harness.client.bidirectional(Stream.fromIterable(requests)).toList(),
+      expectedResult: responses,
+      expectedPath: '/Test/Bidirectional',
+      serverHandlers: [handleRequest, handleRequest, handleRequest],
+      doneHandler: handleDone,
+    );
+
+    harness.tearDown();
+    FakeInterceptor.tearDown();
+  });
 }