-module(glats@handler). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -export([handle_request/5]). -export_type([request/0, response/0, request_outcome/1, request_handler_state/1]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. ?MODULEDOC( " Request handler will handle receiving messages from a subscription,\n" " pass the data to a callback and then reply to NATS with the returned\n" " data.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import gleam/option.{None}\n" " import gleam/result\n" " import gleam/erlang/process\n" " import glats\n" " import glats/handler.{Reply, Request, Response}\n" " \n" " pub fn main() {\n" " use conn <- result.then(glats.connect(\"localhost\", 4222, []))\n" " \n" " // Start a request handler actor that will call `ping_pong_handler` for\n" " // every request received from NATS topic \"do.ping\".\n" " assert Ok(_actor) =\n" " handler.handle_request(conn, [], \"do.ping\", None, ping_pong_handler)\n" " \n" " process.sleep_forever()\n" " \n" " Ok(Nil)\n" " }\n" " \n" " pub fn ping_pong_handler(req: Request, state) {\n" " // Got message: Hello\n" " io.println(\"Got message: \" <> req.body)\n" "\n" " // Reply with a message with the same headers and append to body.\n" " Reply(\n" " Response(\n" " headers: req.headers,\n" " reply_to: None,\n" " body: req.body <> \" from glats!\",\n" " ),\n" " state,\n" " )\n" " }\n" " ```\n" "\n" " Then in a shell with `natscli`.\n" "\n" " ```sh\n" " $ nats req do.ping 'Hello'\n" " 12:16:47 Sending request on \"do.ping\"\n" " 12:16:47 Received with rtt 427.64µs\n" " Hello from glats!\n" " ```\n" "\n" ). -type request() :: {request, gleam@dict:dict(binary(), binary()), binary()}. -type response() :: {response, gleam@dict:dict(binary(), binary()), gleam@option:option(binary()), binary()}. -type request_outcome(GYA) :: {reply, response(), GYA} | {stop, gleam@erlang@process:exit_reason()}. -type request_handler_state(GYB) :: {request_handler_state, gleam@erlang@process:subject(glats:connection_message()), integer(), fun((request(), GYB) -> request_outcome(GYB)), GYB}. -file("src/glats/handler.gleam", 141). -spec request_handler_msg( gleam@erlang@process:subject(glats:connection_message()), glats:message(), request_handler_state(GYL) ) -> gleam@otp@actor:next(any(), request_handler_state(GYL)). request_handler_msg(Conn, Msg, State) -> case erlang:element(4, Msg) of {some, Reply_to} -> Req = {request, erlang:element(3, Msg), erlang:element(5, Msg)}, case (erlang:element(4, State))(Req, erlang:element(5, State)) of {reply, Res, New_inner} -> Pub_res = glats:publish_message( Conn, {message, Reply_to, erlang:element(2, Res), erlang:element(3, Res), erlang:element(4, Res)} ), case Pub_res of {ok, nil} -> {continue, begin _record = State, {request_handler_state, erlang:element(2, _record), erlang:element(3, _record), erlang:element(4, _record), New_inner} end, none}; {error, Err} -> {stop, {abnormal, gleam@string:inspect(Err)}} end; {stop, Reason} -> {stop, Reason} end; none -> {continue, State, none} end. -file("src/glats/handler.gleam", 131). -spec request_handler_loop( glats:subscription_message(), request_handler_state(GYI) ) -> gleam@otp@actor:next(any(), request_handler_state(GYI)). request_handler_loop(Message, State) -> case Message of {received_message, Conn, _, _, Msg} -> request_handler_msg(Conn, Msg, State) end. -file("src/glats/handler.gleam", 106). ?DOC( " Starts an actor that subscribes to the desired NATS topic and calls the\n" " provided request handler with the request data and replies to NATS with\n" " the returned message data from the request handler.\n" ). -spec handle_request( gleam@erlang@process:subject(glats:connection_message()), GYE, binary(), list(glats:subscribe_option()), fun((request(), GYE) -> request_outcome(GYE)) ) -> {ok, gleam@erlang@process:subject(glats:subscription_message())} | {error, gleam@otp@actor:start_error()}. handle_request(Conn, State, Topic, Opts, Handler) -> gleam@otp@actor:start_spec( {spec, fun() -> Subscriber = gleam@erlang@process:new_subject(), Selector = begin _pipe = gleam_erlang_ffi:new_selector(), gleam@erlang@process:selecting( _pipe, Subscriber, fun(Msg) -> Msg end ) end, case glats:subscribe(Conn, Subscriber, Topic, Opts) of {ok, Sid} -> {ready, {request_handler_state, Conn, Sid, Handler, State}, Selector}; {error, Err} -> {failed, gleam@string:inspect(Err)} end end, 5000, fun request_handler_loop/2} ).