Packages
glisten
7.0.0
9.0.1
9.0.0
9.0.0-rc1
8.0.3
8.0.2
8.0.1
8.0.0
8.0.0-rc1
7.0.1
7.0.0
6.0.0
5.0.0
4.0.0
3.0.0
2.0.0
1.0.0
0.11.0
0.10.2
0.10.1
0.10.0
0.9.3
0.9.2
0.9.1
0.9.0
0.8.2
0.8.1
0.8.0
0.7.0
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.1
0.5.0
0.4.0
0.3.2
0.3.1
0.3.0
0.2.2
0.2.1
0.2.0
0.1.3
0.1.2
0.1.1
0.1.0
a shiny Gleam TCP/TLS server
Current section
Files
Jump to
Current section
Files
src/glisten.gleam
import gleam/bytes_tree.{type BytesTree}
import gleam/dynamic.{type Dynamic}
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/supervisor
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.{Certfile, Keyfile}
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 will 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: 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_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 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)
}
/// 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(
interface: options.Interface,
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,
ipv6_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(
interface: options.Loopback,
on_init: on_init,
loop: loop,
on_close: None,
pool_size: 10,
http2_support: False,
ipv6_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)
}
/// 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(
handler: Handler(user_message, data),
interface: String,
) -> Handler(user_message, data) {
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\""
}
Handler(..handler, 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(
handler: Handler(user_message, data),
) -> Handler(user_message, data) {
Handler(..handler, ipv6_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 })
let options = case handler.ipv6_support {
True -> [options.Ip(handler.interface), options.Ipv6]
False -> [options.Ip(handler.interface)]
}
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, 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)
})
}
/// 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 base_options = [
options.Ip(handler.interface),
Certfile(certfile),
Keyfile(keyfile),
]
let default_options = case handler.ipv6_support {
True -> [options.Ipv6, ..base_options]
False -> base_options
}
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.flatten([default_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)
})
}
@external(erlang, "glisten_ffi", "parse_address")
fn parse_address(value: Charlist) -> Result(ip_address, Nil)