Rename callbacks for consistency (kafka_* => kafka_server_*)
MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -206,19 +206,20 @@ defmodule KafkaEx.Server do
| {:stop, reason, reply, new_state}
| {:stop, reason, new_state}
when reply: term, new_state: term, reason: term
- @callback kafka_create_topics(
+ @callback kafka_server_create_topics(
[CreateTopicsRequest.t()],
network_timeout :: integer,
state :: State.t()
) :: {:reply, reply, new_state}
when reply: term, new_state: term
- @callback kafka_delete_topics(
+ @callback kafka_server_delete_topics(
[String.t()],
network_timeout :: integer,
state :: State.t()
) :: {:reply, reply, new_state}
when reply: term, new_state: term
- @callback kafka_api_versions(state :: State.t()) :: {:reply, reply, new_state}
+ @callback kafka_server_api_versions(state :: State.t()) ::
+ {:reply, reply, new_state}
when reply: term, new_state: term
@callback kafka_server_update_metadata(state :: State.t()) ::
{:noreply, new_state}
@@ -337,15 +338,15 @@ defmodule KafkaEx.Server do
end
def handle_call({:create_topics, requests, network_timeout}, _from, state) do
- kafka_create_topics(requests, network_timeout, state)
+ kafka_server_create_topics(requests, network_timeout, state)
end
def handle_call({:delete_topics, topics, network_timeout}, _from, state) do
- kafka_delete_topics(topics, network_timeout, state)
+ kafka_server_delete_topics(topics, network_timeout, state)
end
def handle_call({:api_versions}, _from, state) do
- kafka_api_versions(state)
+ kafka_server_api_versions(state)
end
def handle_info(:update_metadata, state) do

MODIFY
lib/kafka_ex/server_0_p_10_and_later.ex
lib/kafka_ex/server_0_p_10_and_later.ex
@@ -95,6 +95,7 @@ defmodule KafkaEx.Server0P10AndLater do
check_brokers_sockets!(brokers)
{_,
+
%KafkaEx.Protocol.ApiVersions.Response{
api_versions: api_versions,
error_code: error_code
@@ -181,7 +182,7 @@ defmodule KafkaEx.Server0P10AndLater do
{:noreply, update_metadata(state)}
end
- def kafka_api_versions(state) do
+ def kafka_server_api_versions(state) do
response =
state.correlation_id
|> ApiVersions.create_request(@client_id)
@@ -191,7 +192,7 @@ defmodule KafkaEx.Server0P10AndLater do
{:reply, response, %{state | correlation_id: state.correlation_id + 1}}
end
- def kafka_delete_topics(topics, network_timeout, state) do
+ def kafka_server_delete_topics(topics, network_timeout, state) do
api_version =
case DeleteTopics.api_version(state.api_versions) do
{:ok, api_version} ->
@@ -238,7 +239,7 @@ defmodule KafkaEx.Server0P10AndLater do
{:reply, response, state}
end
- def kafka_create_topics(requests, network_timeout, state) do
+ def kafka_server_create_topics(requests, network_timeout, state) do
api_version =
case CreateTopics.api_version(state.api_versions) do
{:ok, api_version} ->

MODIFY
lib/kafka_ex/server_0_p_8_p_0.ex
lib/kafka_ex/server_0_p_8_p_0.ex
@@ -14,9 +14,9 @@ defmodule KafkaEx.Server0P8P0 do
{:nowarn_function, kafka_server_consumer_group: 1},
{:nowarn_function, kafka_server_offset_commit: 2},
{:nowarn_function, kafka_server_offset_fetch: 2},
- {:nowarn_function, kafka_create_topics: 3},
- {:nowarn_function, kafka_delete_topics: 3},
- {:nowarn_function, kafka_api_versions: 1}
+ {:nowarn_function, kafka_server_create_topics: 3},
+ {:nowarn_function, kafka_server_delete_topics: 3},
+ {:nowarn_function, kafka_server_api_versions: 1}
]
use KafkaEx.Server
@@ -89,13 +89,13 @@ defmodule KafkaEx.Server0P8P0 do
"Consumer Group Metadata is not supported in 0.8.0 version of kafka"
)
- def kafka_api_versions(_state),
+ def kafka_server_api_versions(_state),
do: raise("ApiVersions is not supported in 0.8.0 version of kafka")
- def kafka_create_topics(_, _, _state),
+ def kafka_server_create_topics(_, _, _state),
do: raise("CreateTopic is not supported in 0.8.0 version of kafka")
- def kafka_delete_topics(_, _, _state),
+ def kafka_server_delete_topics(_, _, _state),
do: raise("DeleteTopic is not supported in 0.8.0 version of kafka")
defp fetch(request, state) do

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -9,9 +9,9 @@ defmodule KafkaEx.Server0P8P2 do
{:nowarn_function, kafka_server_sync_group: 3},
{:nowarn_function, kafka_server_join_group: 3},
{:nowarn_function, kafka_server_leave_group: 3},
- {:nowarn_function, kafka_create_topics: 3},
- {:nowarn_function, kafka_delete_topics: 3},
- {:nowarn_function, kafka_api_versions: 1}
+ {:nowarn_function, kafka_server_create_topics: 3},
+ {:nowarn_function, kafka_server_delete_topics: 3},
+ {:nowarn_function, kafka_server_api_versions: 1}
]
use KafkaEx.Server
@@ -200,13 +200,13 @@ defmodule KafkaEx.Server0P8P2 do
def kafka_server_heartbeat(_, _, _state),
do: raise("Heartbeat is not supported in 0.8.2 version of kafka")
- def kafka_api_versions(_state),
+ def kafka_server_api_versions(_state),
do: raise("ApiVersions is not supported in 0.8.2 version of kafka")
- def kafka_create_topics(_, _, _state),
+ def kafka_server_create_topics(_, _, _state),
do: raise("CreateTopic is not supported in 0.8.2 version of kafka")
- def kafka_delete_topics(_, _, _state),
+ def kafka_server_delete_topics(_, _, _state),
do: raise("DeleteTopic is not supported in 0.8.2 version of kafka")
defp update_consumer_metadata(state),

MODIFY
lib/kafka_ex/server_0_p_9_p_0.ex
lib/kafka_ex/server_0_p_9_p_0.ex
@@ -6,9 +6,9 @@ defmodule KafkaEx.Server0P9P0 do
# these functions aren't implemented for 0.9.0
@dialyzer [
- {:nowarn_function, kafka_create_topics: 3},
- {:nowarn_function, kafka_delete_topics: 3},
- {:nowarn_function, kafka_api_versions: 1}
+ {:nowarn_function, kafka_server_create_topics: 3},
+ {:nowarn_function, kafka_server_delete_topics: 3},
+ {:nowarn_function, kafka_server_api_versions: 1}
]
use KafkaEx.Server
@@ -52,13 +52,13 @@ defmodule KafkaEx.Server0P9P0 do
defdelegate kafka_server_consumer_group_metadata(state), to: Server0P8P2
defdelegate kafka_server_update_consumer_metadata(state), to: Server0P8P2
- def kafka_api_versions(_state),
+ def kafka_server_api_versions(_state),
do: raise("ApiVersions is not supported in 0.9.0 version of kafka")
- def kafka_create_topics(_, _, _state),
+ def kafka_server_create_topics(_, _, _state),
do: raise("CreateTopic is not supported in 0.9.0 version of kafka")
- def kafka_delete_topics(_, _, _state),
+ def kafka_server_delete_topics(_, _, _state),
do: raise("DeleteTopic is not supported in 0.9.0 version of kafka")
def kafka_server_init([args]) do

Rename callbacks for consistency (kafka_* => kafka_server_*)