%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- % ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et nomod: %%% %%%------------------------------------------------------------------------ %%% @doc %%% ==CloudI Send Queue== %%% For internal services that need to have successful service request sends %%% (i.e., a timeout does not occur because a response is received within the %%% timeout period) with each send done asynchronously and all tracking done %%% with in-memory data. %%% %%% The data is not queued within this module's internal data structure, %%% but like all service requests, the queueing occurs in the destination %%% service process. This module's internal data provides tracking based %%% on the service request transaction id so many asynchronous service %%% requests may be easily managed concurrently. The alternative is to rely %%% on synchronous service requests and increase the sending service's %%% process count as necessary for concurrency, which is a simpler approach. %%% So, cloudi_queue usage is necessary when multiple service requests must %%% be sent concurrently and every request must receive a response %%% (receiving a response is the only proof the transaction was successful). %%% %%% Only one instance of cloudi_queue is necessary within a service's state %%% due to the transaction id being globally unique. %%% @end %%% %%% BSD LICENSE %%% %%% Copyright (c) 2015, Michael Truog %%% All rights reserved. %%% %%% Redistribution and use in source and binary forms, with or without %%% modification, are permitted provided that the following conditions are met: %%% %%% * Redistributions of source code must retain the above copyright %%% notice, this list of conditions and the following disclaimer. %%% * Redistributions in binary form must reproduce the above copyright %%% notice, this list of conditions and the following disclaimer in %%% the documentation and/or other materials provided with the %%% distribution. %%% * All advertising materials mentioning features or use of this %%% software must display the following acknowledgment: %%% This product includes software developed by Michael Truog %%% * The name of the author may not be used to endorse or promote %%% products derived from this software without specific prior %%% written permission %%% %%% THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND %%% CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, %%% INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES %%% OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE %%% DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR %%% CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, %%% SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, %%% BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR %%% SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS %%% INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, %%% WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING %%% NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE %%% OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH %%% DAMAGE. %%% %%% @author Michael Truog %%% @copyright 2015 Michael Truog %%% @version 1.5.1 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_queue). -author('mjtruog [at] gmail (dot) com'). %% external interface -export([new/0, new/1, recv/3, send/4, send/5, send/6, send/7, send/8, timeout/3]). -include("cloudi_core_i_constants.hrl"). -record(request, { name :: cloudi_service:service_name(), request_info :: cloudi_service:request_info(), request :: cloudi_service:request(), timeout :: cloudi_service:timeout_milliseconds(), priority :: cloudi_service:priority(), pattern_pid :: cloudi_service:pattern_pid(), retry_count = 0 :: non_neg_integer() }). -record(cloudi_queue, { retry_max :: non_neg_integer(), validate_request_info :: fun((any()) -> boolean()), validate_request :: fun((any(), any()) -> boolean()), validate_response_info :: fun((any()) -> boolean()), validate_response :: fun((any(), any()) -> boolean()), failures_source_die :: boolean(), failures_source_max_count :: pos_integer(), failures_source_max_period :: infinity | pos_integer(), failures_source = [] :: list(erlang:timestamp()), requests = dict:new() :: dict_proxy(cloudi_service:trans_id(), #request{}) }). -include("cloudi_service.hrl"). -define(DEFAULT_RETRY, 0). % see below: % a retry doesn't count as a failure, until it fails completely % (i.e., hit the max retry count or send returns an error) -define(DEFAULT_VALIDATE_REQUEST_INFO, undefined). -define(DEFAULT_VALIDATE_REQUEST, undefined). -define(DEFAULT_VALIDATE_RESPONSE_INFO, undefined). -define(DEFAULT_VALIDATE_RESPONSE, fun (<<>>, <<>>) -> false; (_, _) -> true end). -define(DEFAULT_FAILURES_SOURCE_DIE, false). -define(DEFAULT_FAILURES_SOURCE_MAX_COUNT, 1). % see below: % (similar to the MaxR configuration value for services) -define(DEFAULT_FAILURES_SOURCE_MAX_PERIOD, infinity). % seconds, see below: % (similar to the MaxT configuration value for services) % If you want the source service to eventually fail, % use the service's MaxT/MaxR as the failures_source_max_period value % (e.g., 300/5 == 60 seconds). Can also use the value 'infinity' % to accumulate a failure count indefinitely. %%%------------------------------------------------------------------------ %%% External interface functions %%%------------------------------------------------------------------------ -type options() :: list({retry, non_neg_integer()} | {validate_request_info, fun((RequestInfo :: cloudi_service:request_info()) -> boolean()) | {Module1 :: module(), Function1 :: atom()}} | {validate_request, fun((RequestInfo :: cloudi_service:request_info(), Request :: cloudi_service:request()) -> boolean()) | {Module2 :: module(), Function2 :: atom()}} | {validate_response_info, fun((ResponseInfo :: cloudi_service:response_info()) -> boolean()) | {Module3 :: module(), Function3 :: atom()}} | {validate_response, fun((ResponseInfo :: cloudi_service:response_info(), Response :: cloudi_service:response()) -> boolean()) | {Module4 :: module(), Function4 :: atom()}} | {failures_source_die, boolean()} | {failures_source_max_count, pos_integer()} | {failures_source_max_period, infinity | pos_integer()}). -type state() :: #cloudi_queue{}. -export_type([options/0, state/0]). %%------------------------------------------------------------------------- %% @doc %% ===Create a CloudI queue.=== %% @end %%------------------------------------------------------------------------- -spec new() -> state(). new() -> new([]). %%------------------------------------------------------------------------- %% @doc %% ===Create a CloudI queue.=== %% @end %%------------------------------------------------------------------------- -spec new(Options :: options()) -> state(). new(Options) when is_list(Options) -> Defaults = [ {retry, ?DEFAULT_RETRY}, {validate_request_info, ?DEFAULT_VALIDATE_REQUEST_INFO}, {validate_request, ?DEFAULT_VALIDATE_REQUEST}, {validate_response_info, ?DEFAULT_VALIDATE_RESPONSE_INFO}, {validate_response, ?DEFAULT_VALIDATE_RESPONSE}, {failures_source_die, ?DEFAULT_FAILURES_SOURCE_DIE}, {failures_source_max_count, ?DEFAULT_FAILURES_SOURCE_MAX_COUNT}, {failures_source_max_period, ?DEFAULT_FAILURES_SOURCE_MAX_PERIOD}], [RetryMax, ValidateRequestInfo0, ValidateRequest0, ValidateResponseInfo0, ValidateResponse0, FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod ] = cloudi_proplists:take_values(Defaults, Options), true = is_integer(RetryMax) andalso (RetryMax >= 0), ValidateRequestInfo1 = cloudi_args_type: function_optional(ValidateRequestInfo0, 1), ValidateRequest1 = cloudi_args_type: function_optional(ValidateRequest0, 2), ValidateResponseInfo1 = cloudi_args_type: function_optional(ValidateResponseInfo0, 1), ValidateResponse1 = cloudi_args_type: function_optional(ValidateResponse0, 2), true = is_boolean(FailuresSrcDie), true = is_integer(FailuresSrcMaxCount) andalso (FailuresSrcMaxCount > 0), true = (FailuresSrcMaxPeriod =:= infinity) orelse (is_integer(FailuresSrcMaxPeriod) andalso (FailuresSrcMaxPeriod > 0)), #cloudi_queue{ retry_max = RetryMax, validate_request_info = ValidateRequestInfo1, validate_request = ValidateRequest1, validate_response_info = ValidateResponseInfo1, validate_response = ValidateResponse1, failures_source_die = FailuresSrcDie, failures_source_max_count = FailuresSrcMaxCount, failures_source_max_period = FailuresSrcMaxPeriod }. %%------------------------------------------------------------------------- %% @doc %% ===Receive a service request.=== %% Must be called from the `cloudi_service_handle_info/3' callback function. %% @end %%------------------------------------------------------------------------- -spec recv(Dispatcher :: cloudi_service:dispatcher(), #return_async_active{}, State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. recv(Dispatcher, #return_async_active{response_info = ResponseInfo, response = Response, trans_id = TransId}, #cloudi_queue{validate_response_info = ResponseInfoF, validate_response = ResponseF, failures_source_die = FailuresSrcDie, failures_source_max_count = FailuresSrcMaxCount, failures_source_max_period = FailuresSrcMaxPeriod, failures_source = FailuresSrc, requests = Requests} = State) when is_pid(Dispatcher) -> #request{} = dict:fetch(TransId, Requests), NewRequests = dict:erase(TransId, Requests), case validate(ResponseInfoF, ResponseF, ResponseInfo, Response) of true -> {ok, State#cloudi_queue{requests = NewRequests}}; false -> NewFailuresSrc = failure(FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, FailuresSrc), {{error, timeout}, State#cloudi_queue{failures_source = NewFailuresSrc, requests = NewRequests}} end. %%------------------------------------------------------------------------- %% @doc %% ===Send a service request.=== %% @end %%------------------------------------------------------------------------- -spec send(Dispatcher :: cloudi_service:dispatcher(), Name :: cloudi_service:service_name(), Request :: cloudi_service:request(), State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. send(Dispatcher, Name, Request, State) -> send(Dispatcher, Name, <<>>, Request, undefined, undefined, undefined, State). %%------------------------------------------------------------------------- %% @doc %% ===Send a service request.=== %% @end %%------------------------------------------------------------------------- -spec send(Dispatcher :: cloudi_service:dispatcher(), Name :: cloudi_service:service_name(), Request :: cloudi_service:request(), Timeout :: cloudi_service:timeout_milliseconds(), State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. send(Dispatcher, Name, Request, Timeout, State) -> send(Dispatcher, Name, <<>>, Request, Timeout, undefined, undefined, State). %%------------------------------------------------------------------------- %% @doc %% ===Send a service request.=== %% @end %%------------------------------------------------------------------------- -spec send(Dispatcher :: cloudi_service:dispatcher(), Name :: cloudi_service:service_name(), Request :: cloudi_service:request(), Timeout :: cloudi_service:timeout_milliseconds(), PatternPid :: cloudi_service:pattern_pid(), State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. send(Dispatcher, Name, Request, Timeout, PatternPid, State) -> send(Dispatcher, Name, <<>>, Request, Timeout, undefined, PatternPid, State). %%------------------------------------------------------------------------- %% @doc %% ===Send a service request.=== %% @end %%------------------------------------------------------------------------- -spec send(Dispatcher :: cloudi_service:dispatcher(), Name :: cloudi_service:service_name(), RequestInfo :: cloudi_service:request_info(), Request :: cloudi_service:request(), Timeout :: cloudi_service:timeout_milliseconds(), Priority :: cloudi_service:priority(), State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. send(Dispatcher, Name, RequestInfo, Request, Timeout, Priority, State) -> send(Dispatcher, Name, RequestInfo, Request, Timeout, Priority, undefined, State). %%------------------------------------------------------------------------- %% @doc %% ===Send a service request.=== %% @end %%------------------------------------------------------------------------- -spec send(Dispatcher :: cloudi_service:dispatcher(), Name :: cloudi_service:service_name(), RequestInfo :: cloudi_service:request_info(), Request :: cloudi_service:request(), Timeout :: cloudi_service:timeout_milliseconds(), Priority :: cloudi_service:priority(), PatternPid :: cloudi_service:pattern_pid() | undefined, State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. send(Dispatcher, Name, RequestInfo, Request, Timeout, Priority, PatternPid, #cloudi_queue{validate_request_info = RequestInfoF, validate_request = RequestF, failures_source_die = FailuresSrcDie, failures_source_max_count = FailuresSrcMaxCount, failures_source_max_period = FailuresSrcMaxPeriod, failures_source = FailuresSrc, requests = Requests} = State) when is_pid(Dispatcher) -> case validate(RequestInfoF, RequestF, RequestInfo, Request) of true -> case cloudi_service:send_async_active(Dispatcher, Name, RequestInfo, Request, Timeout, Priority, PatternPid) of {ok, TransId} -> RequestState = #request{name = Name, request_info = RequestInfo, request = Request, timeout = Timeout, priority = Priority, pattern_pid = PatternPid}, {ok, State#cloudi_queue{requests = dict:store(TransId, RequestState, Requests)}}; {error, _} = Error -> NewFailuresSrc = failure(FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, FailuresSrc), {Error, State#cloudi_queue{failures_source = NewFailuresSrc}} end; false -> NewFailuresSrc = failure(FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, FailuresSrc), {{error, timeout}, State#cloudi_queue{failures_source = NewFailuresSrc}} end. %%------------------------------------------------------------------------- %% @doc %% ===A service request timeout.=== %% `ok' is returned if a service request is retried. Must be called from the %% `cloudi_service_handle_info/3' callback function. %% @end %%------------------------------------------------------------------------- -spec timeout(Dispatcher :: cloudi_service:dispatcher(), #timeout_async_active{}, State :: state()) -> {ok, NewState :: state()} | {{error, Reason :: cloudi_service:error_reason()}, NewState :: state()}. timeout(Dispatcher, #timeout_async_active{trans_id = TransId}, #cloudi_queue{retry_max = RetryMax, failures_source_die = FailuresSrcDie, failures_source_max_count = FailuresSrcMaxCount, failures_source_max_period = FailuresSrcMaxPeriod, failures_source = FailuresSrc, requests = Requests} = State) when is_pid(Dispatcher) -> case dict:fetch(TransId, Requests) of #request{retry_count = RetryMax} -> NewFailuresSrc = failure(FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, FailuresSrc), {{error, timeout}, State#cloudi_queue{failures_source = NewFailuresSrc, requests = dict:erase(TransId, Requests)}}; #request{name = Name, request_info = RequestInfo, request = Request, timeout = Timeout, priority = Priority, pattern_pid = PatternPid, retry_count = I} = RequestState when I < RetryMax -> case cloudi_service:send_async_active(Dispatcher, Name, RequestInfo, Request, Timeout, Priority, TransId, PatternPid) of {ok, TransId} -> NewRequests = dict:store(TransId, RequestState#request{ retry_count = I + 1}, Requests), {ok, State#cloudi_queue{requests = NewRequests}}; {error, _} = Error -> NewFailuresSrc = failure(FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, FailuresSrc), {Error, State#cloudi_queue{failures_source = NewFailuresSrc, requests = dict:erase(TransId, Requests)}} end end. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ validate_f_return(Value) when is_boolean(Value) -> Value. validate(undefined, undefined, _, _) -> true; validate(undefined, RF, RInfo, R) -> validate_f_return(RF(RInfo, R)); validate(RInfoF, undefined, RInfo, _) -> validate_f_return(RInfoF(RInfo)); validate(RInfoF, RF, RInfo, R) -> validate_f_return(RInfoF(RInfo)) andalso validate_f_return(RF(RInfo, R)). failure(false, _, _, FailureList) -> FailureList; failure(true, MaxCount, MaxPeriod, FailureList) -> failure_check(cloudi_timestamp:seconds(), FailureList, MaxCount, MaxPeriod). failure_store(FailureList, MaxCount) -> if erlang:length(FailureList) == MaxCount -> failure_kill(); true -> ok end, FailureList. failure_check(SecondsNow, FailureList, MaxCount, infinity) -> failure_store([SecondsNow | FailureList], MaxCount); failure_check(SecondsNow, FailureList, MaxCount, MaxPeriod) -> NewFailureList = cloudi_timestamp:seconds_filter(FailureList, SecondsNow, MaxPeriod), failure_store([SecondsNow | NewFailureList], MaxCount). failure_kill() -> erlang:exit(cloudi_queue).