Merge pull request #25 from cipherstash/put-stream-bi-direction

Update put stream to handle bi directional stream

Author: fimac

cipherstash-client.gemspec

@@ -32,7 +32,7 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency "aws-sdk-core", "~> 3.0"
   s.add_runtime_dependency "aws-sdk-kms", "~> 1.0"
   s.add_runtime_dependency 'cbor', '~> 0.5.9.6'
-  s.add_runtime_dependency "cipherstash-grpc", "= 0.20220801.1"
+  s.add_runtime_dependency "cipherstash-grpc", "= 0.20220928.0"
   s.add_runtime_dependency "enveloperb", "~> 0.0"
   s.add_runtime_dependency "launchy", "~> 2.5"
   s.add_runtime_dependency "ore-rs", "~> 0.0"

lib/cipherstash/client/rpc.rb

@@ -270,17 +270,26 @@ def put_stream(collection, records)
         # isn't the most idiomatic code on earth...
         requests = [[begin_request], records].lazy.flat_map { |x| x }
 
-        res = @metrics.measure_rpc_call("putStream") do
-          stub.put_stream(requests, metadata: rpc_headers)
-        end
+        num_inserted = 0
+        @logger.debug("Cipherstash::Client::RPC#put_stream") { "Start streaming upsert of documents..." }
+
+        @metrics.measure_rpc_call("putStream") do
+          stub.put_stream_bi_direction(requests, metadata: rpc_headers).each do |res|
+            unless res.is_a?(Documents::StreamingPutReply)
+              raise Error::StreamingPutFailure, "expected Documents::StreamingPutReply response, got #{res.class} instead"
+            end
+
+            raise_if_error(res)
 
-        unless res.is_a?(Documents::StreamingPutReply)
-          raise Error::StreamingPutFailure, "expected Documents::StreamingPutReply response, got #{res.class} instead"
+            @logger.debug("Cipherstash::Client::RPC#put_stream") { "#{res.numInserted} records inserted." }
+
+            num_inserted = res.numInserted
+          end
         end
 
-        raise_if_error(res)
+        @logger.debug("Cipherstash::Client::RPC#put_stream") { "Streaming upsert complete. Number of records inserted: #{res.numInserted}" }
 
-        res.numInserted
+        num_inserted
       rescue ::GRPC::NotFound
         raise Error::RecordPutFailure, "Collection '#{collection.name}' not found"
       rescue ::GRPC::InvalidArgument => ex