-module(gmysql@pool). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -export([disconnect/1, restart_connection/2, restart_all_connections/1, new_connection/1, checkout_connection/2, checkin_connection/2, with_connection/3, connect/3]). -export_type([pool/0, message/0, state/0, slot/0, with_connection_error/0]). -opaque pool() :: {pool, gleam@erlang@process:subject(message())}. -opaque message() :: {update_state, fun((state()) -> state())} | {initialize, gleam@erlang@process:subject(message())} | shutdown | {restart, gmysql:connection()} | restart_all | start_new | {checkout, gleam@erlang@process:subject({ok, gmysql:connection()} | {error, nil})} | {checkin, gmysql:connection()} | {tick, gleam@erlang@process:subject(message())}. -type state() :: {state, list(slot()), gmysql:config(), gleam@otp@intensity_tracker:intensity_tracker()}. -type slot() :: {slot, gleam@option:option(gmysql:connection()), boolean()}. -type with_connection_error() :: could_not_checkout. -spec disconnect(pool()) -> nil. disconnect(Pool) -> gleam@otp@actor:send(erlang:element(2, Pool), shutdown). -spec restart_connection(pool(), gmysql:connection()) -> nil. restart_connection(Pool, Connection) -> gleam@otp@actor:send(erlang:element(2, Pool), {restart, Connection}). -spec restart_all_connections(pool()) -> nil. restart_all_connections(Pool) -> gleam@otp@actor:send(erlang:element(2, Pool), restart_all). -spec new_connection(pool()) -> nil. new_connection(Pool) -> gleam@otp@actor:send(erlang:element(2, Pool), start_new). -spec checkout_connection(pool(), integer()) -> {ok, gmysql:connection()} | {error, nil}. checkout_connection(Pool, Timeout) -> gleam@otp@actor:call( erlang:element(2, Pool), fun(Field@0) -> {checkout, Field@0} end, Timeout ). -spec checkin_connection(pool(), gmysql:connection()) -> nil. checkin_connection(Pool, Connection) -> gleam@otp@actor:send(erlang:element(2, Pool), {checkin, Connection}). -spec with_connection_loop( pool(), integer(), integer(), fun((gmysql:connection()) -> HJM) ) -> {ok, HJM} | {error, with_connection_error()}. with_connection_loop(Pool, Until, Timeout, Function) -> gleam@bool:guard( os:system_time(millisecond) > Until, {error, could_not_checkout}, fun() -> case checkout_connection(Pool, Timeout div 3) of {ok, Connection} -> Result = Function(Connection), checkin_connection(Pool, Connection), {ok, Result}; {error, _} -> gleam_erlang_ffi:sleep(Timeout div 6), with_connection_loop(Pool, Until, Timeout, Function) end end ). -spec with_connection(pool(), integer(), fun((gmysql:connection()) -> HJJ)) -> {ok, HJJ} | {error, with_connection_error()}. with_connection(Pool, Checkout_timeout, Function) -> Until = os:system_time(millisecond) + Checkout_timeout, with_connection_loop(Pool, Until, Checkout_timeout, Function). -spec handle_update_state(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_update_state(Message, State) -> {update_state, Transform} = case Message of {update_state, _} -> Message; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"handle_update_state"/utf8>>, line => 141}) end, New_state = Transform(State), {continue, New_state, none}. -spec handle_crashed_connection(gleam@erlang@process:exit_message()) -> message(). handle_crashed_connection(Exit_message) -> {exit_message, Pid, Reason} = Exit_message, Connection = gmysql_ffi:from_pid(Pid), case Reason of normal -> {update_state, fun(State) -> erlang:setelement( 2, State, gleam@list:filter( erlang:element(2, State), fun(Slot) -> erlang:element(2, Slot) =:= {some, Connection} end ) ) end}; _ -> {update_state, fun(State@1) -> erlang:setelement( 2, State@1, gleam@list:map( erlang:element(2, State@1), fun(Slot@1) -> case erlang:element(2, Slot@1) =:= {some, Connection} of true -> erlang:setelement(2, Slot@1, none); false -> Slot@1 end end ) ) end} end. -spec handle_init(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_init(Message, State) -> {initialize, Self} = case Message of {initialize, _} -> Message; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"handle_init"/utf8>>, line => 177}) end, gleam_erlang_ffi:trap_exits(true), Selector = begin _pipe = gleam_erlang_ffi:new_selector(), _pipe@1 = gleam@erlang@process:selecting( _pipe, Self, fun gleam@function:identity/1 ), gleam@erlang@process:selecting_trapped_exits( _pipe@1, fun handle_crashed_connection/1 ) end, gleam@erlang@process:send(Self, {tick, Self}), {continue, State, {some, Selector}}. -spec handle_shutdown(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_shutdown(_, State) -> gleam@list:each( erlang:element(2, State), fun(Slot) -> case erlang:element(2, Slot) of none -> nil; {some, Conn} -> gmysql_ffi:close(Conn) end end ), {stop, normal}. -spec handle_restart(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_restart(Message, State) -> {restart, Conn} = case Message of {restart, _} -> Message; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"handle_restart"/utf8>>, line => 202}) end, Pid = gmysql_ffi:to_pid(Conn), gleam@erlang@process:kill(Pid), gleam@otp@actor:continue(State). -spec handle_restart_all(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_restart_all(_, State) -> gleam@list:each( erlang:element(2, State), fun(Slot) -> case erlang:element(2, Slot) of none -> nil; {some, Conn} -> _pipe = gmysql_ffi:to_pid(Conn), gleam@erlang@process:kill(_pipe) end end ), gleam@otp@actor:continue(State). -spec handle_start_new(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_start_new(_, State) -> gleam@otp@actor:continue( erlang:setelement( 2, State, [{slot, none, false} | erlang:element(2, State)] ) ). -spec handle_checkout(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_checkout(Message, State) -> {checkout, Client} = case Message of {checkout, _} -> Message; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"handle_checkout"/utf8>>, line => 230}) end, {Connection@1, Slots} = gleam@list:map_fold( erlang:element(2, State), none, fun(Connection, Slot) -> case {Connection, Slot} of {{some, _}, _} -> {Connection, Slot}; {none, {slot, {some, Conn}, false}} -> {{some, Conn}, erlang:setelement(3, Slot, true)}; {_, _} -> {Connection, Slot} end end ), _pipe = Connection@1, _pipe@1 = gleam@option:to_result(_pipe, nil), gleam@erlang@process:send(Client, _pipe@1), gleam@otp@actor:continue(erlang:setelement(2, State, Slots)). -spec handle_checkin(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_checkin(Message, State) -> {checkin, Connection} = case Message of {checkin, _} -> Message; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"handle_checkin"/utf8>>, line => 250}) end, gleam@otp@actor:continue( erlang:setelement( 2, State, gleam@list:map( erlang:element(2, State), fun(Slot) -> case erlang:element(2, Slot) =:= {some, Connection} of true -> erlang:setelement(3, Slot, false); false -> Slot end end ) ) ). -spec handle_tick(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_tick(Message, State) -> {tick, Self} = case Message of {tick, _} -> Message; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"handle_tick"/utf8>>, line => 266}) end, {Rate_limiter, Slots} = gleam@list:map_fold( erlang:element(2, State), erlang:element(4, State), fun(Rate_limit, Slot) -> case {gleam@otp@intensity_tracker:add_event(Rate_limit), Slot} of {{ok, Limiter}, {slot, none, _}} -> case gmysql:connect(erlang:element(3, State)) of {ok, Connection} -> {Limiter, {slot, {some, Connection}, false}}; {error, _} -> {Limiter, Slot} end; {_, {slot, none, _}} -> {Rate_limit, erlang:setelement(3, Slot, false)}; {_, Slot@1} -> {Rate_limit, Slot@1} end end ), gleam@erlang@process:send_after(Self, 250, {tick, Self}), gleam@otp@actor:continue( erlang:setelement(4, erlang:setelement(2, State, Slots), Rate_limiter) ). -spec handle_message(message(), state()) -> gleam@otp@actor:next(message(), state()). handle_message(Message, State) -> case Message of {update_state, _} -> handle_update_state(Message, State); {initialize, _} -> handle_init(Message, State); shutdown -> handle_shutdown(Message, State); {restart, _} -> handle_restart(Message, State); restart_all -> handle_restart_all(Message, State); start_new -> handle_start_new(Message, State); {checkout, _} -> handle_checkout(Message, State); {checkin, _} -> handle_checkin(Message, State); {tick, _} -> handle_tick(Message, State) end. -spec connect(gmysql:config(), integer(), integer()) -> pool(). connect(Config, Count, Max_connections_per_second) -> Slots = begin _pipe = gleam@iterator:repeat({slot, none, false}), _pipe@1 = gleam@iterator:take(_pipe, Count), gleam@iterator:to_list(_pipe@1) end, _assert_subject = gleam@otp@actor:start( {state, Slots, Config, gleam@otp@intensity_tracker:new(Max_connections_per_second, 1000)}, fun handle_message/2 ), {ok, Actor} = case _assert_subject of {ok, _} -> _assert_subject; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Assertion pattern match failed"/utf8>>, value => _assert_fail, module => <<"gmysql/pool"/utf8>>, function => <<"connect"/utf8>>, line => 46}) end, gleam@erlang@process:send(Actor, {initialize, Actor}), {pool, Actor}.