refactor: move version range functions to kpro_api_vsn module

Author: zmstone

src/kpro_api_vsn.erl

@@ -15,7 +15,7 @@
 
 %% Supported versions of THIS lib
 -module(kpro_api_vsn).
--export([range/1, kafka_09_range/1, intersect/2]).
+-export([range/1, kafka_09_range/1, intersect/1, intersect/2]).
 
 -export_type([range/0]).
 
@@ -54,19 +54,76 @@ kafka_09_range(describe_groups) -> {0, 0};
 kafka_09_range(list_groups) -> {0, 0};
 kafka_09_range(_) -> false.
 
-%% @doc Returns the intersection of two version ranges.
+%% @private Returns the intersection of two version ranges.
 %% An error is raised if there is no intersection.
--spec intersect(false | range(), false | range()) -> false | range().
-intersect(false, _) -> false;
-intersect(_, false) -> false;
-intersect({Min1, Max1} = R1, {Min2, Max2} = R2) ->
-  Min = max(Min1, Min2),
-  Max = min(Max1, Max2),
+-spec intersect(atom(), false | range(), false | range()) -> false | range().
+intersect(_API, false, _) -> false;
+intersect(_API, _, false) -> false;
+intersect(API, {Min0, Max0} = Supported, {Min1, Max1} = Received) ->
+  {Min2, Max2} = fix_range(API, Min1, Max1),
+  Min = max(Min0, Min2),
+  Max = min(Max0, Max2),
   case Min > Max of
-    true -> erlang:error({no_intersection, R1, R2});
+    true -> erlang:error({no_intersection, Supported, Received});
     false -> {Min, Max}
   end.
 
+%% Special adjustment for received API range.
+%% - produce: Minimal version is in fact 3, but Kafka may respond 0.
+%% - fetch: Minimal version is in fact 4, but Kafka may respond 0.
+fix_range(produce, Min, Max) ->
+    case Max >= 8 of
+        true ->
+            {max(Min, 3), Max};
+        false ->
+            {Min, Max}
+    end;
+fix_range(fetch, Min, Max) ->
+    case Max >= 11 of
+        true ->
+            {max(Min, 4), Max};
+        false ->
+            {Min, Max}
+    end;
+fix_range(_API, Min, Max) ->
+    {Min, Max}.
+
+%% @doc Return the intersection of supported version ranges and received version ranges.
+-spec intersect(undefined | list()) -> #{atom() => range()}.
+intersect(undefined) ->
+  %% kpro_connection is configured not to query api versions (kafka-0.9)
+  %% always use minimum supported version in this case
+  lists:foldl(
+    fun(API, Acc) ->
+      case kpro_api_vsn:kafka_09_range(API) of
+        false -> Acc;
+        {Min, _Max} -> Acc#{API => {Min, Min}}
+      end
+    end, #{}, kpro_schema:all_apis());
+intersect(Vsns) ->
+  maps:fold(
+    fun(API, {Min, Max}, Acc) ->
+      case intersect(API, {Min, Max}) of
+        false -> Acc;
+        Intersection -> Acc#{API => Intersection}
+      end
+    end, #{}, Vsns).
+
+%% @doc Intersect received api version range with supported range.
+-spec intersect(kpro:api(), range()) -> range().
+intersect(API, Received) ->
+  Supported = kpro_api_vsn:range(API),
+  try
+    intersect(API, Supported, Received)
+  catch
+    error : {no_intersection, _, _} ->
+      Reason = #{reason => incompatible_version_ranges,
+                 supported => Supported,
+                 received => Received,
+                 api => API},
+      erlang:error(Reason)
+  end.
+
 %%%_* Emacs ====================================================================
 %%% Local Variables:
 %%% allout-layout: t

src/kpro_brokers.erl

@@ -116,7 +116,7 @@ connect_controller(Bootstrap, Config, Opts) ->
         {ok, kpro:vsn_ranges()} | {error, any()}.
 get_api_versions(Connection) ->
   case kpro_connection:get_api_vsns(Connection) of
-    {ok, Vsns}      -> {ok, api_vsn_range_intersection(Vsns)};
+    {ok, Vsns}      -> {ok, kpro_api_vsn:intersect(Vsns)};
     {error, Reason} -> {error, Reason}
   end.
 
@@ -269,39 +269,6 @@ discover_and_connect(DiscoverFun, Bootstrap, Config, Timeout) ->
     ],
   kpro_lib:ok_pipe(FL, Timeout).
 
