Replace the current GenEvent based stream implementation with a really
lazy stream implementation, this closes
MODIFY
lib/kafka_ex.ex
lib/kafka_ex.ex
@@ -262,8 +262,6 @@ Optional arguments(KeywordList)
Optional arguments(KeywordList)
- worker_name: the worker we want to run this metadata request through, when none is provided the default worker `:kafka_ex` is used
- offset: When supplied the fetch would start from this offset, otherwise would start from the last committed offset of the consumer_group the worker belongs to. For Kafka < 0.8.2 you should explicitly specify this.
- - handler: the handler we want to handle the streaming events, when none is provided the default KafkaEx.Handler is used
- - handler_init: initial state for the handler - leave the default value [] when using the default handler
- auto_commit: specifies if the last offset should be commited or not. Default is true. You must set this to false when using Kafka < 0.8.2 or `:no_consumer_group`.
@@ -283,32 +281,21 @@ Optional arguments(KeywordList)
def stream(topic, partition, opts \\ []) do
worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
supplied_offset = Keyword.get(opts, :offset)
- handler = Keyword.get(opts, :handler, KafkaEx.Handler)
- handler_init = Keyword.get(opts, :handler_init, [])
auto_commit = Keyword.get(opts, :auto_commit, true)
wait_time = Keyword.get(opts, :wait_time, @wait_time)
min_bytes = Keyword.get(opts, :min_bytes, @min_bytes)
max_bytes = Keyword.get(opts, :max_bytes, @max_bytes)
-
- event_stream = Server.call(worker_name, {:create_stream, handler, handler_init})
+ consumer_group = Keyword.get(opts, :consumer_group)
retrieved_offset = current_offset(supplied_offset, partition, topic, worker_name)
- send(worker_name, {
- :start_streaming,
- %FetchRequest{
+ fetch_request = %FetchRequest{
auto_commit: auto_commit,
topic: topic, partition: partition,
offset: retrieved_offset, wait_time: wait_time,
min_bytes: min_bytes, max_bytes: max_bytes
}
- })
- event_stream
- end
- @spec stop_streaming(Keyword.t) :: :stop_streaming
- def stop_streaming(opts \\ []) do
- worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
- send(worker_name, :stop_streaming)
+ %KafkaEx.Stream{worker_name: worker_name, fetch_request: fetch_request, consumer_group: consumer_group}
end
defp build_worker_options(worker_init) do

MODIFY
lib/kafka_ex/stream.ex
lib/kafka_ex/stream.ex
@@ -9,13 +9,23 @@ defmodule KafkaEx.Stream do
def reduce(data, acc, fun) do
next_fun = fn offset ->
if data.fetch_request.auto_commit do
- GenServer.call(data.worker_name, {:offset_commit, %OffsetCommitRequest{consumer_group: data.consumer_group, topic: data.fetch_request.topic, partition: data.fetch_request.partition, offset: offset, metadata: ""}})
+ GenServer.call(data.worker_name, {
+ :offset_commit,
+ %OffsetCommitRequest{
+ consumer_group: data.consumer_group,
+ topic: data.fetch_request.topic,
+ partition: data.fetch_request.partition,
+ offset: offset, metadata: ""
+ }
+ })
end
- response = GenServer.call(data.worker_name, {:fetch, %{data.fetch_request| offset: offset}})
- last_offset = response |> hd |> Map.get(:partitions) |> hd |> Map.get(:last_offset)
- {response, last_offset}
+ response = data.worker_name
+ |> GenServer.call({:fetch, %{data.fetch_request| offset: offset}})
+ |> hd |> Map.get(:partitions) |> hd
+ last_offset = response |> Map.get(:last_offset)
+ {response.message_set, last_offset}
end
- Stream.resource(fn -> data.fetch_request.offset end, next_fun, fn offset -> offset end).(acc, fun)
+ Stream.resource(&(data.fetch_request.offset), next_fun, &(&1)).(acc, fun)
end
def count(_stream) do
{:error, __MODULE__}

MODIFY
test/integration/consumer_group_test.exs
test/integration/consumer_group_test.exs
@@ -211,7 +211,7 @@ defmodule KafkaEx.ConsumerGroup.Test do
})
stream = KafkaEx.stream(random_string, 0, worker_name: :stream_auto_commit, offset: 0)
log = TestHelper.wait_for_any(
- fn() -> GenEvent.call(stream.manager, KafkaEx.Handler, :messages) |> Enum.take(2) end
+ fn() -> stream |> Enum.take(2) end
)
refute Enum.empty?(log)
@@ -240,13 +240,11 @@ defmodule KafkaEx.ConsumerGroup.Test do
stream = KafkaEx.stream(random_string, 0, worker_name: worker_name)
log = TestHelper.wait_for_any(
- fn() -> GenEvent.call(stream.manager, KafkaEx.Handler, :messages) |> Enum.take(2) end
+ fn() -> stream |> Enum.take(2) end
)
refute Enum.empty?(log)
-
first_message = log |> hd
-
assert first_message.offset == 4
end
@@ -258,7 +256,7 @@ defmodule KafkaEx.ConsumerGroup.Test do
# make sure we consume at least one message before we assert that there is no offset committed
_log = TestHelper.wait_for_any(
- fn() -> GenEvent.call(stream.manager, KafkaEx.Handler, :messages) |> Enum.take(2) end
+ fn() -> stream |> Enum.take(2) end
)
offset_fetch_response = KafkaEx.offset_fetch(:stream_no_auto_commit, %Proto.OffsetFetch.Request{topic: random_string, partition: 0}) |> hd

