-module(dove@client). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function]). -export([connect/3]). -export_type([connection/1, message/1]). -type connection(HJB) :: {connection, binary(), gleam@erlang@process:subject(message(HJB))}. -type message(HJC) :: shutdown | {request, bitstring(), gleam@erlang@process:subject({ok, dove@response:response(HJC)} | {error, dove@error:error()}), gleam@option:option(fun((gleam@dynamic:dynamic_()) -> {ok, HJC} | {error, list(gleam@dynamic:decode_error())})), integer()}. -spec 'receive'( mug:socket(), gleam@erlang@process:selector({ok, bitstring()} | {error, mug:error()}), bitstring(), integer(), integer() ) -> {ok, {integer(), list({binary(), binary()}), binary()}} | {error, dove@error:error()}. 'receive'(Socket, Selector, Storage, Start_time, Timeout) -> case dove@response@decoder:decode(Storage) of {ok, Value} -> {ok, Value}; {error, more_needed} -> case (erlang:monotonic_time() - Start_time) >= (Timeout * 1000000) of true -> {error, {tcp_error, timeout}}; false -> case dove@tcp:'receive'(Socket, Selector, Timeout) of {error, Tcp_error} -> {error, {tcp_error, Tcp_error}}; {ok, Packet} -> 'receive'( Socket, Selector, gleam@bit_array:append(Storage, Packet), Start_time, Timeout ) end end; {error, Error} -> {error, Error} end. -spec handle_message(message(any()), mug:socket()) -> gleam@otp@actor:next(any(), mug:socket()). handle_message(Msg, Socket) -> case Msg of {request, Req, Reply_with, Decoder_option, Timeout} -> case dove@tcp:send(Socket, Req) of {ok, nil} -> Selector = dove@tcp:new_selector(), case 'receive'( Socket, Selector, <<>>, erlang:monotonic_time(), Timeout ) of {ok, Reply} -> _assert_subject = gleam@list:key_find( [{100, continue}, {101, switching_protocols}, {102, processing}, {103, early_hints}, {200, ok}, {201, created}, {202, accepted}, {203, non_authoritative_information}, {204, no_content}, {205, reset_content}, {206, partial_content}, {207, multi_status}, {300, multiple_choices}, {301, moved_permanently}, {302, found}, {303, see_other}, {304, not_modified}, {305, use_proxy}, {307, temporary_redirect}, {400, bad_request}, {401, unauthorized}, {402, payment_required}, {403, forbidden}, {404, not_found}, {405, method_not_allowed}, {406, not_acceptable}, {407, proxy_authentication_required}, {408, request_timeout}, {409, conflict}, {410, gone}, {411, length_required}, {412, precondition_failed}, {413, payload_too_large}, {414, uri_too_long}, {415, unsupported_media_type}, {416, range_not_satisfiable}, {417, expectation_failed}, {418, im_a_teapot}, {422, unprocessable_entity}, {423, locked}, {424, failed_dependency}, {425, too_early}, {426, upgrade_required}, {428, precondition_required}, {429, too_many_requests}, {431, request_header_fields_too_large}, {451, unavailable_for_legal_reasons}, {500, internal_server_error}, {501, not_implemented}, {502, bad_gateway}, {503, service_unavailable}, {504, gateway_timeout}, {505, http_version_not_supported}, {506, variant_also_negotiates}, {507, insufficient_storage}, {508, loop_detected}, {510, not_extended}, {511, network_authentication_required}], erlang:element(1, Reply) ), {ok, Status} = 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 => <<"dove/client"/utf8>>, function => <<"handle_message"/utf8>>, line => 61}) end, case Decoder_option of {some, Decoder} -> case gleam@json:decode( erlang:element(3, Reply), Decoder ) of {ok, Value} -> gleam@otp@actor:send( Reply_with, {ok, {response, Status, erlang:element(2, Reply), {decoded, Value}}} ); {error, Decode_error} -> gleam@otp@actor:send( Reply_with, {error, {decode_error, Decode_error}} ) end; none -> gleam@otp@actor:send( Reply_with, {ok, {response, Status, erlang:element(2, Reply), {raw, erlang:element(3, Reply)}}} ) end, gleam@otp@actor:continue(Socket); {error, Error} -> _ = mug_ffi:shutdown(Socket), gleam@otp@actor:send(Reply_with, {error, Error}), {stop, {abnormal, <<"TCP Error"/utf8>>}} end; {error, Error@1} -> _ = mug_ffi:shutdown(Socket), gleam@otp@actor:send( Reply_with, {error, {tcp_error, Error@1}} ), {stop, {abnormal, <<"TCP Error"/utf8>>}} end; shutdown -> _ = mug_ffi:shutdown(Socket), {stop, normal} end. -spec connect(binary(), integer(), integer()) -> {ok, connection(any())} | {error, gleam@otp@actor:start_error()}. connect(Host, Port, Timeout) -> gleam@result:then( gleam@otp@actor:start_spec( {spec, fun() -> case dove@tcp:connect(Host, Port, Timeout) of {ok, Socket} -> {ready, Socket, gleam_erlang_ffi:new_selector()}; {error, _} -> {failed, <<"Unable to connect to Redis server"/utf8>>} end end, Timeout, fun handle_message/2} ), fun(Subject) -> _pipe = (<<<>/binary, (gleam@int:to_string(Port))/binary>>), _pipe@1 = {connection, _pipe, Subject}, {ok, _pipe@1} end ).