Refactor Stream.reduce and create an helper
Fetch.Response.partition_messages function.
MODIFY
lib/kafka_ex/protocol/fetch.ex
lib/kafka_ex/protocol/fetch.ex
@@ -25,6 +25,12 @@ defmodule KafkaEx.Protocol.Fetch do
@moduledoc false
defstruct topic: nil, partitions: []
@type t :: %Response{topic: binary, partitions: list}
+
+ @spec partition_messages(list, binary, integer) :: list
+ def partition_messages(responses, topic, partition) do
+ response = Enum.find(responses, &(&1.topic == topic)) || %Response{}
+ Enum.find(response.partitions, &(&1.partition == partition))
+ end
end
defmodule Message do

MODIFY
lib/kafka_ex/stream.ex
lib/kafka_ex/stream.ex
@@ -2,8 +2,11 @@ defmodule KafkaEx.Stream do
@moduledoc false
alias KafkaEx.Protocol.OffsetCommit.Request, as: OffsetCommitRequest
alias KafkaEx.Protocol.Fetch.Request, as: FetchRequest
+ alias KafkaEx.Protocol.Fetch.Response, as: FetchResponse
- defstruct worker_name: nil, fetch_request: %FetchRequest{}, consumer_group: nil
+ defstruct worker_name: nil,
+ fetch_request: %FetchRequest{},
+ consumer_group: nil
defimpl Enumerable do
def reduce(data, acc, fun) do
@@ -19,9 +22,7 @@ defmodule KafkaEx.Stream do
}
})
end
- response = data.worker_name
- |> GenServer.call({:fetch, %{data.fetch_request| offset: offset}})
- |> hd |> Map.get(:partitions) |> hd
+ response = fetch_response(data, offset)
if response.error_code == :no_error &&
response.last_offset != nil && response.last_offset != offset do
{response.message_set, response.last_offset}
@@ -29,7 +30,16 @@ defmodule KafkaEx.Stream do
{:halt, offset}
end
end
- Stream.resource(fn -> data.fetch_request.offset end, next_fun, &(&1)).(acc, fun)
+ Stream.resource(
+ fn -> data.fetch_request.offset end, next_fun, &(&1)
+ ).(acc, fun)
+ end
+
+ defp fetch_response(data, offset) do
+ req = data.fetch_request
+ data.worker_name
+ |> GenServer.call({:fetch, %{req| offset: offset}})
+ |> FetchResponse.partition_messages(req.topic, req.partition)
end
def count(_stream) do

Refactor Stream.reduce and create an helper
Fetch.Response.partition_messages function.