Hey everyone,
I want to receive data from a gRPC stream.
So far, I only wrote functions, played around in iex and was able to receive and decode data. It is time to “applicationify” that previous work and add lifecycle management/processes. But somehow, the gRPC behaviour seems different in GenServer compared to the iex session? 
I found a similar post in the forum: Elixir-grpc can't receive stream response. Unexpected when waiting for headers- which ends on
Looks like GRPC.Stub.recv can’t be used in GenServers
But that seems weird to me.
Setup in iex
Basically, the stripped down version looks something like this:
# wrapper around GRPC.Stub.connect() that handles auth etc
{:ok, channel} = SalesforceEvents.Grpc.connect()
grpc_stream = Eventbus.V1.PubSub.Stub.subscribe(channel)
GRPC.Stub.send_request(grpc_stream, Messages.request_data())
{:ok, data_stream} = GRPC.Stub.recv(grpc_stream, timeout: 60_000)
Enum.each(data_stream, &IO.inspect/1)
# shell blocks, protobuf-decoded data is printed as it comes in
Setup with GenServer
Same as above, but in a GenServer.
Handling incoming stream data with Enum.each is spawned as a separate process:
@impl true
def init(args) do
{:ok, channel} = SalesforceEvents.Grpc.connect()
grpc_stream = Eventbus.V1.PubSub.Stub.subscribe(channel)
GRPC.Stub.send_request(grpc_stream, Messages.request_data())
{:ok, data_stream} = GRPC.Stub.recv(grpc_stream, timeout: 60_000)
receiver_process =
spawn_link(fn -> Enum.each(data_stream, &IO.inspect/1) end)
state = %{
channel: channel,
grpc_stream: grpc_stream,
data_stream: data_stream,
receiver_process: receiver_process
}
{:ok, state}
end
def handle_info(info, state) do
Logger.debug(inspect info)
{:noreply, state}
end
Expected Behaviour
Received gRPC messages are still printed to the console
Actual Behaviour
Instead of the messages being printed, the handle_info receives messages like the following:
08:45:41.824 [debug] {:gun_data, #PID<0.339.0>, #Reference<0.2404288951.1709178882.257606>, :nofin, <<0, 0, 0, 1, 198, 10, 145, 3, …>>}
Why does this behave differently (nothing is printed, messages are not decoded)?
How can I receive data from a gRPC stream within a process getting results similar to what I had in iex?
