diff --git a/CHANGELOG.md b/CHANGELOG.md index 12a3f76..3e7247f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,20 @@ # Unreleased +- `new` now takes `listener_name` and `connection_factory_name` as its first two + arguments, created once by the caller with `process.new_name`. This replaces + the previous behaviour of generating these names internally on every + `start` and `start_unix` call, which could exhaust the atom table under + supervised restarts +- Resolve the connection factory supervisor once per acceptor instead of on + every accepted connection +- Fix connection handler not applying user's selector on continue. +- `get_server_info` now takes the listener's `Subject` rather than its `Name` +- Replace `ServerInfo` and `ConnectionInfo` with a single `SocketAddress` type - support in-memory `PEM` and `DER` certificates for TLS - support mTLS - Support unix sockets +- Support custom connection shutdown timeout +- Support updating active state on continue # v9.0.1 diff --git a/examples/echo_server/src/echo_server.gleam b/examples/echo_server/src/echo_server.gleam index f938b23..d237c45 100644 --- a/examples/echo_server/src/echo_server.gleam +++ b/examples/echo_server/src/echo_server.gleam @@ -8,32 +8,38 @@ import logging pub fn main() { let listener_name = process.new_name("glisten_listener") + let connection_factory_name = process.new_name("glisten_connection_factory") logging.configure() logging.set_level(logging.Debug) let assert Ok(_server) = - glisten.new(fn(_conn) { #(Nil, None) }, fn(state, msg, conn) { - let assert Ok(info) = glisten.get_connection_info(conn) - logging.log( - logging.Info, - "Client connected at " - <> glisten.ip_address_to_string(info.ip_address) - <> " at port " - <> int.to_string(info.port), - ) + glisten.new( + listener_name, + connection_factory_name, + fn(_conn) { #(Nil, None) }, + fn(state, msg, conn) { + let assert Ok(glisten.TcpSocketAddress(port:, ip_address:)) = + glisten.get_connection_info(conn) + logging.log( + logging.Info, + "Client connected at " + <> glisten.ip_address_to_string(ip_address) + <> " at port " + <> int.to_string(port), + ) - let assert Packet(msg) = msg - let assert Ok(_) = glisten.send(conn, bytes_tree.from_bit_array(msg)) - glisten.continue(state) - }) + let assert Packet(msg) = msg + let assert Ok(_) = glisten.send(conn, bytes_tree.from_bit_array(msg)) + glisten.continue(state) + }, + ) |> glisten.bind("localhost") |> glisten.with_ipv6 - |> glisten.with_listener_name(listener_name) |> glisten.start(0) - let assert glisten.TcpServerInfo(port:, ip_address:) = - glisten.get_server_info(listener_name, 5000) + let assert glisten.TcpSocketAddress(port:, ip_address:) = + glisten.get_server_info(process.named_subject(listener_name), 5000) io.println( "Listening on " diff --git a/examples/echo_ssl/src/echo_server.gleam b/examples/echo_ssl/src/echo_server.gleam index fe6d4c4..edcf3b4 100644 --- a/examples/echo_ssl/src/echo_server.gleam +++ b/examples/echo_ssl/src/echo_server.gleam @@ -8,30 +8,36 @@ import logging pub fn main() { let listener_name = process.new_name("glisten_listener") + let connection_factory_name = process.new_name("glisten_connection_factory") logging.configure() logging.set_level(logging.Debug) let assert Ok(_server) = - glisten.new(fn(_conn) { #(Nil, None) }, fn(state, msg, conn) { - let assert Ok(info) = glisten.get_connection_info(conn) - logging.log( - logging.Info, - "Client connected at " - <> glisten.ip_address_to_string(info.ip_address) - <> " at port " - <> int.to_string(info.port), - ) - let assert Packet(msg) = msg - let assert Ok(_) = glisten.send(conn, bytes_tree.from_bit_array(msg)) - glisten.continue(state) - }) + glisten.new( + listener_name, + connection_factory_name, + fn(_conn) { #(Nil, None) }, + fn(state, msg, conn) { + let assert Ok(glisten.TcpSocketAddress(port:, ip_address:)) = + glisten.get_connection_info(conn) + logging.log( + logging.Info, + "Client connected at " + <> glisten.ip_address_to_string(ip_address) + <> " at port " + <> int.to_string(port), + ) + let assert Packet(msg) = msg + let assert Ok(_) = glisten.send(conn, bytes_tree.from_bit_array(msg)) + glisten.continue(state) + }, + ) |> glisten.with_tls(certfile: "localhost.crt", keyfile: "localhost.key") - |> glisten.with_listener_name(listener_name) |> glisten.start(0) - let assert glisten.TcpServerInfo(port, ip_address) = - glisten.get_server_info(listener_name, 5000) + let assert glisten.TcpSocketAddress(port, ip_address) = + glisten.get_server_info(process.named_subject(listener_name), 5000) io.println( "Listening on " diff --git a/examples/echo_unix/src/echo_server.gleam b/examples/echo_unix/src/echo_server.gleam index c3ac220..32e93a9 100644 --- a/examples/echo_unix/src/echo_server.gleam +++ b/examples/echo_unix/src/echo_server.gleam @@ -7,24 +7,29 @@ import logging pub fn main() { let listener_name = process.new_name("glisten_listener") + let connection_factory_name = process.new_name("glisten_connection_factory") logging.configure() logging.set_level(logging.Debug) let assert Ok(_server) = - glisten.new(fn(_conn) { #(Nil, None) }, fn(state, msg, conn) { - logging.log(logging.Info, "Client connected via unix socket") + glisten.new( + listener_name, + connection_factory_name, + fn(_conn) { #(Nil, None) }, + fn(state, msg, conn) { + logging.log(logging.Info, "Client connected via unix socket") - let assert Packet(msg) = msg - let assert Ok(_) = glisten.send(conn, bytes_tree.from_bit_array(msg)) - glisten.continue(state) - }) + let assert Packet(msg) = msg + let assert Ok(_) = glisten.send(conn, bytes_tree.from_bit_array(msg)) + glisten.continue(state) + }, + ) |> glisten.with_ipv6 - |> glisten.with_listener_name(listener_name) |> glisten.start_unix("/tmp/test.sock") - let assert glisten.UnixServerInfo(path:) = - glisten.get_server_info(listener_name, 5000) + let assert glisten.UnixSocketAddress(path:) = + glisten.get_server_info(process.named_subject(listener_name), 5000) io.println("Listening on " <> path) diff --git a/src/glisten.gleam b/src/glisten.gleam index 17c9b19..b8e86ea 100644 --- a/src/glisten.gleam +++ b/src/glisten.gleam @@ -19,6 +19,7 @@ import glisten/socket.{ } import glisten/socket/options import glisten/transport.{type Transport} +import logging /// Your provided loop function will receive these message types as the /// second argument. @@ -41,26 +42,31 @@ pub type Socket = pub type SocketReason = InternalSocketReason -pub type ConnectionInfo { - ConnectionInfo(port: Int, ip_address: IpAddress) +/// The address a socket is bound to, or the address of a connected peer. +/// Unix sockets have no port/IP, only the bound path. +pub type SocketAddress { + TcpSocketAddress(port: Int, ip_address: IpAddress) + UnixSocketAddress(path: String) } -pub type ServerInfo { - TcpServerInfo(port: Int, ip_address: IpAddress) - UnixServerInfo(path: String) +@internal +pub fn convert_sock_name(sock_name: socket.SockName) -> SocketAddress { + case sock_name { + socket.TcpSockName(ip, port) -> + TcpSocketAddress(port, convert_ip_address(ip)) + socket.UnixSockName(path) -> UnixSocketAddress(path) + } } /// Returns the user-provided port or the OS-assigned value if 0 was provided. +/// +/// Pass the subject for the `listener_name` given to `new`. pub fn get_server_info( - listener: process.Name(listener.Message), + listener: Subject(listener.Message), timeout: Int, -) -> ServerInfo { - let listener = process.named_subject(listener) +) -> SocketAddress { let state = process.call(listener, timeout, listener.Info) - case state.sock_name { - socket.TcpSockName(ip, port) -> TcpServerInfo(port, convert_ip_address(ip)) - socket.UnixSockName(path) -> UnixServerInfo(path) - } + convert_sock_name(state.sock_name) } /// This type holds useful bits of data for the active connection. @@ -136,14 +142,15 @@ fn ipv6_zeros( } } -/// Tries to read the IP address and port of a connected client. It will -/// return valid IPv4 or IPv6 addresses, attempting to return the most relevant -/// one for the client. +/// Tries to read the address of a connected client. For TCP/TLS connections +/// this is the IPv4 or IPv6 address and port, attempting to return the most +/// relevant one for the client. For unix socket connections this is the +/// bound path, since unix sockets have no per-client address. pub fn get_connection_info( conn: Connection(user_message), -) -> Result(ConnectionInfo, Nil) { +) -> Result(SocketAddress, Nil) { transport.peername(conn.transport, conn.socket) - |> result.map(fn(pair) { ConnectionInfo(pair.1, convert_ip_address(pair.0)) }) + |> result.map(convert_sock_name) } /// Sends a BytesTree message over the socket using the active transport @@ -155,13 +162,17 @@ pub fn send( } pub opaque type Next(user_state, user_message) { - Continue(state: user_state, selector: Option(Selector(user_message))) + Continue( + state: user_state, + selector: Option(Selector(user_message)), + active_state: Option(options.ActiveState), + ) NormalStop AbnormalStop(String) } pub fn continue(state: user_state) -> Next(user_state, user_message) { - Continue(state, None) + Continue(state, None, None) } pub fn with_selector( @@ -169,11 +180,44 @@ pub fn with_selector( selector: Selector(user_message), ) -> Next(user_state, user_message) { case next { - Continue(state, _) -> Continue(state, Some(selector)) + Continue(state, _, active_state) -> + Continue(state, Some(selector), active_state) stop -> stop } } +/// Overrides the socket's `ActiveState` starting with this message. Applied +/// immediately when this `Next` is processed and persists as the new default +/// for future commands until changed again. Allowed are `Once`, `Active` and +/// `Count(n)` where n > 1. An invalid value is ignored. +pub fn set_active_state( + next: Next(user_state, user_message), + active_state: options.ActiveState, +) -> Next(user_state, user_message) { + case active_state { + options.Passive -> { + logging.log( + logging.Warning, + "Ignoring set_active_state: cannot set the connection's `ActiveState` to `Passive`", + ) + next + } + options.Count(n) if n <= 1 -> { + logging.log( + logging.Warning, + "Ignoring set_active_state: Count shall be greater than 1", + ) + next + } + options.Once | options.Active | options.Count(_) -> + case next { + Continue(state, selector, _) -> + Continue(state, selector, Some(active_state)) + stop -> stop + } + } +} + pub fn stop() -> Next(user_state, user_message) { NormalStop } @@ -187,7 +231,8 @@ pub fn convert_next( next: Next(state, user_message), ) -> handler.Next(state, user_message) { case next { - Continue(state, selector) -> handler.Continue(state, selector) + Continue(state, selector, active_state) -> + handler.Continue(state, selector, active_state) NormalStop -> handler.NormalStop AbnormalStop(reason) -> handler.AbnormalStop(reason) } @@ -199,9 +244,13 @@ pub fn map_selector( mapper: fn(user_message) -> other_message, ) -> Next(state, other_message) { case next { - Continue(state, Some(selector)) -> - Continue(state, Some(process.map_selector(selector, mapper))) - Continue(state, None) -> Continue(state, None) + Continue(state, Some(selector), active_state) -> + Continue( + state, + Some(process.map_selector(selector, mapper)), + active_state, + ) + Continue(state, None, active_state) -> Continue(state, None, active_state) AbnormalStop(reason) -> AbnormalStop(reason) NormalStop -> NormalStop } @@ -225,12 +274,11 @@ pub opaque type Builder(state, user_message) { ipv6_support: Bool, tls_options: Option(options.TlsCerts), client_verification: Option(CaCert), - listener_name: Option(process.Name(listener.Message)), - connection_factory_name: Option( - process.Name( - factory.Message(Socket, Subject(handler.Message(user_message))), - ), + listener_name: process.Name(listener.Message), + connection_factory_name: process.Name( + factory.Message(Socket, Subject(handler.Message(user_message))), ), + connection_shutdown_timeout_ms: Int, active_state: options.ActiveState, ) } @@ -256,13 +304,18 @@ fn convert_loop( handler.Custom(msg) -> User(msg) } case loop(data, message, conn) { - Continue(data, selector) -> - case selector { + Continue(data, selector, active_state) -> { + let next = case selector { Some(selector) -> handler.continue(data) |> handler.with_selector(map_user_selector(selector)) _ -> handler.continue(data) } + case active_state { + Some(active_state) -> handler.with_active_state(next, active_state) + None -> next + } + } NormalStop -> handler.stop() AbnormalStop(reason) -> handler.stop_abnormal(reason) @@ -286,13 +339,24 @@ fn convert_on_init( } } -/// Create a new handler for each connection. The required arguments mirror the -/// `actor.start` API from `gleam_otp`. The default pool is 10 accceptor -/// processes. +/// Create a new handler for each connection. The `on_init` and `loop` +/// arguments mirror the `actor.start` API from `gleam_otp`. The default pool +/// is 10 acceptor processes. +/// +/// `listener_name` and `connection_factory_name` are process names `glisten` +/// uses internally to wire the listener and connection factory together. +/// Create each once, at the point your program starts, and pass them in here. +/// +/// `listener_name` is also what you use for `get_server_info` if you need to +/// read back the bound port or path. pub fn new( - on_init: fn(Connection(user_message)) -> + listener_name listener_name: process.Name(listener.Message), + connection_factory_name connection_factory_name: process.Name( + factory.Message(Socket, Subject(handler.Message(user_message))), + ), + on_init on_init: fn(Connection(user_message)) -> #(state, Option(Selector(user_message))), - loop: Loop(state, user_message), + loop loop: Loop(state, user_message), ) -> Builder(state, user_message) { Builder( interface: options.Loopback, @@ -304,8 +368,9 @@ pub fn new( ipv6_support: False, tls_options: None, client_verification: None, - listener_name: None, - connection_factory_name: None, + listener_name:, + connection_factory_name:, + connection_shutdown_timeout_ms: 5000, active_state: options.Once, ) } @@ -426,12 +491,22 @@ pub fn with_active_state( } } +/// Sets how long, in milliseconds, a connection handler is given to shut down +/// once the supervisor asks it to before being forcefully killed. Default is +/// 5000. +pub fn with_connection_shutdown_timeout_ms( + builder: Builder(state, user_message), + timeout_ms: Int, +) -> Builder(state, user_message) { + Builder(..builder, connection_shutdown_timeout_ms: timeout_ms) +} + @internal pub fn with_listener_name( builder: Builder(state, user_message), listener_name: process.Name(listener.Message), ) -> Builder(state, user_message) { - Builder(..builder, listener_name: Some(listener_name)) + Builder(..builder, listener_name:) } @internal @@ -441,7 +516,7 @@ pub fn with_connection_factory_name( factory.Message(socket.Socket, Subject(handler.Message(user_message))), ), ) -> Builder(state, user_message) { - Builder(..builder, connection_factory_name: Some(connection_factory_name)) + Builder(..builder, connection_factory_name:) } /// Start the TCP server with the given handler on the provided port @@ -449,13 +524,8 @@ pub fn start( builder: Builder(state, user_message), port: Int, ) -> Result(actor.Started(supervisor.Supervisor), actor.StartError) { - let listener_name = - option.unwrap(builder.listener_name, process.new_name("glisten_listener")) - let connection_supervisor = - option.unwrap( - builder.connection_factory_name, - process.new_name("glisten_connection_supervisor"), - ) + let listener_name = builder.listener_name + let connection_supervisor = builder.connection_factory_name let options = [options.Ip(builder.interface)] |> list.append(case builder.ipv6_support { @@ -498,6 +568,7 @@ pub fn start( on_close: builder.on_close, transport:, active_state: builder.active_state, + connection_shutdown_timeout_ms: builder.connection_shutdown_timeout_ms, ) |> acceptor.start_pool(transport, port, options, listener_name) } @@ -517,13 +588,8 @@ pub fn start_unix( } } - let listener_name = - option.unwrap(builder.listener_name, process.new_name("glisten_listener")) - let connection_supervisor = - option.unwrap( - builder.connection_factory_name, - process.new_name("glisten_connection_supervisor"), - ) + let listener_name = builder.listener_name + let connection_supervisor = builder.connection_factory_name let options = [options.Ip(options.UnixPath(path))] @@ -550,6 +616,7 @@ pub fn start_unix( on_close: builder.on_close, transport:, active_state: builder.active_state, + connection_shutdown_timeout_ms: builder.connection_shutdown_timeout_ms, ) |> acceptor.start_pool(transport, 0, options, listener_name) } diff --git a/src/glisten/internal/acceptor.gleam b/src/glisten/internal/acceptor.gleam index 8bd8588..21b98b3 100644 --- a/src/glisten/internal/acceptor.gleam +++ b/src/glisten/internal/acceptor.gleam @@ -26,11 +26,15 @@ pub type AcceptorError { ControlError(atom.Atom) } -pub type AcceptorState { +pub type AcceptorState(user_message) { AcceptorState( sender: Subject(AcceptorMessage), socket: Option(Socket), transport: Transport, + connection_factory: factory.Supervisor( + Socket, + Subject(handler.Message(user_message)), + ), ) } @@ -49,7 +53,9 @@ pub fn start( let state = process.call(listener, 750, listener.Info) process.send(subject, AcceptConnection(state.listen_socket)) - AcceptorState(subject, None, pool.transport) + let connection_factory = factory.get_by_name(connection_supervisor) + + AcceptorState(subject, None, pool.transport, connection_factory) |> actor.initialised |> actor.returning(subject) |> actor.selecting( @@ -59,7 +65,7 @@ pub fn start( |> Ok }) |> actor.on_message(fn(state, msg) { - let AcceptorState(sender, ..) = state + let AcceptorState(sender, connection_factory:, ..) = state case msg { AcceptConnection(listener) -> { let res = { @@ -67,7 +73,6 @@ pub fn start( transport.accept(state.transport, listener) |> result.map_error(AcceptError), ) - let connection_factory = factory.get_by_name(connection_supervisor) case factory.start_child(connection_factory, sock) { Ok(start) -> { transport.controlling_process(state.transport, sock, start.pid) @@ -122,6 +127,7 @@ pub type Pool(data, user_message) { on_init: fn(Connection(user_message)) -> #(data, Option(Selector(user_message))), on_close: Option(fn(data) -> Nil), + connection_shutdown_timeout_ms: Int, transport: Transport, active_state: options.ActiveState, ) @@ -166,6 +172,7 @@ pub fn start_pool( active_state: pool.active_state, )) }) + |> factory.timeout(pool.connection_shutdown_timeout_ms) |> factory.named(pool.name) |> factory.restart_strategy(supervision.Temporary) |> factory.supervised, diff --git a/src/glisten/internal/handler.gleam b/src/glisten/internal/handler.gleam index a5a4ab4..0e66567 100644 --- a/src/glisten/internal/handler.gleam +++ b/src/glisten/internal/handler.gleam @@ -6,7 +6,7 @@ import gleam/otp/actor import gleam/result import gleam/string import glisten/socket.{type Socket, type SocketReason} -import glisten/socket/options.{type ActiveState, type IpAddress} +import glisten/socket/options.{type ActiveState} import glisten/transport.{type Transport} import logging @@ -35,7 +35,7 @@ pub type LoopMessage(user_message) { } pub type ClientIp = - Result(#(IpAddress, Int), Nil) + Result(socket.SockName, Nil) pub type LoopState(state, user_message) { LoopState( @@ -44,6 +44,7 @@ pub type LoopState(state, user_message) { sender: Subject(Message(user_message)), transport: Transport, state: state, + active_state: ActiveState, ) } @@ -57,13 +58,17 @@ pub type Connection(user_message) { } pub type Next(user_state, user_message) { - Continue(state: user_state, selector: Option(Selector(user_message))) + Continue( + state: user_state, + selector: Option(Selector(user_message)), + active_state: Option(ActiveState), + ) NormalStop AbnormalStop(reason: String) } pub fn continue(state: user_state) -> Next(user_state, user_message) { - Continue(state, None) + Continue(state, None, None) } pub fn with_selector( @@ -71,7 +76,19 @@ pub fn with_selector( selector: Selector(user_message), ) -> Next(user_state, user_message) { case next { - Continue(state, _) -> Continue(state, Some(selector)) + Continue(state, _, active_state) -> + Continue(state, Some(selector), active_state) + stop -> stop + } +} + +pub fn with_active_state( + next: Next(user_state, user_message), + active_state: ActiveState, +) -> Next(user_state, user_message) { + case next { + Continue(state, selector, _) -> + Continue(state, selector, Some(active_state)) stop -> stop } } @@ -84,6 +101,108 @@ pub fn stop_abnormal(reason: String) -> Next(user_state, user_message) { AbnormalStop(reason) } +fn apply_next( + state: LoopState(state, user_message), + res: Result(Next(state, LoopMessage(user_message)), Dynamic), + packet_consumed: Bool, +) -> actor.Next(LoopState(state, user_message), Message(user_message)) { + case res { + Ok(Continue(next_state, selector, Some(new_active_state))) -> + case + transport.set_opts(state.transport, state.socket, [ + options.ActiveMode(new_active_state), + ]) + { + Ok(Nil) -> + actor.continue( + LoopState( + ..state, + state: next_state, + active_state: new_active_state, + ), + ) + |> apply_selector(state.sender, selector) + Error(_) -> actor.stop() + } + Ok(Continue(next_state, selector, None)) + if packet_consumed && state.active_state == options.Once + -> + case + transport.set_opts(state.transport, state.socket, [ + options.ActiveMode(options.Once), + ]) + { + Ok(Nil) -> + actor.continue(LoopState(..state, state: next_state)) + |> apply_selector(state.sender, selector) + Error(_) -> actor.stop() + } + Ok(Continue(next_state, selector, None)) -> + actor.continue(LoopState(..state, state: next_state)) + |> apply_selector(state.sender, selector) + Ok(NormalStop) -> actor.stop() + Ok(AbnormalStop(reason)) -> actor.stop_abnormal(reason) + Error(reason) -> { + logging.log( + logging.Error, + "Caught error in user handler: " <> string.inspect(reason), + ) + actor.continue(state) + } + } +} + +// Applies a selector returned from `Continue`. +fn apply_selector( + next: actor.Next(LoopState(state, user_message), Message(user_message)), + sender: Subject(Message(user_message)), + selector: Option(Selector(LoopMessage(user_message))), +) -> actor.Next(LoopState(state, user_message), Message(user_message)) { + case selector { + None -> next + Some(selector) -> { + let mapped = + process.map_selector(selector, fn(loop_message) { + case loop_message { + Custom(message) -> User(message) + Packet(bits) -> Internal(ReceiveMessage(bits)) + } + }) + + actor.with_selector( + next, + internal_selector(sender) |> process.merge_selector(mapped), + ) + } + } +} + +// The intenral selector that contains socket events mapped to `Internal` plus +// the connection's own subject. +fn internal_selector( + sender: Subject(Message(user_message)), +) -> Selector(Message(user_message)) { + process.new_selector() + |> process.select_record(atom.create("tcp"), 2, fn(record) { + ReceiveMessage(socket_data(record)) + }) + |> process.select_record(atom.create("ssl"), 2, fn(record) { + ReceiveMessage(socket_data(record)) + }) + |> process.select_record(atom.create("ssl_closed"), 1, fn(_nil) { Closed }) + |> process.select_record(atom.create("tcp_closed"), 1, fn(_nil) { Closed }) + |> process.select_record(atom.create("ssl_passive"), 1, fn(_nil) { Passive }) + |> process.select_record(atom.create("tcp_passive"), 1, fn(_nil) { Passive }) + |> process.select_record(atom.create("tcp_error"), 2, fn(record) { + SocketError(socket_error(record)) + }) + |> process.select_record(atom.create("ssl_error"), 2, fn(record) { + SocketError(socket_error(record)) + }) + |> process.map_selector(Internal) + |> process.merge_selector(process.new_selector() |> process.select(sender)) +} + pub type Loop(state, user_message) = fn(state, LoopMessage(user_message), Connection(user_message)) -> Next(state, LoopMessage(user_message)) @@ -154,6 +273,7 @@ pub fn start( sender: subject, transport: handler.transport, state: initial_state, + active_state: handler.active_state, ) |> actor.initialised() |> actor.selecting(selector) @@ -193,7 +313,7 @@ pub fn start( // Note that the active_state must set to Passive at start of // Listener/Accept and not changed until the Ready message is // received. - let options = [options.ActiveMode(handler.active_state)] + let options = [options.ActiveMode(state.active_state)] case transport.set_opts(state.transport, state.socket, options) { Ok(_) -> actor.continue(state) Error(_) -> actor.stop_abnormal("Failed to set socket active") @@ -203,64 +323,16 @@ pub fn start( User(msg) -> { let msg = Custom(msg) let res = rescue(fn() { handler.loop(state.state, msg, connection) }) - case res { - Ok(Continue(next_state, _selector)) - if handler.active_state == options.Once - -> { - case - transport.set_opts(state.transport, state.socket, [ - options.ActiveMode(options.Once), - ]) - { - Ok(Nil) -> actor.continue(LoopState(..state, state: next_state)) - Error(_) -> actor.stop() - } - } - Ok(Continue(next_state, _selector)) -> - actor.continue(LoopState(..state, state: next_state)) - Ok(NormalStop) -> actor.stop() - Ok(AbnormalStop(reason)) -> actor.stop_abnormal(reason) - Error(reason) -> { - logging.log( - logging.Error, - "Caught error in user handler: " <> string.inspect(reason), - ) - actor.continue(state) - } - } + apply_next(state, res, False) } Internal(ReceiveMessage(msg)) -> { let msg = Packet(msg) let res = rescue(fn() { handler.loop(state.state, msg, connection) }) - case res { - Ok(Continue(next_state, _selector)) - if handler.active_state == options.Once - -> { - case - transport.set_opts(state.transport, state.socket, [ - options.ActiveMode(options.Once), - ]) - { - Ok(Nil) -> actor.continue(LoopState(..state, state: next_state)) - Error(_) -> actor.stop() - } - } - Ok(Continue(next_state, _selector)) -> - actor.continue(LoopState(..state, state: next_state)) - Ok(NormalStop) -> actor.stop() - Ok(AbnormalStop(reason)) -> actor.stop_abnormal(reason) - Error(reason) -> { - logging.log( - logging.Error, - "Caught error in user handler: " <> string.inspect(reason), - ) - actor.continue(state) - } - } + apply_next(state, res, True) } Internal(Passive) -> { let options = [ - options.ActiveMode(handler.active_state), + options.ActiveMode(state.active_state), ] case transport.set_opts(state.transport, state.socket, options) { Ok(_) -> actor.continue(state) diff --git a/src/glisten/ssl.gleam b/src/glisten/ssl.gleam index 1235e6c..f0e95d2 100644 --- a/src/glisten/ssl.gleam +++ b/src/glisten/ssl.gleam @@ -80,9 +80,7 @@ pub fn listen( pub fn negotiated_protocol(socket: Socket) -> Result(String, String) @external(erlang, "glisten_ssl_ffi", "peername") -pub fn peername( - socket: Socket, -) -> Result(#(options.IpAddress, Int), SocketReason) +pub fn peername(socket: Socket) -> Result(socket.SockName, SocketReason) @external(erlang, "glisten_ssl_ffi", "sockname") pub fn sockname(socket: ListenSocket) -> Result(socket.SockName, SocketReason) diff --git a/src/glisten/tcp.gleam b/src/glisten/tcp.gleam index d2b243b..83f2b9b 100644 --- a/src/glisten/tcp.gleam +++ b/src/glisten/tcp.gleam @@ -97,9 +97,7 @@ pub fn handshake(socket: Socket) -> Result(Socket, Nil) { pub fn negotiated_protocol(socket: Socket) -> a @external(erlang, "glisten_tcp_ffi", "peername") -pub fn peername( - socket: Socket, -) -> Result(#(options.IpAddress, Int), SocketReason) +pub fn peername(socket: Socket) -> Result(socket.SockName, SocketReason) @external(erlang, "inet", "getopts") pub fn get_socket_opts( diff --git a/src/glisten/transport.gleam b/src/glisten/transport.gleam index d5cda51..0b99c05 100644 --- a/src/glisten/transport.gleam +++ b/src/glisten/transport.gleam @@ -175,7 +175,7 @@ pub fn decode_ip() -> Decoder(options.IpAddress) { pub fn peername( transport: Transport, socket: Socket, -) -> Result(#(options.IpAddress, Int), Nil) { +) -> Result(socket.SockName, Nil) { case transport { Tcp -> tcp.peername(socket) Ssl -> ssl.peername(socket) diff --git a/src/glisten_ssl_ffi.erl b/src/glisten_ssl_ffi.erl index 6033c2f..283bd00 100644 --- a/src/glisten_ssl_ffi.erl +++ b/src/glisten_ssl_ffi.erl @@ -62,10 +62,10 @@ sockname(Socket) -> peername(Socket) -> case ssl:peername(Socket) of - {ok, {local, _}} -> - {error, enotconn}; + {ok, {local, Path}} -> + {ok, {unix_sock_name, Path}}; {ok, {Ip, Port}} -> - {ok, {normalize_ip(Ip), Port}}; + {ok, {tcp_sock_name, normalize_ip(Ip), Port}}; {error, Reason} -> {error, Reason} end. diff --git a/src/glisten_tcp_ffi.erl b/src/glisten_tcp_ffi.erl index 2ba2e92..b8db0e8 100644 --- a/src/glisten_tcp_ffi.erl +++ b/src/glisten_tcp_ffi.erl @@ -54,10 +54,10 @@ sockname(Socket) -> peername(Socket) -> case inet:peername(Socket) of - {ok, {local, _}} -> - {error, enotconn}; + {ok, {local, Path}} -> + {ok, {unix_sock_name, Path}}; {ok, {Ip, Port}} -> - {ok, {normalize_ip(Ip), Port}}; + {ok, {tcp_sock_name, normalize_ip(Ip), Port}}; {error, Reason} -> {error, Reason} end. diff --git a/test/glisten_test.gleam b/test/glisten_test.gleam index 1e305df..9b2d9f2 100644 --- a/test/glisten_test.gleam +++ b/test/glisten_test.gleam @@ -45,12 +45,19 @@ pub fn it_echoes_messages_test() { pub fn it_accepts_from_the_pool_test() { let client_sender = process.new_subject() + let listener_name = process.new_name("glisten_listener") + let connection_factory_name = process.new_name("glisten_connection_factory") let assert Ok(_server) = - glisten.new(fn(_conn) { #(Nil, None) }, fn(state, msg, conn) { - let assert Packet(msg) = msg - let assert Ok(_) = tcp.send(conn.socket, bytes_tree.from_bit_array(msg)) - glisten.continue(state) - }) + glisten.new( + listener_name, + connection_factory_name, + fn(_conn) { #(Nil, None) }, + fn(state, msg, conn) { + let assert Packet(msg) = msg + let assert Ok(_) = tcp.send(conn.socket, bytes_tree.from_bit_array(msg)) + glisten.continue(state) + }, + ) |> glisten.with_pool_size(1) |> glisten.start(54_321)