feat: first commit

Author: howenyap

.github/workflows/test.yml

@@ -0,0 +1,23 @@
+name: test
+
+on:
+  push:
+    branches:
+      - master
+      - main
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: erlef/setup-beam@v1
+        with:
+          otp-version: "28"
+          gleam-version: "1.14.0"
+          rebar3-version: "3"
+          # elixir-version: "1"
+      - run: gleam deps download
+      - run: gleam test
+      - run: gleam format --check src test

.gitignore

@@ -0,0 +1,4 @@
+*.beam
+*.ez
+/build
+erl_crash.dump

README.md

@@ -0,0 +1,18 @@
+# Usage 
+```gleam
+import gleam/io
+import gleam/erlang/process
+import gleam/otp/actor
+import singleflight
+
+pub fn main() -> Nil {
+  let name = process.new_name("singleflight")
+  let config = singleflight.config(1_000, 1_000)
+  
+  let assert Ok(actor.Started(data: server, ..)) =
+    singleflight.start(config, name)
+    
+  let value = singleflight.fetch(server, "key", fn(key) { key <> "-value" })
+
+  io.debug(value)
+}

gleam.toml

@@ -0,0 +1,21 @@
+name = "singleflight"
+version = "1.0.0"
+
+# Fill out these fields if you intend to generate HTML documentation or publish
+# your project to the Hex package manager.
+#
+# description = ""
+# licences = ["Apache-2.0"]
+# repository = { type = "github", user = "", repo = "" }
+# links = [{ title = "Website", href = "" }]
+#
+# For a full reference of all the available options, you can have a look at
+# https://gleam.run/writing-gleam/gleam-toml/.
+
+[dependencies]
+gleam_stdlib = ">= 0.44.0 and < 2.0.0"
+gleam_otp = ">= 1.2.0 and < 2.0.0"
+gleam_erlang = ">= 1.3.0 and < 2.0.0"
+
+[dev-dependencies]
+gleeunit = ">= 1.0.0 and < 2.0.0"

manifest.toml

@@ -0,0 +1,15 @@
+# This file was generated by Gleam
+# You typically do not need to edit this file
+
+packages = [
+  { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" },
+  { name = "gleam_otp", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "BA6A294E295E428EC1562DC1C11EA7530DCB981E8359134BEABC8493B7B2258E" },
+  { name = "gleam_stdlib", version = "0.70.0", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "86949BF5D1F0E4AC0AB5B06F235D8A5CC11A2DFC33BF22F752156ED61CA7D0FF" },
+  { name = "gleeunit", version = "1.9.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "DA9553CE58B67924B3C631F96FE3370C49EB6D6DC6B384EC4862CC4AAA718F3C" },
+]
+
+[requirements]
+gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" }
+gleam_otp = { version = ">= 1.2.0 and < 2.0.0" }
+gleam_stdlib = { version = ">= 0.44.0 and < 2.0.0" }
+gleeunit = { version = ">= 1.0.0 and < 2.0.0" }

src/singleflight.gleam

@@ -0,0 +1,101 @@
+import gleam/dict
+import gleam/erlang/process
+import gleam/list
+import gleam/otp/actor
+
+pub type Config {
+  Config(initialisation_timeout_ms: Int, fetch_timeout_ms: Int)
+}
+
+pub type Message(k, v) {
+  Request(key: k, work: fn(k) -> v, caller: process.Subject(v))
+  Done(key: k, result: v)
+}
+
+pub opaque type Singleflight(k, v) {
+  Singleflight(subject: process.Subject(Message(k, v)), fetch_timeout_ms: Int)
+}
+
+type State(k, v) {
+  State(
+    in_flight: dict.Dict(k, List(process.Subject(v))),
+    self: process.Subject(Message(k, v)),
+  )
+}
+
+pub fn config(initialisation_timeout_ms: Int, fetch_timeout_ms: Int) -> Config {
+  Config(
+    initialisation_timeout_ms: initialisation_timeout_ms,
+    fetch_timeout_ms: fetch_timeout_ms,
+  )
+}
+
+pub fn start(
+  config: Config,
+  name: process.Name(Message(k, v)),
+) -> actor.StartResult(Singleflight(k, v)) {
+  let Config(initialisation_timeout_ms:, fetch_timeout_ms:) = config
+
+  actor.new_with_initialiser(initialisation_timeout_ms, fn(self) {
+    actor.initialised(State(in_flight: dict.new(), self: self))
+    |> actor.returning(Singleflight(
+      subject: self,
+      fetch_timeout_ms: fetch_timeout_ms,
+    ))
+    |> Ok
+  })
+  |> actor.on_message(handle_message)
+  |> actor.named(name)
+  |> actor.start
+}
+
+pub fn fetch(singleflight: Singleflight(k, v), key: k, work: fn(k) -> v) -> v {
+  let Singleflight(subject:, fetch_timeout_ms:) = singleflight
+
+  actor.call(subject, fetch_timeout_ms, fn(caller) {
+    Request(key: key, work: work, caller: caller)
+  })
+}
+
+fn handle_message(
+  state: State(k, v),
+  message: Message(k, v),
+) -> actor.Next(State(k, v), Message(k, v)) {
+  case message {
+    Request(key, work, caller) ->
+      case dict.get(state.in_flight, key) {
+        Ok(waiters) ->
+          actor.continue(
+            State(
+              ..state,
+              in_flight: dict.insert(state.in_flight, key, [caller, ..waiters]),
+            ),
+          )
+
+        Error(Nil) -> {
+          let self = state.self
+          process.spawn(fn() {
+            let result = work(key)
+            actor.send(self, Done(key: key, result: result))
+          })
+          actor.continue(
+            State(
+              ..state,
+              in_flight: dict.insert(state.in_flight, key, [caller]),
+            ),
+          )
+        }
+      }
+
+    Done(key, result) -> {
+      case dict.get(state.in_flight, key) {
+        Ok(waiters) ->
+          list.each(waiters, fn(waiter) { process.send(waiter, result) })
+        Error(Nil) -> Nil
+      }
+      actor.continue(
+        State(..state, in_flight: dict.delete(state.in_flight, key)),
+      )
+    }
+  }
+}

test/singleflight_test.gleam

@@ -0,0 +1,97 @@
+import gleam/erlang/process
+import gleam/otp/actor
+import gleeunit
+import singleflight
+
+pub fn main() -> Nil {
+  gleeunit.main()
+}
+
+pub fn same_key_deduplicates_to_one_request_test() {
+  let server = start_test_server("singleflight_test_same_key")
+  let started = process.new_subject()
+
+  let first_result = process.new_subject()
+  let second_result = process.new_subject()
+
+  let work = fn(key) {
+    let release = process.new_subject()
+    process.send(started, release)
+
+    let assert Ok("release") = process.receive(release, within: 1000)
+
+    key
+  }
+
+  process.spawn(fn() {
+    let result = singleflight.fetch(server, "foo", work)
+    process.send(first_result, result)
+  })
+
+  let assert Ok(first_release) = process.receive(started, within: 1000)
+
+  process.spawn(fn() {
+    let result = singleflight.fetch(server, "foo", work)
+    process.send(second_result, result)
+  })
+
+  let assert Error(Nil) = process.receive(started, within: 0)
+
+  process.send(first_release, "release")
+
+  let assert Ok("foo") = process.receive(first_result, within: 1000)
+  let assert Ok("foo") = process.receive(second_result, within: 1000)
+
+  let assert Error(Nil) = process.receive(started, within: 0)
+}
+
+pub fn different_keys_run_two_requests_test() {
+  let server = start_test_server("singleflight_test_different_keys")
+  let started = process.new_subject()
+
+  let first_result = process.new_subject()
+  let second_result = process.new_subject()
+
+  let work = fn(key) {
+    let release = process.new_subject()
+    process.send(started, #(key, release))
+
+    let assert Ok("release") = process.receive(release, within: 1000)
+
+    key
+  }
+
+  process.spawn(fn() {
+    let result = singleflight.fetch(server, "foo", work)
+    process.send(first_result, result)
+  })
+
+  process.spawn(fn() {
+    let result = singleflight.fetch(server, "bar", work)
+    process.send(second_result, result)
+  })
+
+  let assert Ok(#("foo", foo_release)) = process.receive(started, within: 1000)
+  let assert Ok(#("bar", bar_release)) = process.receive(started, within: 1000)
+  let assert Error(Nil) = process.receive(started, within: 0)
+
+  process.send(foo_release, "release")
+  process.send(bar_release, "release")
+
+  let assert Ok("foo") = process.receive(first_result, within: 1000)
+  let assert Ok("bar") = process.receive(second_result, within: 1000)
+
+  let assert Error(Nil) = process.receive(started, within: 0)
+}
+
+fn start_test_server(
+  prefix: String,
+) -> singleflight.Singleflight(String, String) {
+  let name = process.new_name(prefix)
+  let config = singleflight.config(1000, 1000)
+
+  let assert Ok(actor.Started(data: server, ..)) =
+    singleflight.start(config, name)
+
+  server
+}