import gleam/bytes_builder.{type BytesBuilder} import gleam/dynamic.{type Dynamic} import gleam/erlang/process.{type Selector, type Subject} import gleam/list import gleam/option.{type Option, None, Some} import gleam/otp/actor import gleam/otp/supervisor import gleam/result 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.{Certfile, Keyfile} import glisten/ssl import glisten/transport.{type Transport} /// Reasons that `serve` might fail pub type StartError { ListenerClosed ListenerTimeout AcceptorTimeout AcceptorFailed(process.ExitReason) AcceptorCrashed(Dynamic) SystemError(SocketReason) } /// Your provided loop function with receive these message types as the /// first 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 /// This holds information about the server. Returned by the `start_server` / /// `start_ssl_server` methods, it will allow you to get access to an /// OS-assigned port. Eventually, it will be used for graceful shutdown, and /// potentially other information. pub opaque type Server { Server( listener: Subject(listener.Message), supervisor: Subject(supervisor.Message), transport: Transport, ) } 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( server: Server, timeout: Int, ) -> Result(ConnectionInfo, process.CallError(listener.State)) { process.try_call(server.listener, listener.Info, timeout) |> result.map(fn(state) { ConnectionInfo(state.port, convert_ip_address(state.ip_address)) }) } /// Gets the underlying supervisor `Subject` from the `Server`. pub fn get_supervisor(server: Server) -> Subject(supervisor.Message) { server.supervisor } /// 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 SSL methods. transport: Transport, subject: Subject(handler.Message(user_message)), ) } @internal pub fn convert_ip_address(ip: transport.IpAddress) -> IpAddress { case ip { transport.IpV4(a, b, c, d) -> IpV4(a, b, c, d) transport.IpV6(a, b, c, d, e, f, g, h) -> IpV6(a, b, c, d, e, f, g, h) } } /// 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_client_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 BytesBuilder message over the socket using the active transport pub fn send( conn: Connection(user_message), msg: BytesBuilder, ) -> Result(Nil, SocketReason) { transport.send(conn.transport, conn.socket, msg) } /// This is the shape of the function you need to provide for the `handler` /// argument to `serve(_ssl)`. pub type Loop(user_message, data) = fn(Message(user_message), data, Connection(user_message)) -> actor.Next(Message(user_message), data) pub opaque type Handler(user_message, data) { Handler( on_init: fn(Connection(user_message)) -> #(data, Option(Selector(user_message))), loop: Loop(user_message, data), on_close: Option(fn(data) -> Nil), pool_size: Int, http2_support: Bool, ) } 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(user_message, data), ) -> handler.Loop(user_message, data) { fn(msg, data, conn: handler.Connection(user_message)) { let conn = Connection(conn.socket, conn.transport, conn.sender) case msg { handler.Packet(msg) -> { case loop(Packet(msg), data, conn) { actor.Continue(data, selector) -> actor.Continue(data, option.map(selector, map_user_selector)) actor.Stop(reason) -> actor.Stop(reason) } } handler.Custom(msg) -> { case loop(User(msg), data, conn) { actor.Continue(data, selector) -> actor.Continue(data, option.map(selector, map_user_selector)) actor.Stop(reason) -> actor.Stop(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 handler( on_init: fn(Connection(user_message)) -> #(data, Option(Selector(user_message))), loop: Loop(user_message, data), ) -> Handler(user_message, data) { Handler( on_init: on_init, loop: loop, on_close: None, pool_size: 10, http2_support: False, ) } /// Adds a function to the handler to be called when the connection is closed. pub fn with_close( handler: Handler(user_message, data), on_close: fn(data) -> Nil, ) -> Handler(user_message, data) { Handler(..handler, on_close: Some(on_close)) } /// Modify the size of the acceptor pool pub fn with_pool_size( handler: Handler(user_message, data), size: Int, ) -> Handler(user_message, data) { Handler(..handler, 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. pub fn with_http2( handler: Handler(user_message, data), ) -> Handler(user_message, data) { Handler(..handler, http2_support: True) } /// Start the TCP server with the given handler on the provided port pub fn serve( handler: Handler(user_message, data), port: Int, ) -> Result(Subject(supervisor.Message), StartError) { start_server(handler, port) |> result.map(get_supervisor) } /// Start the SSL server with the given handler on the provided port. The key /// and cert files must be provided, valid, and readable by the current user. pub fn serve_ssl( handler: Handler(user_message, data), port port: Int, certfile certfile: String, keyfile keyfile: String, ) -> Result(Subject(supervisor.Message), StartError) { start_ssl_server(handler, port, certfile, keyfile) |> result.map(get_supervisor) } /// Starts a TCP server and returns the `Server` construct. This is useful if /// you need access to the port. In the future, it will also allow graceful /// shutdown. There may also be other metadata attached to this return value. pub fn start_server( handler: Handler(user_message, data), port: Int, ) -> Result(Server, StartError) { let return = process.new_subject() let selector = process.new_selector() |> process.selecting(return, fn(subj) { subj }) Pool( handler: convert_loop(handler.loop), pool_count: handler.pool_size, on_init: convert_on_init(handler.on_init), on_close: handler.on_close, transport: transport.Tcp, ) |> acceptor.start_pool(transport.Tcp, port, [], return) |> result.map_error(fn(err) { case err { actor.InitTimeout -> AcceptorTimeout actor.InitFailed(reason) -> AcceptorFailed(reason) actor.InitCrashed(reason) -> AcceptorCrashed(reason) } }) |> result.then(fn(pool) { process.select(selector, 1500) |> result.map(fn(listener) { Server(listener: listener, supervisor: pool, transport: transport.Tcp) }) |> result.replace_error(AcceptorTimeout) }) } /// Starts an SSL server and returns the `Server` construct. This is useful if /// you need access to the port. In the future, it will also allow graceful /// shutdown. There may also be other metadata attached to this return value. pub fn start_ssl_server( handler: Handler(user_message, data), port port: Int, certfile certfile: String, keyfile keyfile: String, ) -> Result(Server, StartError) { let assert Ok(_nil) = ssl.start() let ssl_options = [Certfile(certfile), Keyfile(keyfile)] let protocol_options = case handler.http2_support { True -> [options.AlpnPreferredProtocols(["h2", "http/1.1"])] False -> [options.AlpnPreferredProtocols(["http/1.1"])] } let return = process.new_subject() let selector = process.new_selector() |> process.selecting(return, fn(subj) { subj }) Pool( handler: convert_loop(handler.loop), pool_count: handler.pool_size, on_init: convert_on_init(handler.on_init), on_close: handler.on_close, transport: transport.Ssl, ) |> acceptor.start_pool( transport.Ssl, port, list.concat([ssl_options, protocol_options]), return, ) |> result.map_error(fn(err) { case err { actor.InitTimeout -> AcceptorTimeout actor.InitFailed(reason) -> AcceptorFailed(reason) actor.InitCrashed(reason) -> AcceptorCrashed(reason) } }) |> result.then(fn(pool) { process.select(selector, 1500) |> result.map(fn(listener) { Server(listener: listener, supervisor: pool, transport: transport.Tcp) }) |> result.replace_error(AcceptorTimeout) }) }