Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[#397]: Preserve the order of the response messages when the stream response process handles ":consume_response" #396

Merged
merged 7 commits into from
Jan 9, 2025
Merged

[#397]: Preserve the order of the response messages when the stream response process handles ":consume_response" #396

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
eb4050c
Simple bug demonstration and bugfix
senconscious Jan 9, 2025
5d061c0
Migrate responses from list to erlang queue
senconscious Jan 9, 2025
75d0a43
Use queue.to_list instead of queue.peek in tests
senconscious Jan 9, 2025
55c01a8
Use factory to build message for ordering test
senconscious Jan 9, 2025
48adaea
Fix bye luis test and encoding
senconscious Jan 9, 2025
cf81fc2
Fix mint adapter connection process test after responses migration to…
senconscious Jan 9, 2025
7ff4f0a
Update test/grpc/client/adapters/mint/connection_process_test.exs
polvalente Jan 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions lib/grpc/client/adapters/mint/stream_response_process.ex
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcess do
{{_, message}, rest} ->
# TODO add code here to handle compressor headers
response = codec.decode(message, res_mod)
new_responses = [{:ok, response} | responses]
new_responses = responses ++ [{:ok, response}]
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of using a list, let's change the data structure to a queue: https://www.erlang.org/doc/apps/stdlib/queue.html

Erlang :queue is supposed to be O(1) for enqueueing and dequeueing. This change is O(len(responses)) for enqueueing

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, migrated to erlang queue

new_state = %{state | buffer: rest, responses: new_responses}
{:reply, :ok, new_state, {:continue, :produce_response}}

Expand All @@ -117,7 +117,7 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcess do
)
when type in @header_types do
state = update_compressor({type, headers}, state)
new_responses = [get_headers_response(headers, type) | responses]
new_responses = responses ++ [get_headers_response(headers, type)]
{:reply, :ok, %{state | responses: new_responses}, {:continue, :produce_response}}
end

Expand All @@ -131,7 +131,7 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcess do

case get_headers_response(headers, type) do
{:error, _rpc_error} = error ->
{:reply, :ok, %{state | responses: [error | responses]}, {:continue, :produce_response}}
{:reply, :ok, %{state | responses: responses ++ [error]}, {:continue, :produce_response}}

_any ->
{:reply, :ok, state, {:continue, :produce_response}}
Expand All @@ -143,7 +143,7 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcess do
_from,
%{responses: responses} = state
) do
{:reply, :ok, %{state | responses: [error | responses]}, {:continue, :produce_response}}
{:reply, :ok, %{state | responses: responses ++ [error]}, {:continue, :produce_response}}
end

def handle_call({:consume_response, :done}, _from, state) do
Expand Down
21 changes: 21 additions & 0 deletions test/grpc/client/adapters/mint/stream_response_process_test.exs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,27 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcessTest do
assert {:ok, build(:hello_reply_rpc)} == data
end

test "ordering bug", %{pid: pid} do
hello_luis =
<<0, 0, 0, 0, 12, 10, 10, 72, 101, 108, 108, 111, 32, 76, 117, 105, 115>>

hello_luit =
<<0, 0, 0, 0, 12, 10, 10, 72, 101, 108, 108, 111, 32, 76, 117, 105, 116>>

stream = StreamResponseProcess.build_stream(pid)
StreamResponseProcess.consume(pid, :data, hello_luis)
StreamResponseProcess.consume(pid, :data, hello_luit)
StreamResponseProcess.done(pid)

expected_elements =
[
ok: %Helloworld.HelloReply{message: "Hello Luis", __unknown_fields__: []},
ok: %Helloworld.HelloReply{message: "Hello Luit", __unknown_fields__: []}
]

assert Enum.to_list(stream) == expected_elements
end

test_with_params(
"emits headers to stream",
%{pid: pid},
Expand Down