erlkaf

-module(erlkaf_utils).

-export([
    get_priv_path/1,
    get_local_queue_path/1,
    get_env/1,
    get_env/2,
    lookup/2,
    lookup/3,
    append_props/2,
    to_binary/1,
    safe_call/2,
    safe_call/3,
    call_stats_callback/3,
    call_oauthbearer_token_refresh_callback/2,
    parralel_exec/2
]).

-spec get_local_queue_path(string() | atom()) -> string() | undefined.
get_local_queue_path(File) ->
    case application:get_env(erlkaf, global_client_options) of
        undefined -> 
            get_priv_path(File);
        {ok, Val} -> 
            case lookup(local_queue_path, Val) of
                undefined -> 
                    get_priv_path(File);
                Path ->
                    filename:join(Path, File)
            end
    end.
    
get_priv_path(File) ->
    case code:priv_dir(erlkaf) of
        {error, bad_name} ->
            Ebin = filename:dirname(code:which(?MODULE)),
            filename:join([filename:dirname(Ebin), "priv", File]);
        Dir ->
            filename:join(Dir, File)
    end.

get_env(Key) ->
    get_env(Key, undefined).

get_env(Key, Default) ->
    case application:get_env(erlkaf, Key) of
        undefined ->
            Default;
        {ok, Val} ->
            Val
    end.

lookup(Key, List) ->
    lookup(Key, List, undefined).

lookup(Key, List, Default) ->
    case lists:keyfind(Key, 1, List) of
        {Key, Result}
            -> Result;
        false ->
            Default
    end.

append_props(L1, [{K, _} = H|T]) ->
    case lookup(K, L1) of
        undefined ->
            append_props([H|L1], T);
        _ ->
            append_props(L1, T)
    end;
append_props(L1, []) ->
    L1.

to_binary(V) when is_binary(V) ->
    V;
to_binary(V) when is_list(V) ->
    list_to_binary(V);
to_binary(V) when is_atom(V) ->
    atom_to_binary(V, utf8);
to_binary(V) when is_integer(V) ->
    integer_to_binary(V);
to_binary(V) when is_float(V) ->
    float_to_bin(V).

float_to_bin(Value) ->
    float_to_binary(Value, [{decimals, 8}, compact]).

safe_call(Receiver, Message) ->
    safe_call(Receiver, Message, 5000).

safe_call(Receiver, Message, Timeout) ->
    try
        gen_server:call(Receiver, Message, Timeout)
    catch
        exit:{noproc, _} ->
            {error, not_started};
        _: Exception ->
            {error, Exception}
    end.

call_stats_callback(undefined, _ClientId, _Stats) ->
    ok;
call_stats_callback(C, OAuthBearerConfig, Stats) when is_function(C, 2) ->
    C(OAuthBearerConfig, Stats);
call_stats_callback(C, OAuthBearerConfig, Stats) ->
    C:stats_callback(OAuthBearerConfig, Stats).

call_oauthbearer_token_refresh_callback(undefined, _OauthbearerConfig) ->
    ok;
call_oauthbearer_token_refresh_callback(C, OauthbearerConfig) when is_function(C, 1) ->
    C(OauthbearerConfig);
call_oauthbearer_token_refresh_callback(C, OauthbearerConfig) ->
    C:oauthbearer_token_refresh_callback(OauthbearerConfig).

parralel_exec(Fun, List) ->
    Parent = self(),
    Pids = [spawn_link(fun() -> Fun(E), Parent ! {self(), done} end) || E <- List],
    [receive {Pid, done} -> ok end || Pid <- Pids],
    ok.