////
////
//// #### IP Address
//// - [ip_address_to_string](#ip_address_to_string)
//// #### Information
//// - [get_client_info](#get_client_info)
//// - [get_server_info](#get_server_info)
//// #### Builder
//// - [new](#new)
//// - [bind](#bind)
//// - [bind_all](#bind_all)
//// - [listening](#listening)
//// - [listening_random](#listening_random)
//// - [enable_ipv6](#enable_ipv6)
//// - [enable_tls](#enable_tls)
//// - [with_name](#with_name)
//// - [quiet](#quiet)
//// - [idle_timeout](#idle_timeout)
//// - [on_start](#on_start)
//// - [on_crash](#on_crash)
//// #### Server
//// - [start](#start)
//// - [supervised](#supervised)
//// #### Request
//// - [read_body](#read_body)
//// - [stream_body](#stream_body)
//// #### Response
//// - [file](#file)
//// #### Websocket
//// - [upgrade_websocket](#upgrade_websocket)
//// - [send_binary_frame](#send_binary_frame)
//// - [send_text_frame](#send_text_frame)
//// - [continue](#continue)
//// - [continue_with_selector](#continue_with_selector)
//// - [stop](#stop)
//// - [stop_abnormal](#stop_abnormal)
// -----------------------------------------------------------------------------
// IMPORTS
// -----------------------------------------------------------------------------
import gleam/bit_array
import gleam/bytes_tree.{type BytesTree}
import gleam/dynamic
import gleam/erlang/process.{type Selector}
import gleam/http
import gleam/http/request.{type Request as HttpRequest}
import gleam/http/response.{type Response as HttpResponse}
import gleam/int
import gleam/io
import gleam/option.{type Option, None, Some}
import gleam/otp/actor
import gleam/otp/static_supervisor.{type Supervisor} as supervisor
import gleam/otp/supervision
import gleam/result
import gleam/string_tree.{type StringTree}
import gleam/yielder
import glisten
import glisten/internal/listener
import glisten/socket/options as glisten_options
import glisten/transport
// TODO: replace this once gramps changes are published
import ewe/internal/gramps/websocket as ws
import ewe/internal/file as file_
import ewe/internal/handler as handler_
import ewe/internal/http as http_
import ewe/internal/websocket as websocket_
// -----------------------------------------------------------------------------
// CONNECTION
// -----------------------------------------------------------------------------
/// Represents a default body stored inside a `Request` type. Contains
/// important information for retrieving the original request body or client's
/// information. Can be converted to a `BitArray` using `ewe.read_body`.
///
pub type Connection =
http_.Connection
// -----------------------------------------------------------------------------
// IP ADDRESS
// -----------------------------------------------------------------------------
/// Represents an IP address of a client/server.
///
pub type IpAddress {
IpV4(Int, Int, Int, Int)
IpV6(Int, Int, Int, Int, Int, Int, Int, Int)
}
/// Converts an `IpAddress` to a `String`.
///
pub fn ip_address_to_string(address address: IpAddress) -> String {
ewe_to_glisten_ip(address)
|> glisten.ip_address_to_string()
}
fn glisten_to_ewe_ip(ip: glisten.IpAddress) -> IpAddress {
case ip {
glisten.IpV4(n1, n2, n3, n4) -> IpV4(n1, n2, n3, n4)
glisten.IpV6(n1, n2, n3, n4, n5, n6, n7, n8) ->
IpV6(n1, n2, n3, n4, n5, n6, n7, n8)
}
}
fn glisten_options_to_ewe_ip(ip: glisten_options.IpAddress) -> IpAddress {
case ip {
glisten_options.IpV4(n1, n2, n3, n4) -> IpV4(n1, n2, n3, n4)
glisten_options.IpV6(n1, n2, n3, n4, n5, n6, n7, n8) ->
IpV6(n1, n2, n3, n4, n5, n6, n7, n8)
}
}
fn ewe_to_glisten_ip(ip: IpAddress) -> glisten.IpAddress {
case ip {
IpV4(n1, n2, n3, n4) -> glisten.IpV4(n1, n2, n3, n4)
IpV6(n1, n2, n3, n4, n5, n6, n7, n8) ->
glisten.IpV6(n1, n2, n3, n4, n5, n6, n7, n8)
}
}
// -----------------------------------------------------------------------------
// INFORMATION
// -----------------------------------------------------------------------------
/// Represents client/server information. Can be retrieved using
/// `ewe.get_client_info`/`ewe.get_server_info`.
///
pub type SocketAddress {
SocketAddress(ip: IpAddress, port: Int)
}
/// Attempts to get the client's socket address using request's
/// connection.
///
pub fn get_client_info(
connection connection: Connection,
) -> Result(SocketAddress, Nil) {
transport.peername(connection.transport, connection.socket)
|> result.map(fn(server_info) {
SocketAddress(glisten_options_to_ewe_ip(server_info.0), server_info.1)
})
}
/// Retrieves server's socket address. Requires the same name as the one used in
/// `ewe.with_name` and server to be started.
///
pub fn get_server_info(
listener_name name: process.Name(listener.Message),
) -> SocketAddress {
let server_info = glisten.get_server_info(name, 10_000)
let ip_address = glisten_to_ewe_ip(server_info.ip_address)
SocketAddress(ip: ip_address, port: server_info.port)
}
// -----------------------------------------------------------------------------
// RESPONSE
// -----------------------------------------------------------------------------
/// Represents possible response body options.
///
/// Types for direct usage:
/// - Regular data: `TextData`, `BytesData`, `BitsData`, `StringTreeData`,
/// `Empty`
/// - Chunked data: `ChunkedData`
///
/// Types that should not be used directly:
/// - `File`: see `ewe.file` to construct it.
/// - `WebsocketConnection`: used in `ewe.upgrade_websocket` for correct
/// WebSocket connection handling.
///
pub type ResponseBody {
/// Allows to set response body from a string.
///
TextData(String)
/// Allows to set response body from bytes.
///
BytesData(BytesTree)
/// Allows to set response body from bits.
///
BitsData(BitArray)
/// Allows to set response body from a string tree.
///
StringTreeData(StringTree)
/// Allows to set empty response body.
///
Empty
/// Allows to send response body in chunks with `chunked` transfer encoding.
///
ChunkedData(yielder.Yielder(BitArray))
/// Allows to set response body from a file more efficiently rather than
/// sending contents in regular data types.
///
File(descriptor: file_.IoDevice, offset: Int, size: Int)
/// Allows upgrading request to a WebSocket connection.
///
WebsocketConnection(MonitorSelector)
}
/// Used to monitor a WebSocket connection. Use `MonitorSelector` only when you
/// know what you are doing.
///
@internal
pub type MonitorSelector {
MonitorSelector(Selector(process.Down))
}
/// A convenient alias for a HTTP response with a `ResponseBody` as the body.
///
pub type Response =
HttpResponse(ResponseBody)
fn transform_response_body(resp: Response) -> HttpResponse(http_.ResponseBody) {
response.set_body(resp, case resp.body {
TextData(text) -> http_.TextData(text)
BytesData(bytes) -> http_.BytesData(bytes)
BitsData(bits) -> http_.BitsData(bits)
StringTreeData(string_tree) -> http_.StringTreeData(string_tree)
ChunkedData(yielder) -> http_.ChunkedData(yielder)
File(descriptor, offset, size) -> http_.File(descriptor, offset, size)
WebsocketConnection(MonitorSelector(selector)) ->
http_.WebsocketConnection(selector)
Empty -> http_.Empty
})
}
/// Possible errors that can occur when setting response body from a file.
///
pub type FileError {
/// File does not exist.
///
NoEntry
/// Missing permission for reading the file, or for searching one of the
/// parents directories.
///
NoAccess
/// The named file is a directory.
///
IsDirectory
/// Untypical file error.
///
UnknownFileError(dynamic.Dynamic)
}
fn internal_to_file_error(error: file_.FileError) -> FileError {
case error {
file_.Enoent -> NoEntry
file_.Eacces -> NoAccess
file_.Eisdir -> IsDirectory
file_.Eunknown(error) -> UnknownFileError(error)
}
}
/// Sets response body from file, sets `content-length` header.
///
pub fn file(
path: String,
offset offset: Option(Int),
limit limit: Option(Int),
) -> Result(ResponseBody, FileError) {
// TODO: handle invalid offset + limit?
case file_.open(path) {
Ok(file) ->
Ok(File(
file.descriptor,
offset: option.unwrap(offset, 0),
size: option.unwrap(limit, file.size),
))
Error(error) -> Error(internal_to_file_error(error))
}
}
// -----------------------------------------------------------------------------
// BUILDER
// -----------------------------------------------------------------------------
type Handler =
fn(Request) -> Response
type OnStart =
fn(http.Scheme, SocketAddress) -> Nil
/// Ewe's server builder. Contains all server configurations. Can be adjusted
/// with the following functions:
/// - `ewe.bind`
/// - `ewe.bind_all`
/// - `ewe.listening`
/// - `ewe.listening_random`
/// - `ewe.enable_ipv6`
/// - `ewe.enable_tls`
/// - `ewe.with_name`
/// - `ewe.on_start`
/// - `ewe.quiet`
/// - `ewe.on_crash`
/// - `ewe.idle_timeout`
///
pub opaque type Builder {
Builder(
handler: Handler,
port: Int,
interface: String,
ipv6: Bool,
tls: Option(#(String, String)),
on_start: OnStart,
on_crash: Response,
listener_name: process.Name(listener.Message),
idle_timeout: Int,
)
}
/// Creates new server builder with handler provided.
///
/// Default configuration:
/// - port: `8080`
/// - interface: `127.0.0.1`
/// - No ipv6 support
/// - No TLS support
/// - Default listener name for server information retrieval
/// - on_start: prints `Listening on ://:`
/// - on_crash: empty 500 response
/// - idle_timeout: connection is closed after 10_000ms of inactivity
///
pub fn new(handler: Handler) -> Builder {
Builder(
handler:,
port: 8080,
interface: "127.0.0.1",
ipv6: False,
tls: None,
on_start: fn(scheme, server) {
let address = case server.ip {
IpV6(..) -> "[" <> ip_address_to_string(server.ip) <> "]"
IpV4(..) -> ip_address_to_string(server.ip)
}
let url =
http.scheme_to_string(scheme)
<> "://"
<> address
<> ":"
<> int.to_string(server.port)
io.println("Listening on " <> url)
},
on_crash: response.new(500) |> response.set_body(Empty),
listener_name: process.new_name("glisten_listener"),
idle_timeout: 10_000,
)
}
/// Binds server to a specific interface. Crashes program if the interface is
/// invalid.
///
pub fn bind(builder: Builder, interface interface: String) -> Builder {
Builder(..builder, interface:)
}
/// Binds server to all interfaces.
///
pub fn bind_all(builder: Builder) -> Builder {
Builder(..builder, interface: "0.0.0.0")
}
/// Sets listening port for server.
///
pub fn listening(builder: Builder, port port: Int) -> Builder {
Builder(..builder, port:)
}
/// Sets listening port for server to a random port. Useful for testing.
///
pub fn listening_random(builder: Builder) -> Builder {
Builder(..builder, port: 0)
}
/// Enables IPv6 support.
///
pub fn enable_ipv6(builder: Builder) -> Builder {
Builder(..builder, ipv6: True)
}
/// Enables TLS support, requires certificate and key file.
///
pub fn enable_tls(
builder: Builder,
certificate_file certificate_file: String,
key_file key_file: String,
) -> Builder {
let cert = case file_.open(certificate_file) {
Ok(_) -> certificate_file
Error(_) -> panic as "Failed to find cert file"
}
let key = case file_.open(key_file) {
Ok(_) -> key_file
Error(_) -> panic as "Failed to find key file"
}
Builder(..builder, tls: Some(#(cert, key)))
}
/// Sets a custom process name for server information retrieval, allowing to
/// use `ewe.get_server_info` after the server starts.
///
pub fn with_name(
builder: Builder,
name: process.Name(listener.Message),
) -> Builder {
Builder(..builder, listener_name: name)
}
/// Sets a custom handler that will be called after server starts.
///
pub fn on_start(
builder: Builder,
on_start: fn(http.Scheme, SocketAddress) -> Nil,
) -> Builder {
Builder(..builder, on_start:)
}
/// Sets an empty `on_start` function.
///
pub fn quiet(builder: Builder) -> Builder {
Builder(..builder, on_start: fn(_, _) { Nil })
}
/// Sets a custom response that will be sent when server crashes.
///
pub fn on_crash(builder: Builder, on_crash: Response) -> Builder {
Builder(..builder, on_crash:)
}
/// Sets a custom idle timeout in milliseconds for connections. If
/// provided timeout is less than 0, 10_000ms will be used instead.
///
pub fn idle_timeout(builder: Builder, idle_timeout: Int) -> Builder {
case idle_timeout {
idle_timeout if idle_timeout >= 0 -> Builder(..builder, idle_timeout:)
_ -> Builder(..builder, idle_timeout: 10_000)
}
}
// -----------------------------------------------------------------------------
// SERVER
// -----------------------------------------------------------------------------
/// Starts the server with the provided configuration.
///
pub fn start(
builder: Builder,
) -> Result(actor.Started(Supervisor), actor.StartError) {
let handler = fn(req) { transform_response_body(builder.handler(req)) }
let on_crash = transform_response_body(builder.on_crash)
let glisten_supervisor =
glisten.new(
handler_.init,
handler_.loop(handler, on_crash, builder.idle_timeout),
)
|> glisten.bind(builder.interface)
|> fn(glisten_builder) {
case builder.ipv6 {
True -> glisten.with_ipv6(glisten_builder)
False -> glisten_builder
}
}
|> fn(glisten_builder) {
case builder.tls {
Some(#(cert, key)) -> glisten.with_tls(glisten_builder, cert, key)
None -> glisten_builder
}
}
// https://github.com/rawhat/glisten/blob/master/src/glisten.gleam#L359
|> glisten.start_with_listener_name(builder.port, builder.listener_name)
|> result.map(fn(started) {
let scheme = case builder.tls {
Some(#(_, _)) -> http.Https
None -> http.Http
}
let server_info = glisten.get_server_info(builder.listener_name, 10_000)
let ip_address = glisten_to_ewe_ip(server_info.ip_address)
let server = SocketAddress(ip: ip_address, port: server_info.port)
builder.on_start(scheme, server)
started
})
let glisten_child = supervision.supervisor(fn() { glisten_supervisor })
supervisor.new(supervisor.OneForAll)
|> supervisor.add(glisten_child)
|> supervisor.start()
}
/// Creates a supervisor with the provided configuration that is a child of a
/// supervision tree.
///
pub fn supervised(
builder: Builder,
) -> supervision.ChildSpecification(supervisor.Supervisor) {
supervision.supervisor(fn() { start(builder) })
}
// -----------------------------------------------------------------------------
// REQUEST
// -----------------------------------------------------------------------------
/// Possible errors that can occur when reading a body.
///
pub type BodyError {
/// Body is larger than the provided limit.
BodyTooLarge
/// Body is malformed.
InvalidBody
}
/// A convenient alias for a HTTP request with a `Connection` as the body.
///
pub type Request =
HttpRequest(Connection)
/// Reads body from the request. If request body is malformed, `InvalidBody`
/// error is returned. On success, returns a request with body converted to
/// `BitArray`.
///
/// - When `transfer-encoding` header set as `chunked`, `BodyTooLarge` error is
/// returned if accumulated body is larger than `size_limit`.
/// - Ensures that `content-length` is in `size_limit` scope.
///
pub fn read_body(
req: Request,
bytes_limit bytes_limit: Int,
) -> Result(HttpRequest(BitArray), BodyError) {
case http_.read_body(req, bytes_limit) {
Ok(req) -> Ok(req)
Error(http_.BodyTooLarge) -> Error(BodyTooLarge)
Error(_) -> Error(InvalidBody)
}
}
/// A convenient alias for a consumer that reads `N` amount of bytes from the
/// request body stream.
///
pub type Consumer =
fn(Int) -> Result(Stream, BodyError)
/// Used to track the progress of reading the request body stream.
///
pub type Stream {
/// Chunk of data has been consumed.
///
Consumed(data: BitArray, next: Consumer)
/// Signifies that the request body stream has been fully consumed.
///
Done
}
/// Returns the consumer function that reads `N` amount of bytes from the
/// request body stream.
///
pub fn stream_body(req: Request) -> Result(Consumer, BodyError) {
case http_.stream_body(req) {
Ok(consumer) -> Ok(consumer_adapter(consumer))
Error(_) -> Error(InvalidBody)
}
}
fn consumer_adapter(
internal_consumer: fn(Int) -> Result(http_.Stream, http_.ParseError),
) -> Consumer {
fn(size) {
case internal_consumer(size) {
Ok(http_.Done) -> Ok(Done)
Ok(http_.Consumed(data, next)) -> {
Ok(Consumed(data, consumer_adapter(next)))
}
Error(_) -> Error(InvalidBody)
}
}
}
// -----------------------------------------------------------------------------
// WEBSOCKET
// -----------------------------------------------------------------------------
/// Represents a WebSocket connection between a client and a server.
///
pub type WebsocketConnection =
websocket_.WebsocketConnection
/// Represents an instruction on how WebSocket connection should proceed.
///
/// - continue processing the WebSocket connection.
/// - continue processing the WebSocket connection with selector for custom
/// messages.
/// - stop the WebSocket connection.
/// - stop the WebSocket connection with abnormal reason.
///
pub opaque type Next(user_state, user_message) {
Continue(user_state, Option(Selector(user_message)))
NormalStop
AbnormalStop(reason: String)
}
/// Instructs WebSocket connection to continue processing.
///
pub fn continue(user_state: user_state) -> Next(user_state, user_message) {
Continue(user_state, None)
}
/// Instructs WebSocket connection to continue processing, including selector
/// for custom messages.
///
pub fn continue_with_selector(
user_state: user_state,
selector: Selector(user_message),
) -> Next(user_state, user_message) {
Continue(user_state, Some(selector))
}
/// Instructs WebSocket connection to stop.
///
pub fn stop() -> Next(user_state, user_message) {
NormalStop
}
/// Instructs WebSocket connection to stop with abnormal reason.
///
pub fn stop_abnormal(reason: String) -> Next(user_state, user_message) {
AbnormalStop(reason)
}
fn to_internal_next(
next: Next(user_state, user_message),
) -> websocket_.WebsocketNext(user_state, user_message) {
case next {
Continue(user_state, selector) -> websocket_.Continue(user_state, selector)
NormalStop -> websocket_.NormalStop
AbnormalStop(reason) -> websocket_.AbnormalStop(reason)
}
}
/// Represents a WebSocket message received from the client.
///
pub type WebsocketMessage(user_message) {
/// Indicate that text frame has been received.
///
Text(String)
/// Indicate that binary frame has been received.
///
Binary(BitArray)
/// Indicate that user message has been received from WebSocket selector.
///
User(user_message)
}
fn transform_websocket_message(
message: websocket_.WebsocketMessage(user_message),
) -> Result(WebsocketMessage(user_message), Nil) {
// NOTE: see "https://github.com/rawhat/gramps/pull/7"
case message {
websocket_.WebsocketFrame(ws.Data(frame)) -> {
ws.match_data_frame(
frame,
on_text: fn(payload, _) {
bit_array.to_string(payload) |> result.map(Text)
},
on_binary: fn(payload, _) { Ok(Binary(payload)) },
)
}
websocket_.UserMessage(user_message) -> Ok(User(user_message))
_ -> Error(Nil)
}
}
/// Upgrade request to a WebSocket connection. If the initial request is not
/// valid for WebSocket upgrade, 400 response is sent.
///
/// `on_init` function is called once process that handles WebSocket connection
/// is initialized. It must return a tuple with initial state and selector for
/// custom messages. If there is no custom messages, user can pass the same
/// selector from the argument
///
/// `handler` function is called for every WebSocket message received. It must
/// return instruction on how WebSocket connection should proceed.
///
/// `on_close` function is called when WebSocket process is going to be stopped.
///
pub fn upgrade_websocket(
req: Request,
on_init on_init: fn(WebsocketConnection, Selector(user_message)) ->
#(user_state, Selector(user_message)),
handler handler: fn(
WebsocketConnection,
user_state,
WebsocketMessage(user_message),
) ->
Next(user_state, user_message),
on_close on_close: fn(WebsocketConnection, user_state) -> Nil,
) -> Response {
let handler = fn(conn, state, msg) {
transform_websocket_message(msg)
|> result.map(handler(conn, state, _))
|> result.unwrap(continue(state))
|> to_internal_next()
}
let transport = req.body.transport
let socket = req.body.socket
let resp = {
use #(extensions, permessage_deflate) <- result.try(
http_.upgrade_websocket(req, transport, socket)
|> result.replace_error(response.new(400) |> response.set_body(Empty)),
)
use selector <- result.try(
websocket_.start(
transport,
socket,
on_init,
handler,
on_close,
extensions,
permessage_deflate,
)
|> result.replace_error(response.new(500) |> response.set_body(Empty)),
)
response.new(500)
|> response.set_body(WebsocketConnection(MonitorSelector(selector)))
|> Ok
}
result.unwrap_both(resp)
}
/// Sends a binary frame to the websocket client.
///
pub fn send_binary_frame(
conn: WebsocketConnection,
bits: BitArray,
) -> Result(Nil, glisten.SocketReason) {
websocket_.send_frame(
ws.encode_binary_frame,
conn.transport,
conn.socket,
conn.deflate,
bits,
)
}
/// Sends a text frame to the websocket client.
///
pub fn send_text_frame(
conn: WebsocketConnection,
text: String,
) -> Result(Nil, glisten.SocketReason) {
websocket_.send_frame(
ws.encode_text_frame,
conn.transport,
conn.socket,
conn.deflate,
text,
)
}