Skip to content

Commit 95074b9

Browse files
jayneljaynel
committed
Add elysium_connection:one_shot_query/4
1 parent 4e0be75 commit 95074b9

1 file changed

Lines changed: 106 additions & 72 deletions

File tree

‎src/elysium_connection.erl‎

Lines changed: 106 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
start_link/1,
4646
start_link/2,
4747
stop/1,
48+
one_shot_query/4,
4849
get_buffer_strategy_module/1,
4950
with_connection/4,
5051
with_connection/5
@@ -86,6 +87,94 @@ start_link(Config, restart) ->
8687
Max_Retries = elysium_config:checkout_max_retry (Config),
8788
start_channel(Config, Lb_Queue_Name, Max_Retries, -1, []).
8889

90+
-spec stop(connection_id()) -> ok.
91+
%% @doc Stop an existing seestar_session.
92+
stop(Connection_Id)
93+
when is_pid(Connection_Id) ->
94+
seestar_session:stop(Connection_Id).
95+
96+
-spec one_shot_query(config_type(), cassandra_node(), Query::string(), ssestar:consistency())
97+
-> {ok, seestar_result:result()} | {error, seestar_error:error()}.
98+
%% @doc Connect, execute raw CQL and close a connection to Cassandra.
99+
one_shot_query(Config, {Host, Port} = _Node, Query, Consistency)
100+
when is_list(Host), is_integer(Port), Port > 0 ->
101+
case get_one_shot_connection(Config, Host, Port) of
102+
{ok, Connection_Id} ->
103+
try seestar_session:perform(Connection_Id, Query, Consistency)
104+
after stop(Connection_Id)
105+
end
106+
end.
107+
108+
-spec get_buffer_strategy_module(config_type()) -> {buffering(), buffering_strategy_module()}.
109+
%% @doc Get the module corresponding to the configured buffering strategy.
110+
get_buffer_strategy_module(Config) ->
111+
true = elysium_config:is_valid_config(Config),
112+
Buffering_Strategy = elysium_config:connection_buffering_strategy(Config),
113+
BS_Module = case Buffering_Strategy of
114+
none -> elysium_nobs;
115+
parallel -> elysium_bs_parallel;
116+
serial -> elysium_bs_serial
117+
end,
118+
{Buffering_Strategy, BS_Module}.
119+
120+
-spec with_connection(config_type(), fun((pid(), Args, Consist) -> Result), Args, Consistency)
121+
-> {error, no_db_connections}
122+
| Result when Args :: [any()],
123+
Consist :: seestar:consistency(),
124+
Consistency :: seestar:consistency(),
125+
Result :: any().
126+
%% @doc
127+
%% Obtain an active seestar_session and use it solely
128+
%% for the duration required to execute a fun which
129+
%% requires access to Cassandra.
130+
%% @end
131+
with_connection(Config, Session_Fun, Args, Consistency)
132+
when is_function(Session_Fun, 3), is_list(Args) ->
133+
{Buffering_Strategy, BS_Module} = get_buffer_strategy_module(Config),
134+
case elysium_buffering_strategy:checkout_connection(Config, BS_Module) of
135+
none_available ->
136+
buffer_bare_fun_call(Config, Session_Fun, Args, Consistency, Buffering_Strategy);
137+
{Node, Sid} when is_pid(Sid) ->
138+
case is_process_alive(Sid) of
139+
false -> with_connection(Config, Session_Fun, Args, Consistency);
140+
true -> Reply_Timeout = elysium_config:request_reply_timeout(Config),
141+
Query_Request = {bare_fun, Config, Session_Fun, Args, Consistency},
142+
Reply = elysium_buffering_strategy:handle_pending_request
143+
(Config, BS_Module, 0, Reply_Timeout, Node, Sid, Query_Request),
144+
handle_bare_fun_reply(Buffering_Strategy, Reply, ?MODULE, with_connection, Args)
145+
end
146+
end.
147+
148+
-spec with_connection(config_type(), module(), Fun::atom(), Args::[any()], seestar:consistency())
149+
-> {error, no_db_connections} | any().
150+
%% @doc
151+
%% Obtain an active seestar_session and use it solely
152+
%% for the duration required to execute Mod:Fun
153+
%% which requires access to Cassandra.
154+
%% @end
155+
with_connection(Config, Mod, Fun, Args, Consistency)
156+
when is_atom(Mod), is_atom(Fun), is_list(Args) ->
157+
true = erlang:function_exported(Mod, Fun, 3),
158+
{Buffering_Strategy, BS_Module} = get_buffer_strategy_module(Config),
159+
case elysium_buffering_strategy:checkout_connection(Config, BS_Module) of
160+
none_available ->
161+
buffer_mod_fun_call(Config, Mod, Fun, Args, Consistency, Buffering_Strategy);
162+
{Node, Sid} when is_pid(Sid) ->
163+
case is_process_alive(Sid) of
164+
false -> with_connection(Config, Mod, Fun, Args, Consistency);
165+
true -> Reply_Timeout = elysium_config:request_reply_timeout(Config),
166+
Query_Request = {mod_fun, Config, Mod, Fun, Args, Consistency},
167+
Reply = elysium_buffering_strategy:handle_pending_request
168+
(Config, BS_Module, 0, Reply_Timeout, Node, Sid, Query_Request),
169+
handle_mod_fun_reply(Buffering_Strategy, Reply, Mod, Fun, Args)
170+
end
171+
end.
172+
173+
174+
%%%-----------------------------------------------------------------------
175+
%%% Internal connection support functions
176+
%%%-----------------------------------------------------------------------
177+
89178
start_channel(_Config, _Lb_Queue_Name, Max_Retries, Times_Tried, Attempted_Connections)
90179
when Times_Tried >= Max_Retries ->
91180
{error, {cassandra_not_available, Attempted_Connections}};
@@ -130,61 +219,32 @@ try_connect(Config, Lb_Queue_Name, Max_Retries, Times_Tried, Attempted_Connectio
130219
start_channel(Config, Lb_Queue_Name, Max_Retries, Times_Tried+1,
131220
[Node_Failure | Attempted_Connections])
132221

133-
catch Error:Class ->
134-
Node_Failure = {Node, {Error, Class}},
222+
catch Err:Class ->
223+
Node_Failure = {Node, {Err, Class}},
135224
start_channel(Config, Lb_Queue_Name, Max_Retries, Times_Tried+1,
136225
[Node_Failure | Attempted_Connections])
137226

138227
%% Ensure that we get the Node checked back in.
139228
after _ = ets_buffer:write_dedicated(Lb_Queue_Name, Node)
140229
end.
141230

142-
-spec stop(pid()) -> ok.
143-
%% @doc Stop an existing seestar_session.
144-
stop(Session_Id)
145-
when is_pid(Session_Id) ->
146-
seestar_session:stop(Session_Id).
147-
148-
-spec get_buffer_strategy_module(config_type()) -> {buffering(), module()}.
149-
%% @doc Get the module corresponding to the configured buffering strategy.
150-
get_buffer_strategy_module(Config) ->
151-
true = elysium_config:is_valid_config(Config),
152-
Buffering_Strategy = elysium_config:connection_buffering_strategy(Config),
153-
BS_Module = case Buffering_Strategy of
154-
none -> elysium_nobs;
155-
parallel -> elysium_bs_parallel;
156-
serial -> elysium_bs_serial
157-
end,
158-
{Buffering_Strategy, BS_Module}.
231+
-spec get_one_shot_connection(config_type(), string(), pos_integer()) -> {ok, connection_id()} | any().
232+
get_one_shot_connection(Config, Host, Port) ->
233+
Connect_Timeout = elysium_config:connect_timeout (Config),
234+
try seestar_session:start_link(Host, Port,
235+
[
236+
%% {send_timeout, Send_Timeout}
237+
],
238+
[{connect_timeout, Connect_Timeout}]) of
239+
{ok, Connection_Id} = Connection when is_pid(Connection_Id) -> Connection;
240+
Error -> Error
241+
catch Err:Class -> {Err, Class}
242+
end.
159243

160244

161-
-spec with_connection(config_type(), fun((pid(), Args, Consist) -> Result), Args, Consistency)
162-
-> {error, no_db_connections}
163-
| Result when Args :: [any()],
164-
Consist :: seestar:consistency(),
165-
Consistency :: seestar:consistency(),
166-
Result :: any().
167-
%% @doc
168-
%% Obtain an active seestar_session and use it solely
169-
%% for the duration required to execute a fun which
170-
%% requires access to Cassandra.
171-
%% @end
172-
with_connection(Config, Session_Fun, Args, Consistency)
173-
when is_function(Session_Fun, 3), is_list(Args) ->
174-
{Buffering_Strategy, BS_Module} = get_buffer_strategy_module(Config),
175-
case elysium_buffering_strategy:checkout_connection(Config, BS_Module) of
176-
none_available ->
177-
buffer_bare_fun_call(Config, Session_Fun, Args, Consistency, Buffering_Strategy);
178-
{Node, Sid} when is_pid(Sid) ->
179-
case is_process_alive(Sid) of
180-
false -> with_connection(Config, Session_Fun, Args, Consistency);
181-
true -> Reply_Timeout = elysium_config:request_reply_timeout(Config),
182-
Query_Request = {bare_fun, Config, Session_Fun, Args, Consistency},
183-
Reply = elysium_buffering_strategy:handle_pending_request
184-
(Config, BS_Module, 0, Reply_Timeout, Node, Sid, Query_Request),
185-
handle_bare_fun_reply(Buffering_Strategy, Reply, ?MODULE, with_connection, Args)
186-
end
187-
end.
245+
%%%-----------------------------------------------------------------------
246+
%%% Internal with_connection support functions
247+
%%%-----------------------------------------------------------------------
188248

189249
buffer_bare_fun_call(_Config, _Session_Fun, _Args, _Consistency, none) -> {error, no_db_connections};
190250
buffer_bare_fun_call( Config, Session_Fun, Args, Consistency, Type) ->
@@ -212,32 +272,6 @@ handle_bare_fun_reply(serial, {Err_Type, _Err_Data} = Error, Mod, Fun, Args)
212272
handle_bare_fun_reply(_Buffering_Strategy, Reply, _Mod, _Fun, _Args) ->
213273
Reply.
214274

215-
216-
-spec with_connection(config_type(), module(), Fun::atom(), Args::[any()], seestar:consistency())
217-
-> {error, no_db_connections} | any().
218-
%% @doc
219-
%% Obtain an active seestar_session and use it solely
220-
%% for the duration required to execute Mod:Fun
221-
%% which requires access to Cassandra.
222-
%% @end
223-
with_connection(Config, Mod, Fun, Args, Consistency)
224-
when is_atom(Mod), is_atom(Fun), is_list(Args) ->
225-
true = erlang:function_exported(Mod, Fun, 3),
226-
{Buffering_Strategy, BS_Module} = get_buffer_strategy_module(Config),
227-
case elysium_buffering_strategy:checkout_connection(Config, BS_Module) of
228-
none_available ->
229-
buffer_mod_fun_call(Config, Mod, Fun, Args, Consistency, Buffering_Strategy);
230-
{Node, Sid} when is_pid(Sid) ->
231-
case is_process_alive(Sid) of
232-
false -> with_connection(Config, Mod, Fun, Args, Consistency);
233-
true -> Reply_Timeout = elysium_config:request_reply_timeout(Config),
234-
Query_Request = {mod_fun, Config, Mod, Fun, Args, Consistency},
235-
Reply = elysium_buffering_strategy:handle_pending_request
236-
(Config, BS_Module, 0, Reply_Timeout, Node, Sid, Query_Request),
237-
handle_mod_fun_reply(Buffering_Strategy, Reply, Mod, Fun, Args)
238-
end
239-
end.
240-
241275
buffer_mod_fun_call(_Config, _Mod, _Fun, _Args, _Consistency, none) -> {error, no_db_connections};
242276
buffer_mod_fun_call( Config, Mod, Fun, Args, Consistency, parallel) ->
243277
Reply = elysium_buffering_strategy:pend_request(Config, {mod_fun, Config, Mod, Fun, Args, Consistency}),

0 commit comments

Comments
 (0)