130 lines
5.0 KiB
Erlang
130 lines
5.0 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author anlicheng
|
|
%%% @copyright (C) 2024, <COMPANY>
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 07. 5月 2024 11:17
|
|
%%%-------------------------------------------------------------------
|
|
-module(endpoint_http).
|
|
-author("anlicheng").
|
|
-include("endpoint.hrl").
|
|
|
|
-behaviour(gen_server).
|
|
|
|
%% API
|
|
-export([start_link/2]).
|
|
|
|
%% gen_server callbacks
|
|
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
|
|
|
|
-define(SERVER, ?MODULE).
|
|
|
|
-record(state, {
|
|
endpoint :: #endpoint{},
|
|
buffer :: endpoint_buffer:buffer()
|
|
}).
|
|
|
|
%%%===================================================================
|
|
%%% API
|
|
%%%===================================================================
|
|
|
|
%% @doc Spawns the server and registers the local name (unique)
|
|
-spec(start_link(LocalName :: atom(), Endpoint :: #endpoint{}) ->
|
|
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
|
|
start_link(LocalName, Endpoint = #endpoint{config = #http_endpoint{}}) when is_atom(LocalName) ->
|
|
gen_server:start_link({local, LocalName}, ?MODULE, [Endpoint], []).
|
|
|
|
%%%===================================================================
|
|
%%% gen_server callbacks
|
|
%%%===================================================================
|
|
|
|
%% @private
|
|
%% @doc Initializes the server
|
|
-spec(init(Args :: term()) ->
|
|
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
|
|
{stop, Reason :: term()} | ignore).
|
|
init([Endpoint = #endpoint{matcher = Matcher}]) ->
|
|
endpoint_subscription:subscribe(Matcher, self()),
|
|
Buffer = endpoint_buffer:new(Endpoint, 10),
|
|
{ok, #state{endpoint = Endpoint, buffer = Buffer}}.
|
|
|
|
%% @private
|
|
%% @doc Handling call messages
|
|
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
|
|
State :: #state{}) ->
|
|
{reply, Reply :: term(), NewState :: #state{}} |
|
|
{reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
|
|
{noreply, NewState :: #state{}} |
|
|
{noreply, NewState :: #state{}, timeout() | hibernate} |
|
|
{stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
|
|
{stop, Reason :: term(), NewState :: #state{}}).
|
|
handle_call(get_stat, _From, State = #state{buffer = Buffer}) ->
|
|
Stat = endpoint_buffer:stat(Buffer),
|
|
{reply, {ok, Stat}, State}.
|
|
|
|
%% @private
|
|
%% @doc Handling cast messages
|
|
-spec(handle_cast(Request :: term(), State :: #state{}) ->
|
|
{noreply, NewState :: #state{}} |
|
|
{noreply, NewState :: #state{}, timeout() | hibernate} |
|
|
{stop, Reason :: term(), NewState :: #state{}}).
|
|
handle_cast({forward, Metric}, State = #state{buffer = Buffer}) ->
|
|
NBuffer = endpoint_buffer:append(Metric, Buffer),
|
|
{noreply, State#state{buffer = NBuffer}};
|
|
|
|
handle_cast(cleanup, State = #state{buffer = Buffer}) ->
|
|
endpoint_buffer:cleanup(Buffer),
|
|
{noreply, State}.
|
|
|
|
%% @private
|
|
%% @doc Handling all non call/cast messages
|
|
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
|
|
{noreply, NewState :: #state{}} |
|
|
{noreply, NewState :: #state{}, timeout() | hibernate} |
|
|
{stop, Reason :: term(), NewState :: #state{}}).
|
|
handle_info({next_data, Id, Metric}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{url = Url}}}) ->
|
|
Headers = [
|
|
{<<"Content-Type">>, <<"application/json">>}
|
|
],
|
|
case hackney:request(post, Url, Headers, Metric) of
|
|
{ok, 200, _, ClientRef} ->
|
|
{ok, RespBody} = hackney:body(ClientRef),
|
|
hackney:close(ClientRef),
|
|
lager:debug("[endpoint_http] url: ~p, response is: ~p", [Url, RespBody]),
|
|
NBuffer = endpoint_buffer:ack(Id, Buffer),
|
|
|
|
{noreply, State#state{buffer = NBuffer}};
|
|
{ok, HttpCode, _, ClientRef} ->
|
|
{ok, RespBody} = hackney:body(ClientRef),
|
|
hackney:close(ClientRef),
|
|
lager:debug("[endpoint_http] url: ~p, http_code: ~p, response is: ~p", [Url, HttpCode, RespBody]),
|
|
NBuffer = endpoint_buffer:ack(Id, Buffer),
|
|
{noreply, State#state{buffer = NBuffer}};
|
|
{error, Reason} ->
|
|
lager:warning("[endpoint_http] url: ~p, get error: ~p", [Url, Reason]),
|
|
{noreply, State}
|
|
end.
|
|
|
|
%% @private
|
|
%% @doc This function is called by a gen_server when it is about to
|
|
%% terminate. It should be the opposite of Module:init/1 and do any
|
|
%% necessary cleaning up. When it returns, the gen_server terminates
|
|
%% with Reason. The return value is ignored.
|
|
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
|
|
State :: #state{}) -> term()).
|
|
terminate(_Reason, _State = #state{}) ->
|
|
ok.
|
|
|
|
%% @private
|
|
%% @doc Convert process state when code is changed
|
|
-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
|
|
Extra :: term()) ->
|
|
{ok, NewState :: #state{}} | {error, Reason :: term()}).
|
|
code_change(_OldVsn, State = #state{}, _Extra) ->
|
|
{ok, State}.
|
|
|
|
%%%===================================================================
|
|
%%% Internal functions
|
|
%%%===================================================================
|