import gleam/bytes_tree.{type BytesTree} import gleam/erlang/charlist.{type Charlist} import gleam/erlang/process.{type Selector, type Subject} import gleam/int import gleam/list import gleam/option.{type Option, None, Some} import gleam/otp/actor import gleam/otp/factory_supervisor as factory import gleam/otp/static_supervisor as supervisor import gleam/otp/supervision.{type ChildSpecification} import gleam/result import gleam/string import glisten/internal/acceptor.{Pool} import glisten/internal/handler import glisten/internal/listener import glisten/socket.{ type Socket as InternalSocket, type SocketReason as InternalSocketReason, } import glisten/socket/options import glisten/transport.{type Transport} /// Your provided loop function will receive these message types as the /// second argument. pub type Message(user_message) { /// These are messages received from the socket Packet(BitArray) /// These are any messages received from the selector returned from `on_init` User(user_message) } /// This is used to describe the connecting client's IP address. pub type IpAddress { IpV4(Int, Int, Int, Int) IpV6(Int, Int, Int, Int, Int, Int, Int, Int) } pub type Socket = InternalSocket pub type SocketReason = InternalSocketReason pub type ConnectionInfo { ConnectionInfo(port: Int, ip_address: IpAddress) } /// Returns the user-provided port or the OS-assigned value if 0 was provided. pub fn get_server_info( listener: process.Name(listener.Message), timeout: Int, ) -> ConnectionInfo { let listener = process.named_subject(listener) let state = process.call(listener, timeout, listener.Info) ConnectionInfo(state.port, convert_ip_address(state.ip_address)) } /// This type holds useful bits of data for the active connection. pub type Connection(user_message) { Connection( socket: Socket, /// This provides a uniform interface for both TCP and TLS methods. transport: Transport, subject: Subject(handler.Message(user_message)), ) } @internal pub fn convert_ip_address(ip: options.IpAddress) -> IpAddress { case ip { options.IpV4(a, b, c, d) -> IpV4(a, b, c, d) options.IpV6(a, b, c, d, e, f, g, h) -> IpV6(a, b, c, d, e, f, g, h) } } /// Convenience function for convert an `IpAddress` type into a string. It will /// convert IPv6 addresses to the canonical short-hand (ie. loopback is ::1). pub fn ip_address_to_string(address: IpAddress) -> String { case address { IpV4(a, b, c, d) -> [a, b, c, d] |> list.map(int.to_string) |> string.join(".") IpV6(a, b, c, d, e, f, g, h) -> { let fields = [a, b, c, d, e, f, g, h] case ipv6_zeros(fields, 0, 0, 0, 0) { Error(_) -> join_ipv6_fields(fields) Ok(#(start, end)) -> join_ipv6_fields(list.take(fields, start)) <> "::" <> join_ipv6_fields(list.drop(fields, end)) } |> string.lowercase } } } fn join_ipv6_fields(fields) { list.map(fields, int.to_base16) |> string.join(":") } /// Finds the longest sequence of consecutive all-zero fields in an IPv6. /// If the address contains multiple runs of all-zero fields of the same size, /// it is the leftmost that is compressed. /// /// This returns the start & end indices of the compressed zeros. fn ipv6_zeros(fields, pos, len, max_start, max_len) -> Result(#(Int, Int), Nil) { case fields { [] if max_len > 1 -> Ok(#(max_start, max_start + max_len)) [] -> Error(Nil) [x, ..xs] if x == 0 -> { let len = len + 1 case len > max_len { // Biggest sequence yet True -> ipv6_zeros(xs, pos + 1, len, pos + 1 - len, len) // Continue to grow current sequence False -> ipv6_zeros(xs, pos + 1, len, max_start, max_len) } } // Continue to search for zeros [_, ..xs] -> ipv6_zeros(xs, pos + 1, 0, max_start, max_len) } } /// 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. pub fn get_connection_info( conn: Connection(user_message), ) -> Result(ConnectionInfo, Nil) { transport.peername(conn.transport, conn.socket) |> result.map(fn(pair) { ConnectionInfo(pair.1, convert_ip_address(pair.0)) }) } /// Sends a BytesTree message over the socket using the active transport pub fn send( conn: Connection(user_message), msg: BytesTree, ) -> Result(Nil, SocketReason) { transport.send(conn.transport, conn.socket, msg) } pub opaque type Next(user_state, user_message) { Continue(state: user_state, selector: Option(Selector(user_message))) NormalStop AbnormalStop(String) } pub fn continue(state: user_state) -> Next(user_state, user_message) { Continue(state, None) } pub fn with_selector( next: Next(user_state, user_message), selector: Selector(user_message), ) -> Next(user_state, user_message) { case next { Continue(state, _) -> Continue(state, Some(selector)) stop -> stop } } pub fn stop() -> Next(user_state, user_message) { NormalStop } pub fn stop_abnormal(reason: String) -> Next(user_state, user_message) { AbnormalStop(reason) } @internal pub fn convert_next( next: Next(state, user_message), ) -> handler.Next(state, user_message) { case next { Continue(state, selector) -> handler.Continue(state, selector) NormalStop -> handler.NormalStop AbnormalStop(reason) -> handler.AbnormalStop(reason) } } @internal pub fn map_selector( next: Next(state, user_message), 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) AbnormalStop(reason) -> AbnormalStop(reason) NormalStop -> NormalStop } } /// This is the shape of the function you need to provide for the `handler` /// argument to `start`. pub type Loop(state, user_message) = fn(state, Message(user_message), Connection(user_message)) -> Next(state, Message(user_message)) pub opaque type Builder(state, user_message) { Builder( interface: options.Interface, on_init: fn(Connection(user_message)) -> #(state, Option(Selector(user_message))), loop: Loop(state, user_message), on_close: Option(fn(state) -> Nil), pool_size: Int, http2_support: Bool, ipv6_support: Bool, tls_options: Option(options.TlsCerts), listener_name: Option(process.Name(listener.Message)), connection_factory_name: Option( process.Name( factory.Message(Socket, Subject(handler.Message(user_message))), ), ), active_state: options.ActiveState, ) } fn map_user_selector( selector: Selector(Message(user_message)), ) -> Selector(handler.LoopMessage(user_message)) { process.map_selector(selector, fn(value) { case value { Packet(msg) -> handler.Packet(msg) User(msg) -> handler.Custom(msg) } }) } fn convert_loop( loop: Loop(state, user_message), ) -> handler.Loop(state, user_message) { fn(data, msg, conn: handler.Connection(user_message)) { let conn = Connection(conn.socket, conn.transport, conn.sender) let message = case msg { handler.Packet(msg) -> Packet(msg) handler.Custom(msg) -> User(msg) } case loop(data, message, conn) { Continue(data, selector) -> case selector { Some(selector) -> handler.continue(data) |> handler.with_selector(map_user_selector(selector)) _ -> handler.continue(data) } NormalStop -> handler.stop() AbnormalStop(reason) -> handler.stop_abnormal(reason) } } } fn convert_on_init( on_init: fn(Connection(user_message)) -> #(state, Option(Selector(user_message))), ) -> fn(handler.Connection(user_message)) -> #(state, Option(Selector(user_message))) { fn(conn: handler.Connection(user_message)) { let connection = Connection( subject: conn.sender, socket: conn.socket, transport: conn.transport, ) on_init(connection) } } /// 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. pub fn new( on_init: fn(Connection(user_message)) -> #(state, Option(Selector(user_message))), loop: Loop(state, user_message), ) -> Builder(state, user_message) { Builder( interface: options.Loopback, on_init: on_init, loop: loop, on_close: None, pool_size: 10, http2_support: False, ipv6_support: False, tls_options: None, listener_name: None, connection_factory_name: None, active_state: options.Once, ) } /// Adds a function to the handler to be called when the connection is closed. pub fn with_close( builder: Builder(state, user_message), on_close: fn(state) -> Nil, ) -> Builder(state, user_message) { Builder(..builder, on_close: Some(on_close)) } /// Modify the size of the acceptor pool pub fn with_pool_size( builder: Builder(state, user_message), size: Int, ) -> Builder(state, user_message) { Builder(..builder, pool_size: size) } /// Sets the ALPN supported protocols to include HTTP/2. It's currently being /// exposed only for `mist` to provide this support. For a TCP library, you /// definitely do not need it. @internal pub fn with_http2( builder: Builder(state, user_message), ) -> Builder(state, user_message) { Builder(..builder, http2_support: True) } /// This sets the interface for `glisten` to listen on. It accepts the following /// strings: "localhost", valid IPv4 addresses (i.e. "127.0.0.1"), and valid /// IPv6 addresses (i.e. "::1"). If an invalid value is provided, this will /// panic. pub fn bind( builder: Builder(state, user_message), interface: String, ) -> Builder(state, user_message) { let address = case interface, parse_address(charlist.from_string(interface)) { "0.0.0.0", _ -> options.Any "localhost", _ | "127.0.0.1", _ -> options.Loopback _, Ok(address) -> options.Address(address) _, Error(_nil) -> panic as "Invalid interface provided: must be a valid IPv4/IPv6 address, or \"localhost\"" } Builder(..builder, interface: address) } /// By default, `glisten` listens on `localhost` only over IPv4. With an IPv4 /// address, you can call this builder method to also serve over IPv6 on that /// interface. If it is not supported, your application will crash. If you /// call this with an IPv6 interface specified, it will have no effect. pub fn with_ipv6( builder: Builder(state, user_message), ) -> Builder(state, user_message) { Builder(..builder, ipv6_support: True) } /// To use TLS, provide a path to a certficate and key file. pub fn with_tls( builder: Builder(state, user_message), certfile cert: String, keyfile key: String, ) -> Builder(state, user_message) { Builder(..builder, tls_options: Some(options.CertKeyFiles(cert, key))) } /// Set the server's `ActiveState` for flow control of received packets. /// Default is `Once`. Allowed are `Once`, `Active` and `Count(n)` where n > 1. pub fn with_active_state( builder: Builder(state, user_message), active_state: options.ActiveState, ) -> Builder(state, user_message) { case active_state { options.Once | options.Active -> Builder(..builder, active_state: active_state) options.Count(n) if n > 1 -> Builder(..builder, active_state: active_state) options.Count(_) -> panic as "Count shall be greater than 1" options.Passive -> panic as "You cannot set the server's `ActiveState` to `Passive`" } } @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)) } @internal pub fn with_connection_factory_name( builder: Builder(state, user_message), connection_factory_name: process.Name( factory.Message(socket.Socket, Subject(handler.Message(user_message))), ), ) -> Builder(state, user_message) { Builder(..builder, connection_factory_name: Some(connection_factory_name)) } /// Start the TCP server with the given handler on the provided port 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 options = [options.Ip(builder.interface)] |> list.append(case builder.ipv6_support { True -> [options.Ipv6] False -> [] }) |> list.append(case builder.tls_options { Some(opts) -> [options.CertKeyConfig(opts)] _ -> [] }) |> list.append(case builder.tls_options, builder.http2_support { Some(_), True -> [options.AlpnPreferredProtocols(["h2", "http/1.1"])] Some(_), False -> [options.AlpnPreferredProtocols(["http/1.1"])] None, _ -> [] }) let transport = case builder.tls_options { Some(_) -> transport.Ssl _ -> transport.Tcp } Pool( handler: convert_loop(builder.loop), name: connection_supervisor, pool_count: builder.pool_size, on_init: convert_on_init(builder.on_init), on_close: builder.on_close, transport:, active_state: builder.active_state, ) |> acceptor.start_pool(transport, port, options, listener_name) } @external(erlang, "glisten_ffi", "parse_address") fn parse_address(value: Charlist) -> Result(ip_address, Nil) /// Helper method for building a child specification for use in a supervision /// tree. pub fn supervised( handler: Builder(state, user_message), port: Int, ) -> ChildSpecification(supervisor.Supervisor) { supervision.supervisor(fn() { start(handler, port) }) }