%%------------------------------------------------------------------- %% Copyright (c) 2020 Vasu Dasari vdasari@gmail.com %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. %% You may obtain a copy of the License at: %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, software %% distributed under the License is distributed on an "AS IS" BASIS, %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% See the License for the specific language governing permissions and %% limitations under the License. %% %% @author Vasu Dasari %% @doc Module that deals with message transport methods %% @private %% @end %% Created : 15. Jun 2020 2:44 PM %%------------------------------------------------------------------- -module(ovsdb_comms). -author("Vasu Dasari"). -include("ovsdb_client.hrl"). %% API -export([connect/1, restart/2, recv_data/2, send_data/3]). recv_data(<<>>, #ovsdb_state{socket = Socket} = State) -> ok = setopts(tcp, Socket, [{active, once}]), State; recv_data(Data, #ovsdb_state{socket = Socket, msg_buffer = MsgBuf} = State) -> CompleteBuf = <>, case jsone:try_decode(CompleteBuf) of {ok, Message, Left} -> ?DEBUG("Received ~s", [ovsdb_utils:pretty_print(Message)]), recv_data(Left, ovsdb_protocol:process_message(Message, State#ovsdb_state{msg_buffer = <<>>})); {error, _Error} -> ok = setopts(tcp, Socket, [{active, once}]), State#ovsdb_state{msg_buffer = CompleteBuf} end. send_data(From, #{id := 0} = Message, State) -> send_data(From, Message#{id => ?next_msg_id()}, State); send_data(From, #{id := Id} = Message, #ovsdb_state{ socket = Socket, proto = Protocol, pending_messages = PendingMessages} = State) -> ?DEBUG("Sending ~s", [ovsdb_utils:pretty_print(Message)]), ok = send(Protocol, Socket, jsone:encode(Message)), State#ovsdb_state{ pending_messages = maps:put(Id, #{ from => From, reply => noreply }, PendingMessages) }. connect(#ovsdb_state{ proc = Proc, proto = Protocol, ip_addr = Host, port = Port, timeout = Timeout} = State) -> Lib = transport_lib(Protocol), case Lib:connect(Host, Port, opts(Protocol)) of {ok, Socket} -> ?INFO("~p: Connected: Socket ~p", [Proc, Socket]), ok = setopts(Protocol, Socket, [{active, once}]), self() ! connected, State#ovsdb_state{ socket = Socket }; {error, _Reason} -> erlang:send_after(Timeout, self(), connect), State end; connect(State) -> State. restart(Reason, #ovsdb_state{ proc = Proc, proto = Protocol, socket = Socket, timeout = Timeout} = State) -> ?INFO("~p: Restart reason: ~p", [Proc, Reason]), ok = close(Protocol, Socket), erlang:send_after(Timeout, self(), connect), State#ovsdb_state{socket = not_connected}. opts(tcp) -> [binary, {reuseaddr, true}, {active, once}]; opts(ssl) -> opts(tcp) ++ [{verify, verify_peer}, {fail_if_no_peer_cert, true}] ++ [{cert, base64:decode(Cert)} || {ok, Cert} <- [application:get_env(linc, certificate)]] ++ [{key, {'RSAPrivateKey', base64:decode(Key)}} || {ok, Key} <- [application:get_env(linc, rsa_private_key)]]. setopts(_, not_connected, _) -> ok; setopts(ssl, Socket, Opts) -> ssl:setopts(Socket, Opts); setopts(tcp, Socket, Opts) -> inet:setopts(Socket, Opts). send(_, not_connected, _) -> ok; send(Protocol, Socket, Data) -> (transport_lib(Protocol)):send(Socket, Data). close(_, not_connected) -> ok; close(Protocol, Socket) -> Lib = transport_lib(Protocol), Lib:close(Socket). transport_lib(Protocol) -> case Protocol of tcp -> gen_tcp; ssl -> ssl end.