Current section
Files
Jump to
Current section
Files
src/fcm.erl
%%% Copyright (c) 2016 Olle Mattsson <olle@rymdis.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.
-module(fcm).
-include_lib("fcm/include/fcm.hrl").
-behaviour(gen_server).
-export([start/0, start/1]).
-export([stop/0]).
-export([push/2, push_to_ids/2, push_to_id/2]).
-export([retry/1, abort/1, abort/2]).
%% gen_server callbacks
-export([start_push/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-export([register/2, register/1]).
-export([delete_token/1]).
-export([update_token/2, update_id/2]).
-export([is_registered/1]).
-export([get_tokens/1, get_id/1]).
-export([save_message/1, get_message/1]).
-export_type([token/0, id/0]).
-type token() :: binary().
-type id() :: any().
-export_type([message_id/0]).
-type message_id() :: any().
-record(state, {message :: #fcm_message{},
tries = 0 :: integer(),
timer :: timer:tref() | undefined,
retry_messages :: list()}).
start() ->
start(temporary).
start(Type) ->
case application:ensure_all_started(fcm, Type) of
{ok, _} ->
ok;
Other ->
Other
end.
stop() ->
application:stop(fcm).
%% Start new request
-spec push(token() | [token()],
proplists:proplist() | map()) ->
supervisor:startchild_ret().
push(Tokens, Content) when is_list(Tokens) ->
Message = #fcm_message{to = Tokens,
content = Content},
Child = #{id => make_ref(),
restart => temporary,
start => {fcm, start_push, [Message]}},
supervisor:start_child(fcm_sup, Child);
push(Token, Content) ->
push([Token], Content).
-spec push_to_ids([id()], proplists:proplist() | map()) ->
supervisor:startchild_ret().
push_to_ids(Ids, Content) when is_list(Ids) ->
Tokens = lists:foldl(fun(Id, Acc) -> get_tokens(Id) ++ Acc end, [], Ids),
push(Tokens, Content).
-spec push_to_id(id(), proplists:proplist() | map()) ->
supervisor:startchild_ret().
push_to_id(Id, Content) ->
push_to_ids([Id], Content).
%% Override retry timer
-spec retry(pid()) -> ok.
retry(Pid) ->
gen_server:cast(Pid, retry).
%% Abort all resends
-spec abort(pid()) -> [#fcm_message{}].
abort(Pid) ->
gen_server:call(Pid, abort).
-spec abort(pid(), pos_integer() | infinity) -> [#fcm_message{}].
abort(Pid, Timeout) ->
gen_server:call(Pid, abort, Timeout).
-spec register(token()) -> boolean().
register(Token) ->
Transaction =
fun() ->
case is_registered(Token) of
true -> false;
false -> mnesia:write(#fcm_key{token = Token})
end
end,
case mnesia:transaction(Transaction) of
{atomic, ok} ->
true;
_ ->
false
end.
-spec register(token(), id()) -> boolean().
register(Token, Id) ->
Transaction =
fun() ->
case is_registered(Token) of
true -> false;
false -> mnesia:write(#fcm_key{token = Token, id = Id})
end
end,
case mnesia:transaction(Transaction) of
{atomic, ok} ->
true;
_ ->
false
end.
-spec is_registered(token()) -> boolean().
is_registered(Token) ->
Transaction =
fun() ->
[] /= mnesia:read(fcm_key, Token)
end,
case mnesia:transaction(Transaction) of
{atomic, Result} ->
Result;
_ ->
false
end.
-spec update_token(token(), token()) -> boolean().
update_token(OldToken, NewToken) ->
Transaction =
fun() ->
case mnesia:wread({fcm_key, OldToken}) of
[Key] -> mnesia:write(Key#fcm_key{token = NewToken});
_ -> false
end
end,
case mnesia:transaction(Transaction) of
{atomic, ok} ->
true;
_ ->
false
end.
-spec delete_token(token()) -> boolean().
delete_token(Token) ->
Transaction =
fun() ->
mnesia:delete({fcm_key, Token})
end,
case mnesia:transaction(Transaction) of
{atomic, ok} ->
true;
_ ->
false
end.
-spec update_id(token(), id()) -> boolean().
update_id(Token, Id) ->
Transaction =
fun() ->
case mnesia:wread({fcm_key, Token}) of
[Key] -> mnesia:write(Key#fcm_key{id = Id});
_ -> false
end
end,
case mnesia:transaction(Transaction) of
{atomic, ok} ->
true;
_ ->
false
end.
-spec get_id(token()) -> id() | undefined.
get_id(Token) ->
Transaction =
fun() ->
mnesia:read(fcm_key, Token)
end,
case mnesia:transaction(Transaction) of
{atomic, [#fcm_key{id = Id}]} ->
Id;
_ ->
undefined
end.
-spec get_tokens(id()) -> [token()].
get_tokens(UserId) ->
Transaction =
fun() ->
mnesia:select(fcm_key, [{#fcm_key{id = UserId,
token = '$1',
_ = '_'},
[],
['$1']}])
end,
case mnesia:transaction(Transaction) of
{atomic, Tokens} ->
Tokens;
_ ->
[]
end.
-spec save_message(#fcm_message{}) -> boolean().
save_message(Message = #fcm_message{}) ->
Transaction =
fun() ->
case mnesia:read(fcm_key, Message#fcm_message.id) of
[] -> mnesia:write(Message);
_ -> false
end
end,
case mnesia:transaction(Transaction) of
{atomic, ok} ->
Message#fcm_message.id;
_ ->
false
end.
-spec get_message(message_id()) -> #fcm_message{} | undefined.
get_message(Id) ->
Transaction =
fun() ->
mnesia:read(fcm_message, Id)
end,
case mnesia:transaction(Transaction) of
{atomic, [Message]} ->
Message;
_ ->
undefined
end.
%% Local start function
start_push(Message) ->
gen_server:start_link(?MODULE, [Message], []).
%% Callbacks
init([Message]) ->
lager:debug("Starting worker"),
case send(Message) of
{ok, _ReqId} ->
{ok, #state{message = Message}};
{error, Reason} ->
{error, Reason}
end.
handle_call(abort, _From, State) ->
timer:cancel(State#state.timer),
lager:debug("Message aborted ~p", [lager:pr(State, ?MODULE)]),
AbortedIds =
lists:map(fun(Message) ->
Id = crypto:hash(sha, term_to_binary(Message)),
fcm:save_message(Message#fcm_message{id = Id,
results = [aborted]}),
Id
end,
State#state.retry_messages),
{stop, aborted, AbortedIds, State}.
handle_cast(retry, State) ->
timer:cancel(State#state.timer),
lists:foreach(fun resend/1, State#state.retry_messages),
{noreply, State#state{tries = State#state.tries+1,
retry_messages = []}}.
handle_info({http, {_ReqId, {{_, 200, _}, Headers, Body}}}, State) ->
Json = jsx:decode(Body, [return_maps]),
lager:debug("Return json ~p", [Json]),
Failures = maps:get(<<"failure">>, Json),
CannonicalIds = maps:get(<<"canonical_ids">>, Json),
if Failures == 0 andalso CannonicalIds == 0 ->
lager:debug("Sent successfully"),
{stop, normal, State};
true ->
RetryMessages = get_retry_messages(State#state.message, Json),
Timeout = proplists:get_value("retry-after", Headers),
TimeoutInSeconds = timeout_in_seconds(Timeout, State#state.tries),
{ok, Timer} = timer:apply_after(timer:seconds(TimeoutInSeconds),
?MODULE, retry, [self()]),
{noreply, State#state{retry_messages = RetryMessages,
timer = Timer}}
end;
handle_info({http, {_ReqId, {{_, Code, _}, Headers, Body}}}, State)
when Code >= 500 andalso Code < 600 ->
OrigMessage = State#state.message,
RetryMessages =
try jsx:decode(Body, [return_maps]) of
Json ->
lager:debug("Return json ~p", [Json]),
get_retry_messages(State#state.message, Json)
catch Error:Reason ->
lager:error("Response not json. ~p\nStacktrace:\n~p",
[Body,
lager:pr_stacktrace(erlang:get_stacktrace(),
{Error,Reason})]),
if is_list(OrigMessage#fcm_message.to) ->
lists:map(fun(To) ->
OrigMessage#fcm_message{to = To}
end, OrigMessage#fcm_message.to);
true ->
[OrigMessage]
end
end,
Timeout = proplists:get_value("retry-after", Headers),
TimeoutInSeconds = timeout_in_seconds(Timeout, State#state.tries),
{ok, Timer} = timer:apply_after(timer:seconds(TimeoutInSeconds),
?MODULE, retry, [self()]),
{noreply, State#state{retry_messages = RetryMessages,
timer = Timer}};
handle_info({http, {_ReqId, {{_, 400, _}, _Headers, Body}}}, State) ->
lager:error("Bad request: ~p\nMessage: ~p",
[Body, lager:pr(State, ?MODULE)]),
{stop, {error, bad_request}, State};
handle_info({http, {_ReqId, {{_, 401, _}, _Headers, _Body}}}, State) ->
lager:error("Unauthorized - ~p",
[lager:pr(State, ?MODULE)]),
{stop, {error, unauthorized}, State};
handle_info({http, {_ReqId, {{_, Code, _}, Headers, Body}}}, State) ->
lager:error("Status code: ~b\nHeaders: ~p\nBody: ~p", [Code, Headers, Body]),
{stop, {error, bad_return}, State};
handle_info({http, {_ReqId, {error, Reason}}}, State) ->
lager:error("httpc error ~p", [Reason]),
{stop, {error, {httpc, Reason}}, State}.
terminate(_Reason, State) ->
lager:debug("Exiting worker ~p", [lager:pr(State, ?MODULE)]),
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%% Local functions
send_url() ->
application:get_env(fcm, send_url, "https://fcm.googleapis.com/fcm/send").
send(Message) ->
send(Message, self()).
send(Message = #fcm_message{to = [To]}, Pid) ->
send(Message#fcm_message{to = To}, Pid);
send(#fcm_message{to = To, content = Content}, Pid) ->
case application:get_env(api_key) of
{ok, ApiKey} ->
Headers = [{"authorization", io_lib:format("key=~s", [ApiKey])}],
ToKey =
if is_list(To) -> <<"registration_ids">>;
true -> <<"to">>
end,
FCM =
if is_map(Content) -> Content#{ToKey => To};
is_list(Content) -> [{ToKey, To}|Content]
end,
Json = jsx:encode(FCM),
httpc:request(post, {send_url(), Headers, "application/json", Json}, [],
[{sync, false}, {receiver, Pid}]);
undefined ->
lager:error("No defined api key"),
{error, "Undefined API key"}
end.
resend(Message = #fcm_message{}) ->
Self = self(),
spawn(fun() ->
%% Sleep for 1-5 seconds to not encounter the same error
%% again if there was an internal error at googles side
Min = 1, Max = 5,
timer:sleep(rand:uniform(Max-Min)+Min),
send(Message, Self)
end).
get_retry_messages(Message, Json) when is_map(Json) ->
Results = maps:get(<<"results">>, Json),
MulticastId = maps:get(<<"multicast_id">>, Json),
fcm:save_message(Message#fcm_message{id = MulticastId,
results = Results}),
ResultsWithTo = lists:zip(Message#fcm_message.to, Results),
RetryTokens = parse_results(ResultsWithTo),
lists:map(fun(Token) ->
Message#fcm_message{to = Token}
end, RetryTokens).
parse_results(Results) ->
parse_results(Results, []).
parse_results([], Acc) ->
Acc;
parse_results([{OldToken,
#{<<"message_id">> := _MessageId,
<<"registration_id">> := NewToken}}|Rest], Acc) ->
lager:debug("Token changed"),
fcm:update_token(OldToken, NewToken),
parse_results(Rest, Acc);
parse_results([{_Token,
#{<<"message_id">> := MessageId}}|Rest], Acc) ->
lager:debug("Success sending message ~p", [MessageId]),
parse_results(Rest, Acc);
parse_results([{Token,
#{<<"error">> := <<"NotRegistered">>}}|Rest], Acc) ->
lager:warning("Token not registered, deleting"),
fcm:delete_token(Token),
parse_results(Rest, Acc);
parse_results([{Token,
#{<<"error">> := <<"Unavailable">>}}|Rest], Acc) ->
parse_results(Rest, [Token|Acc]);
parse_results([{Token,
#{<<"error">> := <<"InternalServerError">>}}|Rest], Acc) ->
lager:warning("Internal server error, try again later"),
parse_results(Rest, [Token|Acc]);
parse_results([{_Token,
#{<<"error">> := Error}}|Rest], Acc) ->
lager:error("Unrecoverable error: ~p", [Error]),
parse_results(Rest, Acc);
parse_results([_|Rest], Acc) ->
parse_results(Rest, Acc).
timeout_in_seconds(undefined, Tries) ->
round(math:pow(2,Tries));
timeout_in_seconds(Timeout, Tries) ->
try list_to_integer(Timeout) of
Seconds -> Seconds
catch
_:_ ->
try httpd_util:convert_request_date(Timeout) of
Date ->
calendar:datetime_to_gregorian_seconds(Date) -
calendar:datetime_to_gregorian_seconds(erlang:universaltime())
catch _:_ ->
timeout_in_seconds(undefined, Tries)
end
end.