odbc.erl

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

%%--------------------------------------------------------------------------
handle_info({tcp, Socket, BinData}, State = #state{state = connecting, 
						reply_to = From,
						odbc_socket = Socket}) ->
    case binary_to_term(BinData) of
	{ok, AbsolutSupport, RelativeSupport} ->
	    NewState = State#state{absolute_pos = AbsolutSupport,
				   relative_pos = RelativeSupport},
	    gen_server:reply(From, ok), 
	    {noreply, NewState#state{state = connected,
				     reply_to = undefined}};
	Error ->
	    gen_server:reply(From, Error), 
	    {stop, normal, State#state{reply_to = undefined}}
    end;
        

handle_info({tcp, Socket, _},
	    State = #state{state = connected,
 			   odbc_socket = Socket,
 			   reply_to = skip,
 			   pending_request = undefined}) ->
    %% Disregard this message as it is a answer to a query that has timed
    %% out.
    {noreply, State#state{reply_to = undefined}};

handle_info({tcp, Socket, _},
 	    State = #state{state = connected, odbc_socket = Socket,
 			   reply_to = skip}) ->
    
    %% Disregard this message as it is a answer to a query that has timed
    %% out and process the pending request. 
    {{_, Msg, Timeout}, From} = State#state.pending_request,
    handle_msg(Msg, Timeout, State#state{pending_request=undefined,
					 reply_to = From});

handle_info({tcp, Socket, BinData}, State = #state{state = connected,
						   reply_to = From,
						   odbc_socket = Socket}) ->
    %% Send the reply from the database (received by the erlang control 
    %% process from the port program) to the waiting client.
    gen_server:reply(From, BinData),
    {noreply, State#state{reply_to = undefined,
			  num_timeouts = 0}};

handle_info({tcp, Socket, BinData}, State = #state{state = disconnecting,
						   reply_to = From,
						   odbc_socket = Socket}) ->

    %% The connection will always be closed 
    gen_server:reply(From, ok),  
    
    case binary_to_term(BinData) of
 	ok -> 
 	    ok;
 	{error, Reason} ->
 	    error_logger:error_report("ODBC could not end connection "  
 				      "gracefully due to ~p~n", [Reason])
    end,
    
    {stop, normal, State#state{reply_to = undefined}};

handle_info(timeout, 
	    State = #state{state = disconnecting, 
			   reply_to = From}) when From /= undefined ->
    gen_server:reply(From, ok), 
    {stop, {timeout, "Port program is not responding to disconnect, " 
 	    "will be killed"}, State};

