From c1b64e3493779c0e105b901d2a8f1fb8512d8bef Mon Sep 17 00:00:00 2001
From: ganmacs <ganmacs@gmail.com>
Date: Fri, 22 Jun 2018 00:48:06 +0900
Subject: [PATCH] GRPC::Cancelled should be occured when calling Enumrable#next
 on canceled call

---
 src/ruby/lib/grpc/generic/active_call.rb  | 24 +++++++++++++----------
 src/ruby/spec/generic/client_stub_spec.rb | 12 ++++++++++++
 2 files changed, 26 insertions(+), 10 deletions(-)

diff --git a/src/ruby/lib/grpc/generic/active_call.rb b/src/ruby/lib/grpc/generic/active_call.rb
index 6e601bdc917..688726ef4a8 100644
--- a/src/ruby/lib/grpc/generic/active_call.rb
+++ b/src/ruby/lib/grpc/generic/active_call.rb
@@ -322,18 +322,22 @@ module GRPC
     # @return [Enumerator] if no block was given
     def each_remote_read_then_finish
       return enum_for(:each_remote_read_then_finish) unless block_given?
-      begin
-        loop do
-          resp = remote_read
-          if resp.nil?  # the last response was received
-            receive_and_check_status
-            break
+      loop do
+        resp =
+          begin
+            remote_read
+          rescue GRPC::Core::CallError => e
+            GRPC.logger.warn("In each_remote_read_then_finish: #{e}")
+            nil
           end
-          yield resp
-        end
-      ensure
-        set_input_stream_done
+
+        break if resp.nil?  # the last response was received
+        yield resp
       end
+
+      receive_and_check_status
+    ensure
+      set_input_stream_done
     end
 
     # request_response sends a request to a GRPC server, and returns the
diff --git a/src/ruby/spec/generic/client_stub_spec.rb b/src/ruby/spec/generic/client_stub_spec.rb
index 3f878cc1006..fbf594a6553 100644
--- a/src/ruby/spec/generic/client_stub_spec.rb
+++ b/src/ruby/spec/generic/client_stub_spec.rb
@@ -589,6 +589,18 @@ describe 'ClientStub' do  # rubocop:disable Metrics/BlockLength
           responses.each { |r| p r }
         end
       end
+
+      it 'raises GRPC::Cancelled after the call has been cancelled' do
+        server_port = create_test_server
+        host = "localhost:#{server_port}"
+        th = run_server_streamer(@sent_msg, @replys, @pass)
+        stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
+        resp = get_responses(stub, run_start_call_first: false)
+        expect(resp.next).to eq('reply_1')
+        @op.cancel
+        expect { resp.next }.to raise_error(GRPC::Cancelled)
+        th.join
+      end
     end
   end