Packages
glisten
0.9.3
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/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)
}