MODIFY
test/integration/integration_test.exs
test/integration/integration_test.exs
@@ -339,7 +339,7 @@ defmodule KafkaEx.Integration.Test do
stream = KafkaEx.stream(random_string, 0, worker_name: :stream, offset: 0, auto_commit: false)
log = TestHelper.wait_for_accum(
- fn() -> GenEvent.call(stream.manager, KafkaEx.Handler, :messages) end,
+ fn() -> stream |> Enum.take(2) end,
2
)
@@ -354,85 +354,4 @@ defmodule KafkaEx.Integration.Test do
KafkaEx.stream(random_string, 0, offset: 0)
KafkaEx.stream(random_string, 0, offset: 0)
end
-
- test "stop_streaming stops streaming, and stream starts it up again" do
- random_string = generate_random_string()
- KafkaEx.create_worker(:stream2, uris: uris())
- stream = KafkaEx.stream(random_string, 0, worker_name: :stream2, offset: 0, auto_commit: false)
-
- KafkaEx.create_worker(:producer, uris: uris())
- KafkaEx.produce(%Proto.Produce.Request{topic: random_string, partition: 0, required_acks: 1, messages: [
- %Proto.Produce.Message{value: "one"},
- %Proto.Produce.Message{value: "two"},
- ]
- }, worker_name: :producer)
-
- log = TestHelper.wait_for_accum(
- fn() -> GenEvent.call(stream.manager, KafkaEx.Handler, :messages) end,
- 2
- )
-
- last_offset = hd(Enum.reverse(log)).offset
-
- KafkaEx.stop_streaming(worker_name: :stream2)
- :ok = TestHelper.wait_for(fn() -> !Process.alive?(stream.manager) end)
-
- KafkaEx.produce(%Proto.Produce.Request{topic: random_string, partition: 0, required_acks: 1, messages: [
- %Proto.Produce.Message{value: "three"},
- %Proto.Produce.Message{value: "four"},
- ]
- }, worker_name: :producer)
-
- stream = KafkaEx.stream(random_string, 0, worker_name: :stream2, offset: last_offset+1, auto_commit: false)
-
- :ok = TestHelper.wait_for(fn() -> Process.alive?(stream.manager) end)
-
- log = GenEvent.call(stream.manager, KafkaEx.Handler, :messages)
- assert length(log) == 0
-
- KafkaEx.produce(%Proto.Produce.Request{topic: random_string, partition: 0, required_acks: 1, messages: [
- %Proto.Produce.Message{value: "five"},
- %Proto.Produce.Message{value: "six"},
- ]
- }, worker_name: :producer)
-
- log = TestHelper.wait_for_accum(
- fn() -> GenEvent.call(stream.manager, KafkaEx.Handler, :messages) end,
- 4
- )
-
- assert length(log) == 4
- end
-
- test "streams kafka logs with custom handler and initial state" do
- random_string = generate_random_string()
- KafkaEx.create_worker(:stream3, uris: uris())
- {:ok, offset} = KafkaEx.produce(%Proto.Produce.Request{topic: random_string, partition: 0, required_acks: 1, messages: [
- %Proto.Produce.Message{value: "hey"},
- %Proto.Produce.Message{value: "hi"},
- ]
- }, worker_name: :stream3)
-
- defmodule CustomHandlerSendingMessage do
- use GenEvent
-
- def init(pid) do
- {:ok, pid}
- end
-
- def handle_event(message, pid) do
- send(pid, message)
- {:ok, pid}
- end
- end
- KafkaEx.stream(random_string, 0, worker_name: :stream3, offset: offset, auto_commit: false, handler: CustomHandlerSendingMessage, handler_init: self())
-
- assert_receive %KafkaEx.Protocol.Fetch.Message{key: nil, value: "hey", offset: ^offset, attributes: 0, crc: 4264455069}
- offset = offset + 1
- assert_receive %KafkaEx.Protocol.Fetch.Message{key: nil, value: "hi", offset: ^offset, attributes: 0, crc: 4251893211}
- receive do
- _ -> assert(false, "Should not have received a third message")
- after 100 -> :ok
- end
- end
end

Replace the current GenEvent based stream implementation with a really
lazy stream implementation, this closes #145.