%%%------------------------------------------------------------------- %%% @author cheese %%% @copyright (C) 2016, %%% @doc %%% %%% @end %%% Created : 19. Apr 2016 11:12 %%%------------------------------------------------------------------- -module(rabbit_rpc2_reader). -author("cheese"). -behaviour(gen_server). %% API -export([start_link/1, get_queue_name/1]). %% Internal -export([get_process_name/1, get_logfile_name/1, generate_new_resp_queue_name/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(SERVER, ?MODULE). -include("include/types_rabbit.hrl"). -include("_build/default/lib/amqp_client/include/amqp_client.hrl"). -include_lib("kernel/include/logger.hrl"). -record(state, { connection_config :: #rabbit_connection_config{}, queue, exchange, arguments, log_file, channel, connection, last_message_date }). %%%=================================================================== %%% API %%%=================================================================== start_link(RabbitConnectionConfig) -> rabbit_rpc2_stat:fold_response_queue(), gen_server:start_link({local, get_process_name(RabbitConnectionConfig)}, ?MODULE, [RabbitConnectionConfig], []). get_queue_name(RabbitConnectionConfig) -> ProcessName = get_process_name(RabbitConnectionConfig), gen_server:call(ProcessName, get_queue_name). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== init([RabbitConnectionConfig]) -> erlang:send_after(2000, self(), check_connection), {ok, #state{ arguments = RabbitConnectionConfig#rabbit_connection_config.arguments, connection_config = RabbitConnectionConfig, exchange = RabbitConnectionConfig#rabbit_connection_config.exchange_response, queue = generate_new_resp_queue_name(RabbitConnectionConfig#rabbit_connection_config.queue_response)} }. handle_call(get_queue_name, _From, State) -> {reply, State#state.queue, State}; handle_call(reconnect,_From, #state{connection_config = RabbitConfig, queue = Queue, exchange = Exchange, arguments = Arguments} = State) -> case connectAndDeclare(RabbitConfig, Exchange, Queue,Arguments) of {ok, NewChannel, NewConnection} -> {noreply, State#state{channel = NewChannel, connection = NewConnection, last_message_date = calendar:local_time()}}; {error, Reason} -> erlang:error(sql_connection_error), ?LOG_INFO("Check rabbit connection error: ~w~n", [Reason]), {noreply, State} end; handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(_Request, State) -> {noreply, State}. handle_info(check_connection, #state{connection_config = RabbitConfig, queue = Queue, channel = Channel, exchange = Exchange, arguments = Arguments, last_message_date = LastMessage} = State) -> % erlang:send_after(10000, self(), check_connection), % check_timeout(LastMessage, Channel, Connection, LogFile), case Channel of undefined -> ?LOG_ERROR("connection undefined", []), case connectAndDeclare(RabbitConfig, Exchange, Queue,Arguments) of {ok, NewChannel, NewConnection} -> {noreply, State#state{channel = NewChannel, connection = NewConnection, last_message_date = calendar:local_time()}}; {error, Reason} -> {noreply, State} end; _ -> ?LOG_INFO("reader channel ok~n", []), {noreply, State} end; handle_info(#'basic.consume_ok'{}, #state{} = State) -> ?LOG_INFO("basic.consume_ok~n", []), {noreply, State}; handle_info(#'basic.cancel_ok'{}, #state{} = State) -> ?LOG_INFO("basic.cancel_ok~n", []), exit({error, "basic.cancel_ok"}), {noreply, State}; handle_info({#'basic.deliver'{delivery_tag = Tag}, #amqp_msg{props = #'P_basic'{correlation_id = CorrelationId}, payload = Body}}, #state{channel = Channel, connection_config = RabbitConfig, queue = QueueName} = State) -> ?LOG_INFO("Receive from rabbit ~s: ~s~n", [CorrelationId, bytes_extension:bin_to_hexstr(Body)]), amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag}), rabbit_rpc2:on_rabbit_response(RabbitConfig, CorrelationId, Body, QueueName), {noreply, State#state{last_message_date = calendar:local_time()}}; handle_info(Info, #state{} = State) -> ?LOG_INFO("Unproc info: ~w~n", [Info]), timer:sleep(1000), {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== get_process_name(RabbitConnectionConfig) -> list_to_atom(atom_to_list(RabbitConnectionConfig#rabbit_connection_config.process_name) ++ "_rpc2_reader_" ++ binary_to_list(RabbitConnectionConfig#rabbit_connection_config.queue_response)). get_logfile_name(RabbitConnectionConfig) -> "log/" ++ atom_to_list(RabbitConnectionConfig#rabbit_connection_config.process_name) ++ "_rpc2_reader_" ++ binary_to_list(RabbitConnectionConfig#rabbit_connection_config.queue_response) ++ ".log". generate_new_resp_queue_name(Name) -> UUID = bytes_extension:generate_uuid(), {{Year, Month, Day}, {Hour, Min, Second}} = calendar:local_time(), Date = list_to_binary(io_lib:fwrite("~w.~w.~w_~w:~w:~w", [Year, Month, Day, Hour, Min, Second])), <>. % check_timeout(LastMessage, Channel, Connection, LogFile) -> % case LastMessage of % {{Year, Month, Day}, {Hour, Min, Second}} -> % NowSeconds = time_utilites:get_seconds(calendar:local_time()), % LastSeconds = time_utilites:get_seconds({{Year, Month, Day}, {Hour, Min, Second}}), % DiffSeconds = NowSeconds - LastSeconds, % case DiffSeconds > 180 of % true -> % ?LOG_INFO("~w seconds without messages, try to close rabbit response queue ~w~n", [DiffSeconds, Channel], LogFile), % amqp_channel:close(Channel), % amqp_connection:close(Connection), % erlang:error(rabbit_rpc_too_long_time_without_messages); % _ -> % ok % end; % _ -> % ?LOG_INFO("LastMessage: ~w~n", [LastMessage], LogFile), % ok % end. connectAndDeclare(RabbitConfig, Exchange, Queue,Arguments) -> Process = get_process_name(RabbitConfig), case rabbit_utilites:connect(Process, RabbitConfig ) of {ok, NewConnection, NewChannel} -> spawn(rabbit_utilites,connectionMonitor,[NewConnection, Process]), ?LOG_INFO("connection create ok. Exchange: ~s~n", [Exchange]), case amqp_channel:call(NewChannel, #'exchange.declare'{exchange = Exchange, durable = true}) of #'exchange.declare_ok'{}-> ok; _ -> #'exchange.declare_ok'{} = amqp_channel:call(NewChannel, #'exchange.declare'{exchange = Exchange, durable = false }) end, ?LOG_INFO("exchange.declare_ok. Queue: ~s~n", [Queue]), case amqp_channel:call(NewChannel, #'queue.declare'{queue = Queue,durable = true, auto_delete = true, arguments = Arguments}) of #'queue.declare_ok'{} -> ok; _ -> #'queue.declare_ok'{} = amqp_channel:call(NewChannel, #'queue.declare'{queue = Queue,durable = false, auto_delete = true, arguments = Arguments}) end, ?LOG_INFO("queue.declare_ok~n", []), #'basic.consume_ok'{} = amqp_channel:call(NewChannel, #'basic.consume'{queue = Queue}), ?LOG_INFO("basic.consume_ok~n", []), {ok, NewChannel, NewConnection}; {error, Reason} -> spawn(rabbit_utilites,connectionMonitor,[undefined, Process]), erlang:error(sql_connection_error), ?LOG_INFO("Check rabbit connection error: ~w~n", [Reason]), {error, Reason} end.