%%------------------------------------------------------------------- %% @doc Polling server %% @private %% @end %%------------------------------------------------------------------- -module(ldclient_update_poll_server). -behaviour(gen_server). %% Supervision -export([start_link/1, init/1]). %% Behavior callbacks -export([code_change/3, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). -type state() :: #{ sdk_key := string(), feature_store := atom(), storage_tag := atom(), requestor := atom(), requestor_state := any(), poll_uri := string(), poll_interval := pos_integer(), timer_ref => timer:tref() }. %% Constants -define(LATEST_ALL_PATH, "/sdk/latest-all"). -ifdef(TEST). -compile(export_all). -endif. %%=================================================================== %% Supervision %%=================================================================== %% @doc Starts the server %% %% @end -spec start_link(Tag :: atom()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}. start_link(Tag) -> error_logger:info_msg("Starting polling update server for ~p", [Tag]), gen_server:start_link(?MODULE, [Tag], []). -spec init(Args :: term()) -> {ok, State :: state()} | {ok, State :: state(), timeout() | hibernate} | {stop, Reason :: term()} | ignore. init([Tag]) -> SdkKey = ldclient_config:get_value(Tag, sdk_key), FeatureStore = ldclient_config:get_value(Tag, feature_store), Requestor = ldclient_config:get_value(Tag, polling_update_requestor), PollUri = ldclient_config:get_value(Tag, base_uri) ++ ?LATEST_ALL_PATH, PollInterval = ldclient_config:get_value(Tag, polling_interval) * 1000, % Need to trap exit so supervisor:terminate_child calls terminate callback process_flag(trap_exit, true), State = #{ sdk_key => SdkKey, feature_store => FeatureStore, storage_tag => Tag, requestor => Requestor, requestor_state => Requestor:init(Tag, SdkKey), poll_uri => PollUri, poll_interval => PollInterval }, self() ! {listen}, {ok, State}. %%=================================================================== %% Behavior callbacks %%=================================================================== -type from() :: {pid(), term()}. -spec handle_call(Request :: term(), From :: from(), State :: state()) -> {reply, Reply :: term(), NewState :: state()} | {stop, normal, {error, atom(), term()}, state()}. handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(_Request, State) -> {noreply, State}. handle_info({listen}, #{poll_interval := PollInterval} = State) -> case maps:find(timer_ref, State) of {ok, _ExistingTimer} -> {noreply, ok, State}; error -> {ok, TimerRef} = timer:send_interval(PollInterval, {poll}), {noreply, poll(State#{timer_ref => TimerRef})} end; handle_info({poll}, State) -> {noreply, poll(State)}; handle_info(_Info, State) -> {noreply, State}. -spec terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), State :: state()) -> term(). terminate(Reason, #{timer_ref := TimerRef} = _State) -> error_logger:info_msg("Terminating polling, reason: ~p", [Reason]), _ = timer:cancel(TimerRef), ok; terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%=================================================================== %% Internal functions %%=================================================================== -spec poll(state()) -> state(). poll(#{ feature_store := FeatureStore, requestor := Requestor, requestor_state := RequestorState, poll_uri := Uri, storage_tag := Tag } = State) -> {Result, NewRequestorState} = Requestor:all(Uri, RequestorState), ok = process_response(Result, FeatureStore, Tag, Uri), true = ldclient_update_processor_state:set_initialized_state(Tag, true), State#{requestor_state := NewRequestorState}. -spec process_response(ldclient_update_requestor:response(), atom(), atom(), string()) -> ok. process_response({error, {bad_status, 401, _Reason}}, _, _, Uri) -> error_logger:warning_msg("Invalid SDK key when when polling for updates at URL ~p. Verify that your SDK key is correct.", [Uri]), ok; process_response({error, {bad_status, 404, _Reason}}, _, _, Uri) -> error_logger:warning_msg("Resource not found when polling for updates at URL ~p.", [Uri]), ok; process_response({error, {bad_status, StatusCode, Reason}}, _, _, Uri) when StatusCode >= 300 -> error_logger:warning_msg("Unexpected response code: ~p when polling for updates at URL ~p: ~p.", [StatusCode, Uri, Reason]), ok; process_response({error, network_error}, _, _, Uri) -> error_logger:warning_msg("Failed to connect to update server at: %p", [Uri]), ok; process_response({ok, not_modified}, _, _, _) -> ok; process_response({ok, ResponseBody}, FeatureStore, Tag, _) -> StorageDown = ldclient_update_processor_state:get_storage_initialized_state(Tag), case StorageDown of false -> ldclient_update_processor_state:set_storage_initialized_state(Tag, reload); _ -> ok end, process_response_body(ResponseBody, FeatureStore, Tag). -spec process_response_body(binary(), atom(), atom()) -> ok. process_response_body(ResponseBody, ldclient_storage_redis, Tag) -> Data = jsx:decode(ResponseBody, [return_maps]), [Flags, Segments] = get_put_items(Data), error_logger:info_msg("Received poll event with ~p flags and ~p segments", [maps:size(Flags), maps:size(Segments)]), ok = ldclient_storage_redis:upsert_clean(Tag, features, Flags), ok = ldclient_storage_redis:upsert_clean(Tag, segments, Segments), ok = ldclient_storage_redis:set_init(Tag); process_response_body(ResponseBody, FeatureStore, Tag) -> Data = jsx:decode(ResponseBody, [return_maps]), [Flags, Segments] = get_put_items(Data), ParsedFlags = maps:map( fun(_K, V) -> ldclient_flag:new(V) end , Flags), ParsedSegments = maps:map( fun(_K, V) -> ldclient_segment:new(V) end , Segments), ok = FeatureStore:upsert_clean(Tag, features, ParsedFlags), ok = FeatureStore:upsert_clean(Tag, segments, ParsedSegments). -spec get_put_items(Data :: map()) -> [map()]. get_put_items(#{<<"flags">> := Flags, <<"segments">> := Segments}) -> [Flags, Segments].