odbc.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 911 行 · 第 1/3 页

select(ConnectionReference, {relative, Pos} , N, infinity) 
  when pid(ConnectionReference), integer(Pos), Pos > 0, integer(N), N > 0 ->
    ODBCCmd = [?SELECT, ?SELECT_RELATIVE,
	       integer_to_list(Pos), ";", integer_to_list(N), ";"],
    call(ConnectionReference, {select_cmd, relative, ODBCCmd},
	 infinity);

select(ConnectionReference, {relative, Pos} , N, TimeOut) 
  when pid(ConnectionReference), integer(Pos), Pos >0, integer(N),  N > 0,
  integer(TimeOut), TimeOut > 0 ->
    ODBCCmd = [?SELECT,?SELECT_RELATIVE,
	       integer_to_list(Pos), ";", integer_to_list(N), ";"],
    call(ConnectionReference, {select_cmd, relative, ODBCCmd},
	 TimeOut);

select(ConnectionReference, {absolute, Pos} , N, infinity) 
  when pid(ConnectionReference), integer(Pos), Pos > 0, integer(N), N > 0 ->
    ODBCCmd = [?SELECT, ?SELECT_ABSOLUTE,
	       integer_to_list(Pos), ";", integer_to_list(N), ";"],
    call(ConnectionReference, {select_cmd, absolute, ODBCCmd},
	 infinity);

select(ConnectionReference, {absolute, Pos} , N, TimeOut) 
  when pid(ConnectionReference), integer(Pos), Pos > 0, integer(N),  N > 0, 
  integer(TimeOut), TimeOut > 0  ->
    ODBCCmd = [?SELECT, ?SELECT_ABSOLUTE,
	       integer_to_list(Pos), ";", integer_to_list(N), ";"],
    call(ConnectionReference, {select_cmd, absolute, ODBCCmd},
	 TimeOut).
%%--------------------------------------------------------------------------
%% param_query(ConnectionReference, SQLQuery, Params, <TimeOut>) -> 
%%                             ok | {error, Reason} 
%%                                    
%% Description: Executes a parameterized update/delete/insert-query. 
%%--------------------------------------------------------------------------
param_query(ConnectionReference, SQLQuery, Params) ->
    param_query(ConnectionReference, SQLQuery, Params, ?DEFAULT_TIMEOUT).

param_query(ConnectionReference, SQLQuery, Params, infinity) 
  when pid(ConnectionReference), list(SQLQuery), list(Params) ->
    [{_, Values} | _] = Params,
    NoRows = length(Values),
    NewParams = lists:map(fun fix_params/1, Params),
    ODBCCmd = [?PARAM_QUERY, term_to_binary({SQLQuery ++ [?STR_TERMINATOR],
					     NoRows, NewParams})],
    call(ConnectionReference, {param_query, ODBCCmd}, infinity);

param_query(ConnectionReference, SQLQuery, Params, TimeOut)
  when pid(ConnectionReference), list(SQLQuery), list(Params),
       integer(TimeOut), TimeOut > 0 ->
    [{_, Values} | _] = Params,
    NoRows = length(Values),
    NewParams = lists:map(fun fix_params/1, Params),
    ODBCCmd = [?PARAM_QUERY, term_to_binary({SQLQuery ++ [?STR_TERMINATOR],
					     NoRows, NewParams})],
    call(ConnectionReference, {param_query, ODBCCmd}, TimeOut).

%%--------------------------------------------------------------------------
%% describe_table(ConnectionReference, Table, <TimeOut>) -> {ok, Desc} 
%%
%% Desc - [{ColName, Datatype}]
%% ColName - atom()
%% Datatype - atom()                                    
%% Description: Queries the database to find out the datatypes of the
%%              table <Table>
%%--------------------------------------------------------------------------
describe_table(ConnectionReference, Table) ->
    describe_table(ConnectionReference, Table, ?DEFAULT_TIMEOUT).

describe_table(ConnectionReference, Table, infinity) when 
  pid(ConnectionReference), list(Table) -> 
    ODBCCmd = [?DESCRIBE, "SELECT * FROM " ++ Table],
    call(ConnectionReference, {describe_table, ODBCCmd}, infinity);

describe_table(ConnectionReference, Table, TimeOut) 
  when pid(ConnectionReference),list(Table),integer(TimeOut),TimeOut>0 -> 
    ODBCCmd = [?DESCRIBE, "SELECT * FROM " ++ Table],
    call(ConnectionReference, {describe_table, ODBCCmd}, TimeOut).
