-module(rabbitMQ_logger_handler). -include("_build/default/lib/amqp_client/include/amqp_client.hrl"). -include("include/types_rabbit.hrl"). -include_lib("kernel/include/logger.hrl"). -export([adding_handler/1, removing_handler/1, log/2, changing_config/3]). -export([init/1, handle_call/3, handle_cast/2, terminate/2]). adding_handler(Config) -> MyConfig = maps:get(config,Config), {ok, Pid} = gen_server:start(?MODULE, MyConfig, []), {ok, Config#{config => MyConfig#{pid => Pid}}}. removing_handler(#{config := #{pid := Pid}}) -> gen_server:stop(Pid). changing_config(SetOrUpdate, OldConfig, NewConfig) -> {ok, OldConfig}. log(LogEvent,#{config := #{pid := Pid}} = Config) -> gen_server:cast(Pid, {log, LogEvent, Config}). init(#{rabbitConfig := RabbitConfig, exchange := ExchangeName, queueName := QueueName}) -> RmqConnectionConfig = convert_array_to_rabbit_connection_config(RabbitConfig,#rabbit_connection_config{}), case connect_to_rabbit(RmqConnectionConfig) of {ok, Connection, Channel} -> case amqp_channel:call(Channel, #'exchange.declare'{exchange = ExchangeName}) of #'exchange.declare_ok'{} -> ok; _-> #'exchange.declare_ok'{} = amqp_channel:call(Channel, #'exchange.declare'{exchange = ExchangeName, durable = true}) end, {ok, #{rabbitConfig => RabbitConfig, connection => Connection, channel => Channel, exchange => ExchangeName, queueName => QueueName}}; _ -> {error, "NoConnect to Log Rabbit"} end. handle_call(_, _, State) -> {reply, {error, bad_request}, State}. handle_cast({log, LogEvent, Config}, #{connection := _, channel := _, exchange := _, queueName := _} = State) -> do_log(State, LogEvent, Config), {noreply, State}; handle_cast(need_reconnect, #{rabbitConfig := RC} = State) -> case connect_to_rabbit(RC) of {ok, Connection, NewChannel} -> {noreply, State#{channel => NewChannel, connection => Connection}}; _else -> {noreply, State} end. terminate(_Reason, #{fd := Fd}) -> _ = file:close(Fd), ok. do_log(#{connection := Connection, channel := Channel, exchange := ExchangeName, queueName := QueueName}, LogEvent, #{formatter := {FModule, FConfig}}) -> String = FModule:format(LogEvent, FConfig), Message = unicode:characters_to_binary(String, unicode, utf8), Props = #'P_basic'{}, amqp_channel:cast(Channel, #'basic.publish'{exchange = ExchangeName}, #amqp_msg{payload = Message, props = Props}). %% так надо, потом уберу connect_to_rabbit(RabbitConfig) -> case connect(RabbitConfig) of {ok, Connection, NewChannel} -> spawn(fun() -> connectionMonitor(Connection, self())end), {ok, Connection, NewChannel}; % #'exchange.declare_ok'{} = amqp_channel:call(NewChannel, #'exchange.declare'{exchange = Exchange}), % #'queue.declare_ok'{} = amqp_channel:call(NewChannel, #'queue.declare'{queue = Queue, auto_delete = false, arguments = RabbitConfig#rabbit_connection_config.arguments}), % #'queue.bind_ok'{} = amqp_channel:call(NewChannel, #'queue.bind'{queue = Queue, exchange = Exchange}), Other -> spawn(fun() -> connectionMonitor(undefined, self()) end), Other end. connect(#rabbit_connection_config{ host = Host, username = Username, password = Password, virtual_host=VirtHost}) -> case amqp_connection:start(#amqp_params_network{host = Host, password = Password, username = Username, heartbeat = 10, frame_max = 8388608, virtual_host = VirtHost}) of {ok, Connection} -> case amqp_connection:open_channel(Connection) of {ok, Channel} -> {ok, Connection, Channel}; ErrorChannel -> {error, ErrorChannel} end; ErrorConnection -> {error, ErrorConnection} end. convert_array_to_rabbit_connection_config([], Info) -> Info; convert_array_to_rabbit_connection_config([{Key, Value} | Tail], Info) -> case Key of process_name -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{process_name = Value}); host -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{host = Value}); password -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{password = Value}); username -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{username = Value}); virtual_host -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ virtual_host = Value}); queue_response -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ queue_response = Value}); exchange_response -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ exchange_response = Value}); arguments -> convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ arguments = Value}); _Other -> convert_array_to_rabbit_connection_config(Tail, Info) end. connectionMonitor(Connection, Pid) -> Ref = erlang:monitor(process, Connection), receive {'DOWN', _Reference, process, Pid, Reason} -> ?LOG_INFO("I (parent) My worker (connection) ~p died (~p)~n", [Pid, Reason]), gen_server:call(Pid, need_reconnect) end.