Packages

diint_utilites_common_app

1.4.21
1.4.23 1.4.22 1.4.21 1.4.20 1.4.19 1.4.18 1.4.17 1.4.16 1.4.15 1.4.14 1.4.12 1.4.11 1.4.10 1.4.9 1.4.8 1.4.7 1.4.6 1.4.5 1.4.4 1.4.3 1.4.2 1.4.1 1.4.0 1.3.9 1.3.8 1.3.7 1.3.6 1.3.5 1.3.4 1.3.3 1.3.2 1.3.1 1.3.0 1.2.101 1.2.11 1.2.10 1.2.9 1.2.8 1.2.7 1.2.6 1.2.5 1.2.4 1.2.3 1.2.2 1.2.1 1.2.0 1.1.7 1.1.6 1.1.5 1.1.4 1.1.3 1.1.2 1.1.1 1.1.0 1.0.13 1.0.12 1.0.11 1.0.10 1.0.9 1.0.8 1.0.7 1.0.6 1.0.5 1.0.4 1.0.3 1.0.2 1.0.1

библиотеки для работы с ребитом и протоколы

HexDocs

Current section

Files

Jump to
Readme
67 Versions
4 Dependencies
Dependants
Files
Activity
diint_utilites_common_app src mq rabbit_utilites.erl
Raw

src/mq/rabbit_utilites.erl 

%%%-------------------------------------------------------------------

%%% @author cheese

%%% @copyright (C) 2015, <COMPANY>

%%% @doc

%%%

%%% @end

%%% Created : 09. Jan 2015 8:56

%%%-------------------------------------------------------------------

-module(rabbit_utilites).

-author("cheese").

-include("_build/default/lib/amqp_client/include/amqp_client.hrl").

-include("include/types_rabbit.hrl").

-include_lib("kernel/include/logger.hrl").



%% API

-export([close_rpc_channel/3, convert_array_to_rabbit_connection_config/1, connectionMonitor/2]).

-export([connect/2]).



-export([internal_convert_array_to_rabbit_connection_config/2]).



convert_array_to_rabbit_connection_config(Arr) ->

  internal_convert_array_to_rabbit_connection_config(Arr, #rabbit_connection_config{}).



internal_convert_array_to_rabbit_connection_config([], Info) ->

  Info;

internal_convert_array_to_rabbit_connection_config([{Key, Value} | Tail], Info) ->

  case Key of

    process_name ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{process_name = Value});

    host ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{host = Value});

    password ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{password = Value});

    username ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{username = Value});

    virtual_host ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ virtual_host = Value});

    queue_response ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ queue_response = Value});

    exchange_response ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ exchange_response = Value});

    arguments ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info#rabbit_connection_config{ arguments = Value});

    _Other ->

      internal_convert_array_to_rabbit_connection_config(Tail, Info)

  end.



close_rpc_channel(ProcessName, Connection, Channel) ->

  ?LOG_INFO("Try to close ~w: ~w-~w~n", [ProcessName, Connection, Channel]),

  try amqp_channel:close(Channel) of

    ResCloseChannel ->

      ?LOG_INFO("Close ~w chanell ~w: ~w~n", [ProcessName, Channel, ResCloseChannel]),

      try amqp_connection:close(Connection) of

        ResCloseConnection ->

          ?LOG_INFO("Close ~w connection ~w: ~w~n", [ProcessName, Connection, ResCloseConnection])

      catch

        ErrCloseConnection ->

          ?LOG_INFO("Close ~w connection ~w error: ~w~n", [ProcessName, Connection, ErrCloseConnection])

      end

  catch

    ErrCloseChannel ->

      ?LOG_INFO("Close ~w channel ~w error: ~w~n", [ProcessName, Channel, ErrCloseChannel])

  end.



connect(Process, #rabbit_connection_config{ host=Host,  username=Username,  password=Password, virtual_host=VirtHost}) ->

  ?LOG_INFO("Start create connection. Host: ~s, User: ~s, VirtHost: ~s~n", [Host, Username, 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} ->

      ?LOG_INFO("Create ~w connection: ~p~n", [Process, Connection]),

      case amqp_connection:open_channel(Connection) of

        {ok, Channel} ->

          ?LOG_INFO("Create ~w channel: ~p~n", [Process, Channel]),

          {ok, Connection, Channel};

        ErrorChannel ->

          ?LOG_INFO("Create ~w connection error: ~p~n", [Process, ErrorChannel]),

          {error, ErrorChannel}

      end;

    ErrorConnection ->

      ?LOG_INFO("Create ~w connection error: ~p~n", [Process, ErrorConnection]),

      {error, ErrorConnection}

  end.



connectionMonitor(Connection, ProcessName) ->   

   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(ProcessName, reconnect)

   end.