%%%=========================================================================
%%% Start/stop
%%%=========================================================================
%%--------------------------------------------------------------------------
%% start_link_sup(Args) -> {ok, Pid} | {error, Reason} 
%%                                    
%% Description: Callback function for the odbc supervisor. It is called 
%%            : when connect/2 calls supervisor:start_child/2 to start an 
%%            : instance of the erlang odbc control process.
%%--------------------------------------------------------------------------
start_link_sup(Args) ->
    gen_server:start_link(?MODULE, Args, []).

%%% Stop functionality is handled by disconnect/1

%%%========================================================================
%%% Callback functions from gen_server
%%%========================================================================

%%-------------------------------------------------------------------------
%% init(Args) -> {ok, State} | {ok, State, Timeout} | {stop, Reason}
%% Description: Initiates the erlang process that manages the connection
%%              and starts the port-program that use the odbc driver
%%		to communicate with the database.
%%-------------------------------------------------------------------------
init(Args) ->
    process_flag(trap_exit, true),
    {value, {client, ClientPid}} = lists:keysearch(client, 1, Args),
    
    erlang:monitor(process, ClientPid),
    
    Inet = case inet:getaddr("localhost", inet6) of
	       {ok, {_,_,_,_}} ->
		   inet;
	       {ok, {0, 0, 0, 0, 0, 16#ffff, _, _}} ->
		   inet;
	       {ok, {_,_,_,_,_,_,_,_}} ->
		   inet6;
	       _ ->
		   inet
	   end,

    {ok, ListenSocketSup} =
	gen_tcp:listen(0, [Inet, binary, {packet, ?LENGTH_INDICATOR_SIZE},
			   {active, false}, {nodelay, true}]),
    {ok, ListenSocketOdbc} =
	gen_tcp:listen(0, [Inet, binary, {packet, ?LENGTH_INDICATOR_SIZE},
			   {active, false}, {nodelay, true}]),

    %% Start the port program (a c program) that utilizes the odbc driver 
    case os:find_executable(?SERVERPROG, ?SERVERDIR) of
	FileName when list(FileName)->
	    Port  = open_port({spawn, FileName},
			      [{packet, ?LENGTH_INDICATOR_SIZE}, binary,
			       exit_status]),
	    State = #state{listen_sockets = 
			   [ListenSocketSup, ListenSocketOdbc],
			   erlang_port = Port, owner = ClientPid},
	    {ok, State};
	false ->
	    {stop, port_program_executable_not_found}
    end.
		    
%%--------------------------------------------------------------------------
%% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State}               |
%%                                      {noreply, State, Timeout}      |
%%                                      {stop, Reason, Reply, State}   |
%%                                      {stop, Reason, Reply, State}     
%% Description: Handle incoming requests. Only requests from the process
%%              that created the connection are allowed in order to preserve
%%              the semantics of result sets.
%% Note: The order of the function clauses is significant.
%%--------------------------------------------------------------------------
handle_call({Client, Msg, Timeout}, From, State = 
	    #state{owner = Client, reply_to = undefined})  ->
    handle_msg(Msg, Timeout, State#state{reply_to = From});

%% The client has caught the timeout and is sending a new request, but
%% we must preserve a synchronous communication with the port. This
%% request will be handled when we have received the answer to the
%% timed out request and thrown it away, if it has not already been
%% timed out itself in which case the request is thrown away.
handle_call(Request = {Client, _, Timeout}, From, 
	    State = #state{owner = Client, reply_to = skip,
			   num_timeouts = N}) when N < ?MAX_SEQ_TIMEOUTS ->
    {noreply, State#state{pending_request = {Request, From}}, Timeout};

%% The client has sent so many sequential requests that has timed out that 
%% there might be something radically wrong causing the ODBC-driver to
%% hang. So we give up and close the connection. 
handle_call({Client, _, _}, From, 
	    State = #state{owner = Client,  
			   num_timeouts = N}) when N >= ?MAX_SEQ_TIMEOUTS ->
    gen_server:reply(From, {error, connection_closed}), 
    {stop, too_many_sequential_timeouts, State#state{reply_to = undefined}};

handle_call(_, _, State) ->
    {reply, {error, process_not_owner_of_odbc_connection}, 
     State#state{reply_to = undefined}}.

%%--------------------------------------------------------------------------
%% Func: handle_msg(Msg, Timeout, State) -> same as handle_call/3.
%% Description: Sends requests to the port-program.
%% Note: The order of the function clauses is significant.
%%--------------------------------------------------------------------------
handle_msg({connect, ODBCCmd, AutoCommitMode, SrollableCursors},
	   Timeout, State) ->

    [ListenSocketSup, ListenSocketOdbc] = State#state.listen_sockets,
    
    %% Inform c-client so it knows where to send answers
    {ok, InetPortSup} = inet:port(ListenSocketSup),
    {ok, InetPortOdbc} = inet:port(ListenSocketOdbc),
    
    port_command(State#state.erlang_port, 
		 [integer_to_list(InetPortSup), ";", 
		  integer_to_list(InetPortOdbc) , ?STR_TERMINATOR]),
    
    NewState = State#state{auto_commit_mode = AutoCommitMode,
			   scrollable_cursors = SrollableCursors},
    
    case gen_tcp:accept(ListenSocketSup, 5000) of
	{ok, SupSocket} ->
	    gen_tcp:close(ListenSocketSup),
	    case gen_tcp:accept(ListenSocketOdbc, 5000) of
		{ok, OdbcSocket} ->
		    gen_tcp:close(ListenSocketOdbc),
		    odbc_send(OdbcSocket, ODBCCmd), 
		    {noreply, NewState#state{odbc_socket = OdbcSocket,
					     sup_socket = SupSocket}, 
		     Timeout};
		{error, Reason} ->
		    {stop, Reason, {error, connection_closed}, NewState}
	    end;
	{error, Reason} ->
	    {stop, Reason, {error, connection_closed}, NewState}
    end;    
    
handle_msg({disconnect, ODBCCmd}, Timeout, State) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State#state{state = disconnecting}, Timeout};

handle_msg({commit, _ODBCCmd}, Timeout, 
	   State = #state{auto_commit_mode = on}) ->
    {reply, {error, not_an_explicit_commit_connection}, 
     State#state{reply_to = undefined}, Timeout};

handle_msg({commit, ODBCCmd}, Timeout, 
	   State = #state{auto_commit_mode = off}) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State, Timeout};

handle_msg({sql_query, ODBCCmd}, Timeout, State) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State#state{result_set = undefined}, Timeout};

handle_msg({param_query, ODBCCmd}, Timeout, State) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State#state{result_set = undefined}, Timeout};

handle_msg({describe_table, ODBCCmd}, Timeout, State) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State#state{result_set = undefined}, Timeout};

handle_msg({select_count, ODBCCmd}, Timeout, State) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State#state{result_set = exists}, Timeout};

handle_msg({select_cmd, absolute, ODBCCmd}, Timeout,
	   State = #state{result_set = exists, absolute_pos = true}) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State, Timeout};

handle_msg({select_cmd, relative, ODBCCmd}, Timeout, 
	   State = #state{result_set = exists, relative_pos = true}) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State, Timeout};

handle_msg({select_cmd, next, ODBCCmd}, Timeout,
	   State = #state{result_set = exists}) ->
    odbc_send(State#state.odbc_socket, ODBCCmd),
    {noreply, State, Timeout};

handle_msg({select_cmd, _Type, _ODBCCmd}, _Timeout,
	   State = #state{result_set = undefined}) ->
    {reply, {error, result_set_does_not_exist}, 
     State#state{reply_to = undefined}};

handle_msg({select_cmd, _Type, _ODBCCmd}, _Timeout, State) ->
    Reply = case State#state.scrollable_cursors of
		on ->
		    {error, driver_does_not_support_function};
		off ->
		    {error, scrollable_cursors_disabled}
	    end,
	    
    {reply, Reply, State#state{reply_to = undefined}};

%---------------------------------------------------------------------------
%% Catch all -  This can oly happen if the application programmer writes 
%% really bad code that violates the API.
handle_msg(Request, _Timeout, State) ->
    {stop, {'API_violation_connection_colsed', Request},
     {error, connection_closed}, State#state{reply_to = undefined}}.

%%--------------------------------------------------------------------------
%% handle_cast(Request, State) -> {noreply, State} | 
%%                                {noreply, State, Timeout} |
%%                                {stop, Reason, State} 
%% Description: Handles cast messages.         
%% Note: The order of the function clauses is significant.
%%-------------------------------------------------------------------------
%% Catch all - This can only happen if the application programmer writes 
%% really bad code that violates the API.
handle_cast(Msg, State) ->
    {stop, {'API_violation_connection_colsed', Msg}, State}.

%%--------------------------------------------------------------------------
%% handle_info(Msg, State) -> {noreply, State} | {noreply, State, Timeout} |
%%			      {stop, Reason, State}
%% Description: Handles timouts, replys from the port-program and EXIT and
%%		down messages.
%% Note: The order of the function clauses is significant.