-api_vsn_range_intersection(undefined) ->
-  %% kpro_connection is configured not to query api versions (kafka-0.9)
-  %% always use minimum supported version in this case
-  lists:foldl(
-    fun(API, Acc) ->
-        case kpro_api_vsn:kafka_09_range(API) of
-          false -> Acc;
-          {Min, _Max} -> Acc#{API => {Min, Min}}
-        end
-    end, #{}, kpro_schema:all_apis());
-api_vsn_range_intersection(Vsns) ->
-  maps:fold(
-    fun(API, {Min, Max}, Acc) ->
-        case api_vsn_range_intersection(API, {Min, Max}) of
-          false -> Acc;
-          Intersection -> Acc#{API => Intersection}
-        end
-    end, #{}, Vsns).
-
-%% Intersect received api version range with supported range.
-api_vsn_range_intersection(API, Received) ->
-  Expected = kpro_api_vsn:range(API),
-  try
-    kpro_api_vsn:intersect(Expected, Received)
-  catch
-    error : {no_intersection, _, _} ->
-      Reason = #{reason => incompatible_version_ranges,
-                 expected => Expected,
-                 received => Received,
-                 api => API},
-      erlang:error(Reason)
-  end.
-
 connect_any([], _Config, Errors) ->
   {error, lists:reverse(Errors)};
 connect_any([{Host, Port} | Rest], Config, Errors) ->
@@ -332,9 +299,9 @@ api_vsn_range_intersection_test() ->
     Received = {0, 0},
     ?assertError(#{api := API,
                    reason := incompatible_version_ranges,
-                   expected := _,
+                   supported := _,
                    received := Received},
-                 api_vsn_range_intersection(API, Received)).
+                 kpro_api_vsn:intersect(API, Received)).
 
 -endif.
 %%%_* Emacs ====================================================================

src/kpro_connection.erl

@@ -282,7 +282,7 @@ query_api_versions(Sock, Mod, ClientId, Deadline) ->
           case is_atom(API) of
             true ->
               %% known API for client
-              Acc#{API => adjust_vsn(API, MinVsn, MaxVsn)};
+              Acc#{API => {MinVsn, MaxVsn}};
             false ->
               %% a broker-only (ClusterAction) API
               Acc
@@ -293,26 +293,6 @@ query_api_versions(Sock, Mod, ClientId, Deadline) ->
       erlang:error({failed_to_query_api_versions, ErrorCode})
   end.
 
-%% Special adjustment for API rage.
-%% - produce: Minimal version is in fact 3, but Kafka may respond 0.
-%% - fetch: Minimal version is in fact 4, but Kafka may respond 0.
-adjust_vsn(produce, Min, Max) ->
-    case Max >= 8 of
-        true ->
-            {max(Min, 3), Max};
-        false ->
-            {Min, Max}
-    end;
-adjust_vsn(fetch, Min, Max) ->
-    case Max >= 11 of
-        true ->
-            {max(Min, 4), Max};
-        false ->
-            {Min, Max}
-    end;
-adjust_vsn(_API, Min, Max) ->
-    {Min, Max}.
-
 get_tcp_mod(_SslOpts = true)  -> ssl;
 get_tcp_mod(_SslOpts = [_|_]) -> ssl;
 get_tcp_mod(_)                -> gen_tcp.