Current section

Files

Jump to
glisten src glisten.gleam
Raw

src/glisten.gleam

import gleam/bit_builder.{type BitBuilder}
import gleam/dynamic.{type Dynamic}
import gleam/erlang/process.{type Selector, type Subject}
import gleam/option.{type Option, None, Some}
import gleam/result
import glisten/acceptor.{Pool}
import glisten/handler.{type ClientIp}
import glisten/socket.{type Socket, type SocketReason, Closed, Timeout}
import glisten/socket/transport.{type Transport}
import glisten/tcp
import glisten/ssl
import gleam/otp/actor
import gleam/otp/supervisor
import glisten/socket/options.{Certfile, Keyfile}
/// 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 type holds useful bits of data for the active connection.
pub type Connection(user_message) {
Connection(
/// This will be optionally a tuple for the IPv4 of the other end of the
/// socket
client_ip: ClientIp,
socket: Socket,
/// This provides a uniform interface for both TCP and SSL methods.
transport: Transport,
subject: Subject(handler.Message(user_message)),
)
}
/// Sends a BitBuilder message over the socket using the active transport
pub fn send(
conn: Connection(user_message),
msg: BitBuilder,
) -> Result(Nil, SocketReason) {
conn.transport.send(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() -> #(data, Option(Selector(user_message))),
loop: Loop(user_message, data),
on_close: Option(fn(data) -> Nil),
pool_size: Int,
)
}
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.client_ip, 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)
}
}
}
}
}
/// 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() -> #(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)
}
/// 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)
}
/// 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) {
port
|> tcp.listen([])
|> result.map_error(fn(err) {
case err {
Closed -> ListenerClosed
Timeout -> ListenerTimeout
err -> SystemError(err)
}
})
|> result.then(fn(socket) {
Pool(
socket,
convert_loop(handler.loop),
handler.pool_size,
handler.on_init,
handler.on_close,
transport.tcp(),
)
|> acceptor.start_pool
|> result.map_error(fn(err) {
case err {
actor.InitTimeout -> AcceptorTimeout
actor.InitFailed(reason) -> AcceptorFailed(reason)
actor.InitCrashed(reason) -> AcceptorCrashed(reason)
}
})
})
}
/// 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(Nil, StartError) {
let assert Ok(_nil) = ssl.start()
port
|> ssl.listen([Certfile(certfile), Keyfile(keyfile)])
|> result.map_error(fn(err) {
case err {
Closed -> ListenerClosed
Timeout -> ListenerTimeout
err -> SystemError(err)
}
})
|> result.then(fn(socket) {
Pool(
socket,
convert_loop(handler.loop),
handler.pool_size,
handler.on_init,
handler.on_close,
transport.ssl(),
)
|> acceptor.start_pool
|> result.map_error(fn(err) {
case err {
actor.InitTimeout -> AcceptorTimeout
actor.InitFailed(reason) -> AcceptorFailed(reason)
actor.InitCrashed(reason) -> AcceptorCrashed(reason)
}
})
})
|> result.replace(Nil)
}