Merge pull request #3 from tejanium/infinite-timeout

Infinite timeout
cjab · Dec 3, 2018 · 347e48d · 347e48d
2 parents 66ad69d + 09480b0
commit 347e48d
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 7 deletions.
diff --git a/lib/weddell/client.ex b/lib/weddell/client.ex
@@ -125,15 +125,18 @@ defmodule Weddell.Client do
     host = Keyword.get(opts, :host, @default_host)
     port = Keyword.get(opts, :port, @default_port)
     {:ok, channel} =
-      Stub.connect("#{host}:#{port}", cred: cred)
+      Stub.connect("#{host}:#{port}", cred: cred, adapter_opts: %{
+        http2_opts: %{keepalive: :infinity}
+      })
     {:ok, %__MODULE__{channel: channel,
                       project: project}}
   end
 
   @doc false
   @spec request_opts() :: Keyword.t
-  def request_opts() do
+  def request_opts(extra_opts \\ []) do
     [metadata: auth_header(), content_type: "application/grpc"]
+    |> Enum.concat(extra_opts)
   end
 
   @doc false

diff --git a/lib/weddell/client/subscriber/stream.ex b/lib/weddell/client/subscriber/stream.ex
@@ -18,7 +18,7 @@ defmodule Weddell.Client.Subscriber.Stream do
   defstruct [:client, :subscription, :grpc_stream]
 
   @default_ack_deadline 10
-  @deadline_expired 4
+  @unavailable 14
 
   @doc """
   Open a new stream on a subscription.
@@ -37,7 +37,7 @@ defmodule Weddell.Client.Subscriber.Stream do
     stream =
       %__MODULE__{client: client,
                   subscription: subscription,
-                  grpc_stream: Stub.streaming_pull(client.channel, Client.request_opts())}
+                  grpc_stream: Stub.streaming_pull(client.channel, Client.request_opts(timeout: :infinity))}
     request =
       StreamingPullRequest.new(subscription: Util.full_subscription(client.project, subscription),
                                stream_ack_deadline_seconds: @default_ack_deadline)
@@ -134,14 +134,13 @@ defmodule Weddell.Client.Subscriber.Stream do
   """
   @spec recv(stream :: t) :: Enumerable.t
   def recv(stream) do
-    case GRPCStub.recv(stream.grpc_stream) do
+    case GRPCStub.recv(stream.grpc_stream, timeout: :infinity) do
       {:ok, recv} ->
         recv
         |> Stream.map(fn
           {:ok, response} ->
             Enum.map(response.received_messages, &Message.new/1)
-          {:error, %RPCError{status: @deadline_expired}} ->
-            # Deadline expired and stream ended, this is expected
+          {:error, %RPCError{status: @unavailable}} ->
             []
           {:error, e} ->
             raise e