Corrects the cancel_after_first_response behaviour
- introduces a #wait method on the call operation view
- invokes #wait on a Notifier that is created for all operations
- ensures the Notifier is invoked if necessary whenever a client request
completes
- updates the interop_test to use op.wait before checking if the call
was cancelled.
diff --git a/src/ruby/.rubocop_todo.yml b/src/ruby/.rubocop_todo.yml
index c35e970..05db404 100644
--- a/src/ruby/.rubocop_todo.yml
+++ b/src/ruby/.rubocop_todo.yml
@@ -12,7 +12,7 @@
# Offense count: 3
# Configuration parameters: CountComments.
Metrics/ClassLength:
- Max: 192
+ Max: 200
# Offense count: 35
# Configuration parameters: CountComments.
diff --git a/src/ruby/bin/interop/interop_client.rb b/src/ruby/bin/interop/interop_client.rb
index 16fb1b1..da4caa8 100755
--- a/src/ruby/bin/interop/interop_client.rb
+++ b/src/ruby/bin/interop/interop_client.rb
@@ -284,7 +284,8 @@
op = @stub.full_duplex_call(ppp.each_item, return_op: true)
ppp.canceller_op = op # causes ppp to cancel after the 1st message
op.execute.each { |r| ppp.queue.push(r) }
- assert(op.cancelled, 'call operation should be CANCELLED')
+ op.wait
+ assert(op.cancelled, 'call operation was not CANCELLED')
p 'OK: cancel_after_first_response'
end
diff --git a/src/ruby/lib/grpc/generic/active_call.rb b/src/ruby/lib/grpc/generic/active_call.rb
index 3814ef3..215c006 100644
--- a/src/ruby/lib/grpc/generic/active_call.rb
+++ b/src/ruby/lib/grpc/generic/active_call.rb
@@ -120,6 +120,7 @@
@started = started
@unmarshal = unmarshal
@metadata_tag = metadata_tag
+ @op_notifier = nil
end
# output_metadata are provides access to hash that can be used to
@@ -148,6 +149,7 @@
# operation provides a restricted view of this ActiveCall for use as
# a Operation.
def operation
+ @op_notifier = Notifier.new
Operation.new(self)
end
@@ -167,6 +169,7 @@
batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
return unless assert_finished
@call.status = batch_result.status
+ op_is_done
batch_result.check_status
end
@@ -184,6 +187,7 @@
end
end
@call.status = batch_result.status
+ op_is_done
batch_result.check_status
end
@@ -415,7 +419,7 @@
def bidi_streamer(requests, **kw, &blk)
start_call(**kw) unless @started
bd = BidiCall.new(@call, @cq, @marshal, @unmarshal, @deadline)
- bd.run_on_client(requests, &blk)
+ bd.run_on_client(requests, @op_notifier, &blk)
end
# run_server_bidi orchestrates a BiDi stream processing on a server.
@@ -434,6 +438,19 @@
bd.run_on_server(gen_each_reply)
end
+ # Waits till an operation completes
+ def wait
+ return if @op_notifier.nil?
+ GRPC.logger.debug("active_call.wait: on #{@op_notifier}")
+ @op_notifier.wait
+ end
+
+ # Signals that an operation is done
+ def op_is_done
+ return if @op_notifier.nil?
+ @op_notifier.notify(self)
+ end
+
private
# Starts the call if not already started
@@ -468,6 +485,6 @@
# Operation limits access to an ActiveCall's methods for use as
# a Operation on the client.
Operation = view_class(:cancel, :cancelled, :deadline, :execute,
- :metadata, :status, :start_call)
+ :metadata, :status, :start_call, :wait)
end
end
diff --git a/src/ruby/lib/grpc/generic/bidi_call.rb b/src/ruby/lib/grpc/generic/bidi_call.rb
index 489dd51..3b0c713 100644
--- a/src/ruby/lib/grpc/generic/bidi_call.rb
+++ b/src/ruby/lib/grpc/generic/bidi_call.rb
@@ -66,6 +66,7 @@
@cq = q
@deadline = deadline
@marshal = marshal
+ @op_notifier = nil # signals completion on clients
@readq = Queue.new
@unmarshal = unmarshal
end
@@ -76,8 +77,10 @@
# block that can be invoked with each response.
#
# @param requests the Enumerable of requests to send
+ # @op_notifier a Notifier used to signal completion
# @return an Enumerator of requests to yield
- def run_on_client(requests, &blk)
+ def run_on_client(requests, op_notifier, &blk)
+ @op_notifier = op_notifier
@enq_th = Thread.new { write_loop(requests) }
@loop_th = start_read_loop
each_queued_msg(&blk)
@@ -105,6 +108,13 @@
END_OF_READS = :end_of_reads
END_OF_WRITES = :end_of_writes
+ # signals that bidi operation is complete
+ def notify_done
+ return unless @op_notifier
+ GRPC.logger.debug("bidi-notify-done: notifying #{@op_notifier}")
+ @op_notifier.notify(self)
+ end
+
# each_queued_msg yields each message on this instances readq
#
# - messages are added to the readq by #read_loop
@@ -143,11 +153,13 @@
@call.status = batch_result.status
batch_result.check_status
GRPC.logger.debug("bidi-write-loop: done status #{@call.status}")
+ notify_done
end
GRPC.logger.debug('bidi-write-loop: finished')
rescue StandardError => e
GRPC.logger.warn('bidi-write-loop: failed')
GRPC.logger.warn(e)
+ notify_done
raise e
end
diff --git a/src/ruby/lib/grpc/logconfig.rb b/src/ruby/lib/grpc/logconfig.rb
index 9681217..e9b4aa3 100644
--- a/src/ruby/lib/grpc/logconfig.rb
+++ b/src/ruby/lib/grpc/logconfig.rb
@@ -38,6 +38,6 @@
Logging.logger.root.level = :info
# TODO: provide command-line configuration for logging
-Logging.logger['GRPC'].level = :debug
+Logging.logger['GRPC'].level = :info
Logging.logger['GRPC::ActiveCall'].level = :info
Logging.logger['GRPC::BidiCall'].level = :info