diff options
Diffstat (limited to 'lib/dialyzer/test/r9c_SUITE_data/src/inets/httpc_manager.erl')
| -rw-r--r-- | lib/dialyzer/test/r9c_SUITE_data/src/inets/httpc_manager.erl | 542 |
1 files changed, 542 insertions, 0 deletions
diff --git a/lib/dialyzer/test/r9c_SUITE_data/src/inets/httpc_manager.erl b/lib/dialyzer/test/r9c_SUITE_data/src/inets/httpc_manager.erl new file mode 100644 index 0000000000..29659ce1ce --- /dev/null +++ b/lib/dialyzer/test/r9c_SUITE_data/src/inets/httpc_manager.erl @@ -0,0 +1,542 @@ +%% ``The contents of this file are subject to the Erlang Public License, +%% Version 1.1, (the "License"); you may not use this file except in +%% compliance with the License. You should have received a copy of the +%% Erlang Public License along with this software. If not, it can be +%% retrieved via the world wide web at http://www.erlang.org/. +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and limitations +%% under the License. +%% +%% The Initial Developer of the Original Code is Mobile Arts AB +%% Portions created by Mobile Arts are Copyright 2002, Mobile Arts AB +%% All Rights Reserved.'' +%% +%% +%% Created : 18 Dec 2001 by Johan Blom <[email protected]> +%% + +-module(httpc_manager). + +-behaviour(gen_server). + +-include("http.hrl"). + +-define(HMACALL, ?MODULE). +-define(HMANAME, ?MODULE). + +%%-------------------------------------------------------------------- +%% External exports +-export([start_link/0,start/0, + request/1,cancel_request/1, + next_request/2, + register_socket/3, + abort_session/3,close_session/2,close_session/3 + ]). + +%% Debugging only +-export([status/0]). + +%% gen_server callbacks +-export([init/1,handle_call/3,handle_cast/2,handle_info/2,terminate/2, + code_change/3]). + +%%% address_db - ets() Contains mappings from a tuple {Host,Port} to a tuple +%%% {LastSID,OpenSessions,ets()} where +%%% LastSid is the last allocated session id, +%%% OpenSessions is the number of currently open sessions and +%%% ets() contains mappings from Session Id to #session{}. +%%% +%%% Note: +%%% - Only persistent connections are stored in address_db +%%% - When automatically redirecting, multiple requests are performed. +-record(state,{ + address_db, % ets() + reqid % int() Next Request id to use (identifies request). + }). + +%%==================================================================== +%% External functions +%%==================================================================== +%%-------------------------------------------------------------------- +%% Function: start_link/0 +%% Description: Starts the server +%%-------------------------------------------------------------------- +start() -> + ensure_started(). + +start_link() -> + gen_server:start_link({local,?HMACALL}, ?HMANAME, [], []). + + +%% Find available session process and store in address_db. If no +%% available, start new handler process. +request(Req) -> + ensure_started(), + ClientClose=http_lib:connection_close(Req#request.headers), + gen_server:call(?HMACALL,{request,ClientClose,Req},infinity). + +cancel_request(ReqId) -> + gen_server:call(?HMACALL,{cancel_request,ReqId},infinity). + + +%%% Close Session +close_session(Addr,Sid) -> + gen_server:call(?HMACALL,{close_session,Addr,Sid},infinity). +close_session(Req,Addr,Sid) -> + gen_server:call(?HMACALL,{close_session,Req,Addr,Sid},infinity). + +abort_session(Addr,Sid,Msg) -> + gen_server:call(?HMACALL,{abort_session,Addr,Sid,Msg},infinity). + + +%%% Pick next in request que +next_request(Addr,Sid) -> + gen_server:call(?HMACALL,{next_request,Addr,Sid},infinity). + +%%% Session handler has succeded to set up a new session, now register +%%% the socket +register_socket(Addr,Sid,Socket) -> + gen_server:cast(?HMACALL,{register_socket,Addr,Sid,Socket}). + + +%%% Debugging +status() -> + gen_server:cast(?HMACALL,status). + + +%%-------------------------------------------------------------------- +%% Function: init/1 +%% Description: Initiates the server +%% Returns: {ok, State} | +%% {ok, State, Timeout} | +%% ignore | +%% {stop, Reason} +%%-------------------------------------------------------------------- +init([]) -> + process_flag(trap_exit, true), + {ok,#state{address_db=ets:new(address_db,[private]), + reqid=0}}. + + +%%-------------------------------------------------------------------- +%% Function: handle_call/3 +%% Description: Handling call messages +%% Returns: {reply, Reply, State} | +%% {reply, Reply, State, Timeout} | +%% {noreply, State} | +%% {noreply, State, Timeout} | +%% {stop, Reason, Reply, State} | (terminate/2 is called) +%% {stop, Reason, State} (terminate/2 is called) +%%-------------------------------------------------------------------- +%%% Note: +%%% - We may have multiple non-persistent connections, each will be handled in +%%% separate processes, thus don't add such connections to address_db +handle_call({request,false,Req},_From,State) -> + case ets:lookup(State#state.address_db,Req#request.address) of + [] -> + STab=ets:new(session_db,[private,{keypos,2},set]), + case persistent_new_session_request(0,Req,STab,State) of + {Reply,LastSid,State2} -> + ets:insert(State2#state.address_db, + {Req#request.address,{LastSid,1,STab}}), + {reply,Reply,State2}; + {ErrorReply,State2} -> + {reply,ErrorReply,State2} + end; + [{_,{LastSid,OpenS,STab}}] -> + case lookup_session_entry(STab) of + {ok,Session} -> + old_session_request(Session,Req,STab,State); + need_new_session when OpenS<(Req#request.settings)#client_settings.max_sessions -> + case persistent_new_session_request(LastSid,Req, + STab,State) of + {Reply,LastSid2,State2} -> + ets:insert(State2#state.address_db, + {Req#request.address, + {LastSid2,OpenS+1,STab}}), + {reply,Reply,State2}; + {ErrorReply,State2} -> + {reply,ErrorReply,State2} + end; + need_new_session -> + {reply,{error,too_many_sessions},State} + end + end; +handle_call({request,true,Req},_From,State) -> + {Reply,State2}=not_persistent_new_session_request(Req,State), + {reply,Reply,State2}; +handle_call({cancel_request,true,_ReqId},_From,State) -> +%% FIXME Should be possible to scan through all requests made, but perhaps +%% better to give some more hints (such as Addr etc) + Reply=ok, + {reply,Reply,State}; +handle_call({next_request,Addr,Sid},_From,State) -> + case ets:lookup(State#state.address_db,Addr) of + [] -> + {reply,{error,no_connection},State}; + [{_,{_,_,STab}}] -> + case ets:lookup(STab,Sid) of + [] -> + {reply,{error,session_not_registered},State}; + [S=#session{pipeline=[],quelength=QueLen}] -> + if + QueLen==1 -> + ets:insert(STab,S#session{quelength=0}); + true -> + ok + end, + {reply,no_more_requests,State}; + [S=#session{pipeline=Que}] -> + [Req|RevQue]=lists:reverse(Que), + ets:insert(STab,S#session{pipeline=lists:reverse(RevQue), + quelength=S#session.quelength-1}), + {reply,Req,State} + end + end; +handle_call({close_session,Addr,Sid},_From,State) -> + case ets:lookup(State#state.address_db,Addr) of + [] -> + {reply,{error,no_connection},State}; + [{_,{LastSid,OpenS,STab}}] -> + case ets:lookup(STab,Sid) of + [#session{pipeline=Que}] -> + R=handle_close_session(lists:reverse(Que),STab,Sid,State), + ets:insert(State#state.address_db, + {Addr,{LastSid,OpenS-1,STab}}), + {reply,R,State}; + [] -> + {reply,{error,session_not_registered},State} + end + end; +handle_call({close_session,Req,Addr,Sid},_From,State) -> + case ets:lookup(State#state.address_db,Addr) of + [] -> + {reply,{error,no_connection},State}; + [{_,{LastSid,OpenS,STab}}] -> + case ets:lookup(STab,Sid) of + [#session{pipeline=Que}] -> + R=handle_close_session([Req|lists:reverse(Que)], + STab,Sid,State), + ets:insert(State#state.address_db, + {Addr,{LastSid,OpenS-1,STab}}), + {reply,R,State}; + [] -> + {reply,{error,session_not_registered},State} + end + end; +handle_call({abort_session,Addr,Sid,Msg},_From,State) -> + case ets:lookup(State#state.address_db,Addr) of + [] -> + {reply,{error,no_connection},State}; + [{_,{LastSid,OpenS,STab}}] -> + case ets:lookup(STab,Sid) of + [#session{pipeline=Que}] -> + R=abort_request_que(Que,{error,Msg}), + ets:delete(STab,Sid), + ets:insert(State#state.address_db, + {Addr,{LastSid,OpenS-1,STab}}), + {reply,R,State}; + [] -> + {reply,{error,session_not_registered},State} + end + end. + + +%%-------------------------------------------------------------------- +%% Function: handle_cast/2 +%% Description: Handling cast messages +%% Returns: {noreply, State} | +%% {noreply, State, Timeout} | +%% {stop, Reason, State} (terminate/2 is called) +%%-------------------------------------------------------------------- +handle_cast(status, State) -> + io:format("Status:~n"), + print_all(lists:sort(ets:tab2list(State#state.address_db))), + {noreply, State}; +handle_cast({register_socket,Addr,Sid,Socket},State) -> + case ets:lookup(State#state.address_db,Addr) of + [] -> + {noreply,State}; + [{_,{_,_,STab}}] -> + case ets:lookup(STab,Sid) of + [Session] -> + ets:insert(STab,Session#session{socket=Socket}), + {noreply,State}; + [] -> + {noreply,State} + end + end. + +print_all([]) -> + ok; +print_all([{Addr,{LastSid,OpenSessions,STab}}|Rest]) -> + io:format(" Address:~p LastSid=~p OpenSessions=~p~n",[Addr,LastSid,OpenSessions]), + SortedList=lists:sort(fun(A,B) -> + if + A#session.id<B#session.id -> + true; + true -> + false + end + end,ets:tab2list(STab)), + print_all2(SortedList), + print_all(Rest). + +print_all2([]) -> + ok; +print_all2([Session|Rest]) -> + io:format(" Session:~p~n",[Session#session.id]), + io:format(" Client close:~p~n",[Session#session.clientclose]), + io:format(" Socket:~p~n",[Session#session.socket]), + io:format(" Pipe: length=~p Que=~p~n",[Session#session.quelength,Session#session.pipeline]), + print_all2(Rest). + +%%-------------------------------------------------------------------- +%% Function: handle_info/2 +%% Description: Handling all non call/cast messages +%% Returns: {noreply, State} | +%% {noreply, State, Timeout} | +%% {stop, Reason, State} (terminate/2 is called) +%%-------------------------------------------------------------------- +handle_info({'EXIT',_Pid,normal}, State) -> + {noreply, State}; +handle_info(Info, State) -> + io:format("ERROR httpc_manager:handle_info ~p~n",[Info]), + {noreply, State}. + +%%-------------------------------------------------------------------- +%% Function: terminate/2 +%% Description: Shutdown the server +%% Returns: any (ignored by gen_server) +%%-------------------------------------------------------------------- +terminate(_Reason, State) -> + ets:delete(State#state.address_db). + +%%-------------------------------------------------------------------- +%% Func: code_change/3 +%% Purpose: Convert process state when code is changed +%% Returns: {ok, NewState} +%%-------------------------------------------------------------------- +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%-------------------------------------------------------------------- +%%% Internal functions +%%-------------------------------------------------------------------- + +%%% From RFC 2616, Section 8.1.4 +%%% A client, server, or proxy MAY close the transport connection at any +%%% time. For example, a client might have started to send a new request +%%% at the same time that the server has decided to close the "idle" +%%% connection. From the server's point of view, the connection is being +%%% closed while it was idle, but from the client's point of view, a +%%% request is in progress. +%%% +%%% This means that clients, servers, and proxies MUST be able to recover +%%% from asynchronous close events. Client software SHOULD reopen the +%%% transport connection and retransmit the aborted sequence of requests +%%% without user interaction so long as the request sequence is +%%% idempotent (see section 9.1.2). Non-idempotent methods or sequences +%%% +%%% FIXME +%%% Note: +%%% - If this happen (server close because of idle) there can't be any requests +%%% in the que. +%%% - This is the main function for closing of sessions +handle_close_session([],STab,Sid,_State) -> + ets:delete(STab,Sid); +handle_close_session(Que,STab,Sid,_State) -> + ets:delete(STab,Sid), + abort_request_que(Que,{error,aborted_request}). + + +%%% From RFC 2616, Section 8.1.2.2 +%%% Clients which assume persistent connections and pipeline immediately +%%% after connection establishment SHOULD be prepared to retry their +%%% connection if the first pipelined attempt fails. If a client does +%%% such a retry, it MUST NOT pipeline before it knows the connection is +%%% persistent. Clients MUST also be prepared to resend their requests if +%%% the server closes the connection before sending all of the +%%% corresponding responses. +%%% FIXME! I'm currently not checking if tis is the first attempt on the session +%%% FIXME! Pipeline size must be dynamically variable (e.g. 0 if resend, 2 else) +%%% The que contains requests that have been sent ok previously, but the session +%%% was closed prematurely when reading the response. +%%% Try setup a new session and resend these requests. +%%% Note: +%%% - This MUST be a persistent session +% handle_closed_pipelined_session_que([],_State) -> +% ok; +% handle_closed_pipelined_session_que(_Que,_State) -> +% ok. + + +%%% From RFC 2616, Section 8.2.4 +%%% If an HTTP/1.1 client sends a request which includes a request body, +%%% but which does not include an Expect request-header field with the +%%% "100-continue" expectation, and if the client is not directly +%%% connected to an HTTP/1.1 origin server, and if the client sees the +%%% connection close before receiving any status from the server, the +%%% client SHOULD retry the request. If the client does retry this +%%% request, it MAY use the following "binary exponential backoff" +%%% algorithm to be assured of obtaining a reliable response: +%%% ... +%%% FIXME! I'm currently not checking if a "Expect: 100-continue" has been sent. +% handle_remotely_closed_session_que([],_State) -> +% ok; +% handle_remotely_closed_session_que(_Que,_State) -> +% % resend_que(Que,Socket), +% ok. + +%%% Resend all requests in the request que +% resend_que([],_) -> +% ok; +% resend_que([Req|Que],Socket) -> +% case catch httpc_handler:http_request(Req,Socket) of +% ok -> +% resend_que(Que,Socket); +% {error,Reason} -> +% {error,Reason} +% end. + + +%%% From RFC 2616, +%%% Section 8.1.2.2: +%%% Clients SHOULD NOT pipeline requests using non-idempotent methods or +%%% non-idempotent sequences of methods (see section 9.1.2). Otherwise, a +%%% premature termination of the transport connection could lead to +%%% indeterminate results. A client wishing to send a non-idempotent +%%% request SHOULD wait to send that request until it has received the +%%% response status for the previous request. +%%% Section 9.1.2: +%%% Methods can also have the property of "idempotence" in that (aside +%%% from error or expiration issues) the side-effects of N > 0 identical +%%% requests is the same as for a single request. The methods GET, HEAD, +%%% PUT and DELETE share this property. Also, the methods OPTIONS and +%%% TRACE SHOULD NOT have side effects, and so are inherently idempotent. +%%% +%%% Note that POST and CONNECT are idempotent methods. +%%% +%%% Tries to find an open, free session i STab. Such a session has quelength +%%% less than ?MAX_PIPELINE_LENGTH +%%% Don't care about non-standard, user defined methods. +%%% +%%% Returns {ok,Session} or need_new_session where +%%% Session is the session that may be used +lookup_session_entry(STab) -> + MS=[{#session{quelength='$1',max_quelength='$2', + id='_',clientclose='_',socket='$3',scheme='_',pipeline='_'}, + [{'<','$1','$2'},{is_port,'$3'}], + ['$_']}], + case ets:select(STab,MS) of + [] -> + need_new_session; + SessionList -> % Now check if any of these has an empty pipeline. + case lists:keysearch(0,2,SessionList) of + {value,Session} -> + {ok,Session}; + false -> + {ok,hd(SessionList)} + end + end. + + +%%% Returns a tuple {Reply,State} where +%%% Reply is the response sent back to the application +%%% +%%% Note: +%%% - An {error,einval} from a send should sometimes rather be {error,closed} +%%% - Don't close the session from here, let httpc_handler take care of that. +%old_session_request(Session,Req,STab,State) +% when (Req#request.settings)#client_settings.max_quelength==0 -> +% Session1=Session#session{pipeline=[Req]}, +% ets:insert(STab,Session1), +% {reply,{ok,ReqId},State#state{reqid=ReqId+1}}; +old_session_request(Session,Req,STab,State) -> + ReqId=State#state.reqid, + Req1=Req#request{id=ReqId}, + case catch httpc_handler:http_request(Req1,Session#session.socket) of + ok -> + Session1=Session#session{pipeline=[Req1|Session#session.pipeline], + quelength=Session#session.quelength+1}, + ets:insert(STab,Session1), + {reply,{ok,ReqId},State#state{reqid=ReqId+1}}; + {error,Reason} -> + ets:insert(STab,Session#session{socket=undefined}), +% http_lib:close(Session#session.sockettype,Session#session.socket), + {reply,{error,Reason},State#state{reqid=ReqId+1}} + end. + +%%% Returns atuple {Reply,Sid,State} where +%%% Reply is the response sent back to the application, and +%%% Sid is the last used Session Id +persistent_new_session_request(Sid,Req,STab,State) -> + ReqId=State#state.reqid, + case setup_new_session(Req#request{id=ReqId},false,Sid) of + {error,Reason} -> + {{error,Reason},State#state{reqid=ReqId+1}}; + {NewSid,Session} -> + ets:insert(STab,Session), + {{ok,ReqId},NewSid,State#state{reqid=ReqId+1}} + end. + +%%% Returns a tuple {Reply,State} where +%%% Reply is the response sent back to the application +not_persistent_new_session_request(Req,State) -> + ReqId=State#state.reqid, + case setup_new_session(Req#request{id=ReqId},true,undefined) of + {error,Reason} -> + {{error,Reason},State#state{reqid=ReqId+1}}; + ok -> + {{ok,ReqId},State#state{reqid=ReqId+1}} + end. + +%%% As there are no sessions available, setup a new session and send the request +%%% on it. +setup_new_session(Req,ClientClose,Sid) -> + S=#session{id=Sid,clientclose=ClientClose, + scheme=Req#request.scheme, + max_quelength=(Req#request.settings)#client_settings.max_quelength}, + spawn_link(httpc_handler,init_connection,[Req,S]), + case ClientClose of + false -> + {Sid+1,S}; + true -> + ok + end. + + +%%% ---------------------------------------------------------------------------- +%%% Abort all requests in the request que. +abort_request_que([],_Msg) -> + ok; +abort_request_que([#request{from=From,ref=Ref,id=Id}|Que],Msg) -> + gen_server:cast(From,{Ref,Id,Msg}), + abort_request_que(Que,Msg); +abort_request_que(#request{from=From,ref=Ref,id=Id},Msg) -> + gen_server:cast(From,{Ref,Id,Msg}). + + +%%% -------------------------------- +% C={httpc_manager,{?MODULE,start_link,[]},permanent,1000, +% worker,[?MODULE]}, +% supervisor:start_child(inets_sup, C), +ensure_started() -> + case whereis(?HMANAME) of + undefined -> + start_link(); + _ -> + ok + end. + + +%%% ============================================================================ +%%% This is deprecated code, to be removed + +% format_time() -> +% {_,_,MicroSecs}=TS=now(), +% {{Y,Mon,D},{H,M,S}}=calendar:now_to_universal_time(TS), +% lists:flatten(io_lib:format("~4.4.0w-~2.2.0w-~2.2.0w,~2.2.0w:~2.2.0w:~6.3.0f", +% [Y,Mon,D,H,M,S+(MicroSecs/1000000)])). |