handle_info(timeout, 
	    State = #state{state = connecting, 
			   reply_to = From}) when From /= undefined ->
    gen_server:reply(From, timeout),
    {stop, normal, State#state{reply_to = undefined}};

handle_info(timeout, 
	    State = #state{state = connected, 
			   pending_request = undefined,
			   reply_to = From}) when From /= undefined ->
    gen_server:reply(From, timeout),
    {noreply, State#state{reply_to = skip,
			  num_timeouts = State#state.num_timeouts + 1}};

handle_info(timeout, State =
	    #state{state = connected,      
		   pending_request = {{_, {disconnect, _}, _}, 
				      PendingFrom}}) ->
    gen_server:reply(PendingFrom, ok),
    {stop, {timeout, "Port-program busy when trying to disconnect,  "
	    "will be killed"},
     State#state{pending_request = undefined, reply_to = undefined,
		 num_timeouts = State#state.num_timeouts + 1}};

handle_info(timeout, State =
	    #state{state = connected, 
		   pending_request = {_, PendingFrom}}) ->
    gen_server:reply(PendingFrom, timeout),
    %% The state variable reply_to should continue to have the value skip 
    {noreply, State#state{pending_request = undefined,  
 			  num_timeouts = State#state.num_timeouts + 1}};

handle_info({Port, {exit_status, ?EXIT_SUCCESS}},   
	    State = #state{erlang_port = Port, state = disconnecting}) ->
    {noreply, State}; % Ignore as this is perfectly normal in this case 

handle_info({Port, {exit_status, Status}}, 
	    State = #state{erlang_port = Port}) ->
    {stop, {port_exit, ?PORT_EXIT_REASON(Status)}, State};

handle_info({'EXIT', Port, _}, State = #state{erlang_port = Port,
					      state = disconnecting}) ->
    {noreply, State}; % Ignore as this is perfectly normal in this case 

handle_info({'EXIT', Port, Reason}, State = #state{erlang_port = Port}) ->
    {stop, Reason, State};

%%% If the owning process dies there is no reson to go on
handle_info({'DOWN', _Ref, _Type, _Process, normal}, State) ->
    {stop, normal, State#state{reply_to = undefined}};
    
handle_info({'DOWN', _Ref, _Type, _Process, timeout}, State) ->
    {stop, normal, State#state{reply_to = undefined}};
 
handle_info({'DOWN', _Ref, _Type, Process, Reason}, State) ->
    {stop, {stopped, {'EXIT', Process, Reason}}, 
     State#state{reply_to = undefined}};
    
%---------------------------------------------------------------------------
%% Catch all - throws away unknown messages (This could happen by "accident"
%% so we do not want to crash, but we make a log entry as it is an
%% unwanted behaviour.) 
handle_info(Info, State) ->
    error_logger:error_report("ODBC: received unexpected info: ~p~n",
			      [Info]),
    {noreply, State}.

%%--------------------------------------------------------------------------
%% terminate/2 and code_change/3
%%--------------------------------------------------------------------------

terminate({port_exit, _Reason}, State = #state{reply_to = undefined}) ->
    %% Port program crashed
    gen_tcp:close(State#state.odbc_socket),
    gen_tcp:close(State#state.sup_socket),
    ok;

terminate(_Reason,  State = #state{reply_to = undefined}) ->

    catch gen_tcp:send(State#state.sup_socket, 
		       [?SHUTDOWN, ?STR_TERMINATOR]),
    catch gen_tcp:close(State#state.odbc_socket),
    catch gen_tcp:close(State#state.sup_socket),
    catch port_close(State#state.erlang_port),
    ok;

terminate(Reason, State = #state{reply_to = From}) ->
    gen_server:reply(From, {error, connection_closed}),
    terminate(Reason, State#state{reply_to = undefined}).

%---------------------------------------------------------------------------
code_change(_Vsn, State, _Extra) ->
    {ok, State}.


%%%========================================================================
%%% Internal functions
%%%========================================================================

connect(ConnectionReferense, ConnectionStr, Options) ->
    {C_AutoCommitMode, ERL_AutoCommitMode} = 
	connection_config(auto_commit, Options),
    TimeOut = connection_config(timeout, Options),
    {C_TraceDriver, _} = connection_config(trace_driver, Options),
    {C_SrollableCursors, ERL_SrollableCursors} = 
	connection_config(scrollable_cursors, Options),
    {C_TupleRow, _} = 
	connection_config(tuple_row, Options),
    ODBCCmd = 
	[?OPEN_CONNECTION, C_AutoCommitMode, C_TraceDriver, 
	 C_SrollableCursors, C_TupleRow, ConnectionStr],
    
    %% Send request, to open a database connection, to the control process.
    case call(ConnectionReferense, 
	      {connect, ODBCCmd, ERL_AutoCommitMode, ERL_SrollableCursors},
	      TimeOut) of
	ok ->
	    {ok, ConnectionReferense};
	Error ->
	    Error
    end.

%%-------------------------------------------------------------------------
odbc_send(Socket, Msg) -> %% Note currently all allowed messages are lists
    NewMsg = Msg ++ [?STR_TERMINATOR],
    ok = gen_tcp:send(Socket, NewMsg),
    inet:setopts(Socket, [{active, once}]).

%%--------------------------------------------------------------------------
connection_config(Key, Options) ->
    case lists:keysearch(Key, 1, Options) of
	{value,{Key, on}} ->
	    {?ON, on};
	{value,{Key, off}} ->
	    {?OFF, off};
	{value,{Key, Value}} ->
	    Value;
	_ ->
	    connection_default(Key)
    end.

%%--------------------------------------------------------------------------
connection_default(auto_commit) ->
    {?ON, on};

connection_default(timeout) ->
    ?DEFAULT_TIMEOUT;

connection_default(tuple_row) ->
  {?ON, on};

connection_default(trace_driver) ->
    {?OFF, off};

connection_default(scrollable_cursors) ->
    {?ON, on}.

%%-------------------------------------------------------------------------
call(ConnectionReference, Msg, Timeout) ->
    
    Result = (catch gen_server:call(ConnectionReference, 
				    {self(), Msg, Timeout}, infinity)),
    case Result of
	%% Normal case, the result from the port-program has directly 
	%% been forwarded to the client
	Binary when binary(Binary) -> 
	     decode(Binary);
	timeout -> 
	    exit(timeout);
	{'EXIT', _} ->
	    {error, connection_closed};
	%% At some occasions the erlang control process will have an
	%% answer that was not directly received from the port-program.
	Term ->  
	    Term
    end.    

%%-------------------------------------------------------------------------
decode(Binary) ->
    case binary_to_term(Binary) of
	[ResultSet | []] -> 
	    ResultSet;
	param_badarg ->
	    exit({badarg, odbc, param_query, 'Params'}); 
	MultipleResultSets_or_Other ->
	    MultipleResultSets_or_Other
    end.

%%-------------------------------------------------------------------------
fix_params({sql_integer, Values}) ->
    {?USER_INT, [256 | Values]};
fix_params({sql_smallint, Values}) ->
    {?USER_SMALL_INT, [256 | Values]};
fix_params({sql_tinyint, Values}) ->
    {?USER_TINY_INT, [256 | Values]};
fix_params({{sql_decimal, Precision, 0}, 
	    Values}) when Precision >= 0, Precision =< 9 ->
    {?USER_DECIMAL, Precision, 0, [256 | Values]};
fix_params({{sql_decimal, Precision, Scale}, Values}) ->
   {?USER_DECIMAL, Precision, Scale, Values};
fix_params({{sql_numeric, Precision, 0}, 
	    Values}) when Precision >= 0, Precision =< 9 ->
    {?USER_NUMERIC, Precision, 0, [256 | Values]};
fix_params({{sql_numeric, Precision, Scale}, Values}) ->
    {?USER_NUMERIC, Precision, Scale, Values};
fix_params({{sql_char, Max}, Values}) ->
    NewValues =
	case (catch 
	      lists:map(fun(Str) -> Str ++ [?STR_TERMINATOR] end, Values)) of
	    {'EXIT', {badarg, _}} -> 
		exit({badarg, odbc, param_query, 'Params'}); 
	    Result ->
		Result
	end,
    {?USER_CHAR, Max, NewValues};
fix_params({{sql_varchar, Max}, Values}) ->
    NewValues =
	case (catch 
	      lists:map(fun(Str) -> Str ++ [?STR_TERMINATOR] end, Values)) of
	    {'EXIT', {badarg, _}} -> 
		exit({badarg, odbc, param_query, 'Params'}); 
	    Result ->
		Result
	end,
    {?USER_VARCHAR, Max, NewValues};
fix_params({{sql_float, Precision}, Values}) ->
    {?USER_FLOAT, Precision, Values};
fix_params({sql_real, Values}) ->
    {?USER_REAL, Values};
fix_params({sql_double, Values}) ->
    {?USER_DOUBLE, Values};
fix_params({sql_bit, Values}) ->
    {?USER_BOOLEAN, Values}.