Make it safe & Gleam native

Author: alii

README.md

@@ -2,18 +2,13 @@
 
 Gleam bindings to [Manifold](https://github.com/discord/manifold) - an Elixir library for fast message passing between BEAM nodes.
 
-> [!WARNING]
-> This library depends on internal implementation details of Gleam's `erlang/process` module. It is not officially supported and may break with future Gleam releases. Use at your own risk.
-
 ## What is Manifold?
 
 Manifold is an Elixir library developed by Discord that optimizes sending the same message to many processes. Instead of sending messages sequentially (which can be slow with thousands of processes), Manifold uses a divide-and-conquer approach that distributes the work across multiple sender processes, achieving much better performance at scale.
 
 ## Installation
 
-This library is intentionally not published to hex, because it depends on internal implementation details of Gleam's `erlang/process` module. If Gleam opens up the Subject type, we can implement this library in a more official way and publish to hex.
-
-For now, please depend on it as a git dependency in your `gleam.toml`:
+Add to your `gleam.toml` as a git dependency:
 
 ```toml
 [dependencies]
@@ -22,31 +17,23 @@ gleam_manifold = { git = "[email protected]:otters/gleam_manifold.git", ref = "<com
 
 ## Usage
 
-### Type-safe sending to Subjects
+### Creating and using Subjects
 
-The primary way to use this library is with Gleam's type-safe `Subject` system:
+This library provides its own Subject type for type-safe message passing:
 
 ```gleam
 import gleam/erlang/process
 import gleam_manifold as manifold
 
 pub fn example() {
-  let subject = process.new_subject()
+  let subject = manifold.new_subject()
+  let pid = process.self()
 
   // Send a message through Manifold
-  manifold.send(subject, "Hello world")
-}
-```
-
-### Sending to PIDs
-
-For cases where you need to send to raw PIDs (not type-safe):
-
-```gleam
-import gleam_manifold as manifold
+  manifold.send(pid, subject, "Hello world")
 
-pub fn send_to_pid(pid: process.Pid) {
-  manifold.send_pid(pid, "Hello")
+  // Receive the message
+  let assert Ok(message) = manifold.receive(subject, 1000)
 }
 ```
 
@@ -55,24 +42,18 @@ pub fn send_to_pid(pid: process.Pid) {
 Send the same message to multiple processes at once:
 
 ```gleam
+import gleam/erlang/process
 import gleam_manifold as manifold
 
-pub fn broadcast_join(pids: List(process.Pid), name: String) {
-  manifold.send_multi_pid(pids, "Hello, " <> name)
+pub fn broadcast(pids: List(process.Pid), message: String) {
+  let subject = manifold.new_subject()
+  manifold.send_multi(pids, subject, message)
 }
 ```
 
-## Implementation Notes
-
-### Internals Hack
-
-This library uses internals of Gleam's `erlang/process` implementation by unwrapping the opaque `Subject` type in Erlang. This is technically a hack since `Subject` is meant to be opaque, but it's necessary to extract the PID and reference tag needed for sending to Gleam's Subject type. The implementation pattern matches on the internal `{subject, Pid, Ref}` tuple structure in `gleam_manifold_ffi.erl`.
-
-### Current Limitations
-
-1. **No options parameter support**: The Elixir Manifold library supports an options parameter for tuning performance characteristics (like partition size). This Gleam wrapper doesn't currently expose these options, though this could be added in the future.
+## Current Limitations
 
-2. **No named subjects support**: Named subjects (registered process names) are not currently supported. The library will fail with an assertion error if you try to use a named subject. This is fixable but hasn't been implemented yet.
+**No options parameter support**: The Elixir Manifold library supports an options parameter for tuning performance characteristics (like partition size). This Gleam wrapper doesn't currently expose these options, though this could be added in the future.
 
 ## Testing
 

src/gleam_manifold.gleam

@@ -1,18 +1,26 @@
 import gleam/erlang/process
 import gleam/erlang/reference
 
-pub fn send(subject: process.Subject(message), message: message) -> Nil {
-  let assert Ok(#(pid, ref)) = unwrap_subject(subject)
-    as "Named subjects are currently unsupported in gleam_manifold.send()"
-  manifold_send_subject(pid, #(ref, message))
-  Nil
+pub opaque type Subject(message) {
+  Subject(tag: reference.Reference)
+}
+
+type Message(message) =
+  #(reference.Reference, message)
+
+pub fn new_subject() -> Subject(message) {
+  Subject(reference.new())
 }
 
 /// You almost certainly do not want to use this. 
 /// Absolutely prefer to use the type-safe variant of
 /// this function `manifold.send(subject, message)`
-pub fn send_pid(pid: process.Pid, message: message) -> Nil {
-  manifold_send_raw(pid, message)
+pub fn send(
+  pid: process.Pid,
+  subject: Subject(message),
+  message: message,
+) -> Nil {
+  manifold_send(pid, #(subject.tag, message))
   Nil
 }
 
@@ -26,26 +34,28 @@ pub fn send_pid(pid: process.Pid, message: message) -> Nil {
 /// impossible because they are unique. You would have to use something
 /// like `process.select_other` to take advantage of receiving values
 /// with this function still within Gleam.
-pub fn send_multi_pid(pids: List(process.Pid), message: message) -> Nil {
-  manifold_send_multi(pids, message)
+pub fn send_multi(
+  pids: List(process.Pid),
+  subject: Subject(message),
+  message: message,
+) -> Nil {
+  manifold_send_multi(pids, #(subject.tag, message))
   Nil
 }
 
 type DoNotLeak
 
-@external(erlang, "gleam_manifold_ffi", "unwrap_subject")
-fn unwrap_subject(
-  subject: process.Subject(message),
-) -> Result(#(process.Pid, reference.Reference), Nil)
+@external(erlang, "gleam_manifold_ffi", "receive")
+pub fn receive(subject: Subject(message), timeout: Int) -> Result(message, Nil)
 
-@external(erlang, "Elixir.Manifold", "send")
-fn manifold_send_subject(
-  pid: process.Pid,
-  message: #(reference.Reference, message),
-) -> DoNotLeak
+@external(erlang, "gleam_manifold_ffi", "receive")
+pub fn receive_forever(subject: Subject(message)) -> message
 
 @external(erlang, "Elixir.Manifold", "send")
-fn manifold_send_raw(pid: process.Pid, message: message) -> DoNotLeak
+fn manifold_send(pid: process.Pid, message: Message(message)) -> DoNotLeak
 
 @external(erlang, "Elixir.Manifold", "send")
-fn manifold_send_multi(pid: List(process.Pid), message: message) -> DoNotLeak
+fn manifold_send_multi(
+  pid: List(process.Pid),
+  message: Message(message),
+) -> DoNotLeak

src/gleam_manifold_ffi.erl

@@ -1,7 +1,14 @@
 -module(gleam_manifold_ffi).
--export([unwrap_subject/1]).
+-export(['receive'/1, 'receive'/2]).
 
-unwrap_subject({subject, Pid, Ref}) ->
-    {ok, {Pid, Ref}};
-unwrap_subject(_Subject) ->
-    {error, nil}.
+'receive'({subject, Ref}) ->
+    receive
+        {Ref, Message} -> Message
+    end.
+
+'receive'({subject, Ref}, Timeout) ->
+    receive
+        {Ref, Message} -> {ok, Message}
+    after Timeout ->
+        {error, nil}
+    end.

test/gleam_manifold_test.gleam

@@ -7,48 +7,51 @@ pub fn main() -> Nil {
 }
 
 pub fn simple_test() {
-  let subject = process.new_subject()
-  process.spawn(fn() { manifold.send(subject, "Hello world") })
-  assert process.receive(subject, 100) == Ok("Hello world")
+  let subject = manifold.new_subject()
+  let pid = process.self()
+  process.spawn(fn() { manifold.send(pid, subject, "Hello world") })
+  assert manifold.receive(subject, 100) == Ok("Hello world")
 }
 
 pub fn many_pids_test() {
-  let subject = process.new_subject()
+  let subject = manifold.new_subject()
+  let pid = process.self()
 
   let proc = fn() {
     let selector = process.new_selector() |> process.select_other(identity)
-    let assert Ok(result) = process.selector_receive(selector, 5)
-    process.send(subject, result)
+    let assert Ok(#(_ref, result)) = process.selector_receive(selector, 5)
+    manifold.send(pid, subject, result)
   }
 
   let a = process.spawn(proc)
   let b = process.spawn(proc)
 
-  manifold.send_pid(a, "hello from a")
-  manifold.send_pid(b, "hello from b")
+  manifold.send(a, subject, "hello from a")
+  manifold.send(b, subject, "hello from b")
 
-  assert process.receive(subject, 5) == Ok("hello from a")
-  assert process.receive(subject, 5) == Ok("hello from b")
-  assert process.receive(subject, 5) == Error(Nil)
+  assert manifold.receive(subject, 5) == Ok("hello from a")
+  assert manifold.receive(subject, 5) == Ok("hello from b")
+  assert manifold.receive(subject, 5) == Error(Nil)
 }
 
 pub fn multi_pids_test() {
-  let subject = process.new_subject()
+  let subject = manifold.new_subject()
+  let pid = process.self()
 
   let proc = fn() {
     let selector = process.new_selector() |> process.select_other(identity)
-    let assert Ok(result) = process.selector_receive(selector, 5)
-    process.send(subject, result)
+    let assert Ok(#(_ref, result)) = process.selector_receive(selector, 5)
+    manifold.send(pid, subject, result)
   }
 
   let a = process.spawn(proc)
   let b = process.spawn(proc)
 
-  manifold.send_multi_pid([a, b], "hello from both")
+  manifold.send_multi([a, b], subject, "hello from both")
 
-  assert process.receive(subject, 5) == Ok("hello from both")
-  assert process.receive(subject, 5) == Ok("hello from both")
-  assert process.receive(subject, 5) == Error(Nil)
+  assert manifold.receive(subject, 5) == Ok("hello from both")
+  assert manifold.receive(subject, 5) == Ok("hello from both")
+  assert manifold.receive(subject, 5) == Error(Nil)
 }
 
 @external(erlang, "gleam_erlang_ffi", "identity")