@spec ack(lease_handle()) :: :ok

Releases a tuple held under a {:monitor, :ack} lease.

handle is the third element returned by in/2 or inp/2 in that mode. Returns :ok whether or not the lease is still held, so acknowledging twice, or acknowledging one that has already expired, is harmless rather than an error.

@spec eval((-> Tuplex.Template.t())) :: {:ok, pid()}

Runs fun in a fresh process and writes its result into the space.

Returns {:ok, pid} immediately. This is Linda's eval: a way to say "the tuple is the result of this computation" without the caller waiting on it.

Examples

Tuplex.eval(fn -> {:report, expensive_calculation()} end)
#=> {:ok, #PID<0.123.0>}

@spec Tuplex.Template.template() in keyword() ::
  {:ok, Tuplex.Template.t()}
  | {:ok, Tuplex.Template.t(), lease_handle()}
  | {:error, :timeout}

Blocks until a tuple matching template is available, then removes and returns it.

Returns {:ok, tuple}, or {:error, :timeout} if the wait runs out.

Options

• :timeout — milliseconds to wait, or :infinity (the default). 0 does not register a waiter at all; it behaves exactly like inp/2, reporting {:error, :timeout} where inp/2 would say :empty.

• :lease — false by default. :monitor or {:monitor, :ack} hold the tuple against the calling process rather than removing it outright, so it comes back if that process dies without finishing. See below.

Exactly one blocked in/2 receives any given tuple. Every rd/2 waiting on a template that also matches is woken first, before this call consumes it, so a reader never ends up blocked on a tuple that has already been taken away.

Waiters are served in the order they arrived.

Leasing

Without a lease, a tuple taken by a process that then crashes is simply gone. A lease keeps it in the space, marked as held, so that it comes back if the holder never finishes. There are two ways to say what "finished" means.

lease: {:monitor, :ack}

Returns {:ok, tuple, handle}. The tuple is held until you call Tuplex.ack/1 with the handle, and any exit before that returns it to the space — a normal one included, because without an acknowledgement nothing says the work was done.

This is the one to reach for by default. It is the only form that suits a worker taking many tuples over its life, because the lease ends when the work does rather than when the process does.

{:ok, job, handle} = Tuplex.in({:job, :_}, lease: {:monitor, :ack})
process(job)
Tuplex.ack(handle)

lease: :monitor

Returns {:ok, tuple}, unchanged from an unleased take. The tuple is held for the lifetime of the calling process: a normal exit discards it, any other exit returns it.

That makes it right for callers whose lifetime is the unit of work — a Task per tuple — and wrong for anything longer-lived. A worker that takes 500 tuples this way holds 500 live leases, and a crash at the 501st requeues all of them, including the 499 it had already finished. Use {:monitor, :ack} there.

Task.async(fn ->
  {:ok, job} = Tuplex.in({:job, :_}, lease: :monitor)
  process(job)
  # exiting normally here discards the tuple; crashing returns it to the space
end)

Both forms

:shutdown and {:shutdown, _} requeue along with every other abnormal reason. A supervisor stopping a worker part-way through is orderly, but the work still did not happen.

A requeued tuple goes to the back of the queue, not back to its original position. That is deliberate: at the front, a tuple that crashes whoever takes it would be handed straight back to the next taker in a tight loop. At the back it starves instead, which is visible rather than fatal.

Leased tuples are invisible to rd/2, rdp/1, rd_all/1 and other in/2 callers while they are held.

The name

in is a binary operator in Elixir, so this is defined as def unquote(:in). That is invisible at the call site, but it does mean import Tuplex fails: Kernel.in/2 is a macro auto-imported everywhere, and two in/2s cannot both be in scope. Call it qualified, or use take/2. See the module docs.

Examples

Task.async(fn -> Tuplex.in({:job, :_}) end)
Tuplex.out({:job, 1})
#=> the blocked task returns {:ok, {:job, 1}}

Tuplex.in({:nothing, :_}, timeout: 10)
#=> {:error, :timeout}

@spec inp(
  Tuplex.Template.template(),
  keyword()
) ::
  {:ok, Tuplex.Template.t()}
  | {:ok, Tuplex.Template.t(), lease_handle()}
  | :empty

Removes and returns the oldest tuple matching template, without blocking.

Returns {:ok, tuple}, or :empty if nothing matches right now.

:empty is deliberately not {:error, :empty}. An empty space is an ordinary state for a tuple space to be in — unlike the timeout that the blocking in/2 can fail with, which really is exceptional.

This is the exact read: it is serialised through the tag's shard, so a tuple it returns was in the space and is now removed from it. No other consumer can also have it — within this node. That exactness is shard-local: it rests on one GenServer per tag serialising its own destructive reads, and it is exactly the guarantee that a distributed version of this library could not offer unchanged. v0.1 is single-node, so the guarantee holds; do not build on it surviving a future that spans nodes.

Tuples currently held under a lease are invisible here, as they are to every reader.

Examples

Tuplex.out({:job, 1})
Tuplex.inp({:job, :_})
#=> {:ok, {:job, 1}}
Tuplex.inp({:job, :_})
#=> :empty

Options

• :lease — as for in/2.

@spec out(Tuplex.Template.t()) :: :ok

Writes tuple into the space.

The tuple's tag selects its shard, which is started on demand. Raises ArgumentError if the tuple has no concrete atom tag.

Identical tuples do not collapse: writing {:token} three times means three tuples, and three inp/2 calls to drain them.

Examples

Tuplex.out({:job, 1, "payload"})
#=> :ok

@spec rd(
  Tuplex.Template.template(),
  keyword()
) :: {:ok, Tuplex.Template.t()} | {:error, :timeout}

Blocks until a tuple matching template is available, then returns it without removing it.

Returns {:ok, tuple}, or {:error, :timeout} if the wait runs out. Takes the same :timeout option as in/2, where 0 behaves like rdp/1.

Every waiting rd/2 whose template matches an arriving tuple is woken, not just one — a non-destructive read has no reason to be exclusive. That holds even when a blocked in/2 consumes the same tuple in the same instant: the reader is still served.

Leased tuples are invisible

A tuple held under a lease is not visible to readers, so a blocked rd/2 can sit waiting while a tuple matching its template is sitting in the space, held by someone else. That is deliberate — surfacing an in-flight claim would make rd results depend on consumer timing — but it makes the two endings of a lease observably different:

• the holder fails, the tuple is requeued, and the requeue wakes the waiting rd/2 like any other arrival;
• the holder finishes, the tuple is consumed, and the waiting rd/2 never sees it at all.

So a reader blocked on a leased tuple is woken by failure and not by success.

Examples

Task.async(fn -> Tuplex.rd({:config, :_}) end)
Tuplex.out({:config, %{mode: :fast}})
#=> the blocked task returns {:ok, {:config, %{mode: :fast}}}, tuple still in place

@spec rd_all(Tuplex.Template.template()) :: [Tuplex.Template.t()]

Returns every tuple matching template, oldest first, leaving them all in place.

Runs in the calling process, like rdp/1, and carries the same caveat: the list is a snapshot, and any of it may be taken by somebody else before you act on it. Identical tuples appear once each. Tuples currently held under a lease are invisible here; see rd/2 for what that means for a reader waiting on one.

Examples

Tuplex.out({:job, 1})
Tuplex.out({:job, 2})
Tuplex.rd_all({:job, :_})
#=> [{:job, 1}, {:job, 2}]

@spec rdp(Tuplex.Template.template()) :: {:ok, Tuplex.Template.t()} | :empty

Returns the oldest tuple matching template without removing it, and without blocking.

Returns {:ok, tuple}, or :empty if nothing matches right now.

Unlike inp/2, this runs in the calling process — a registry lookup and one ETS select, with no shard round-trip — so reads are fully parallel and never queue behind pending writes. Tuples currently held under a lease are invisible here; see rd/2.

The trade is freshness, and it is worth stating plainly rather than papering over: what you get back is a snapshot. Another process may take the tuple you just read before you do anything with it. If you need the tuple to be yours, use inp/2.

Examples

Tuplex.out({:job, 1})
Tuplex.rdp({:job, :_})
#=> {:ok, {:job, 1}}
Tuplex.rdp({:job, :_})
#=> {:ok, {:job, 1}}

@spec tags() :: [atom()]

Returns the tags that currently have a shard, in no particular order.

A template's tag has to be a concrete atom, because the tag is what selects the shard. This is the door out of that for the cases that genuinely want the whole space — debugging, introspection, a deliberate sweep:

Enum.flat_map(Tuplex.tags(), fn tag -> Tuplex.rdp({tag, :_, :_}) end)

Written that way the fan-out is explicit and its cost is visible at the call site, which is exactly where it belongs. Note that arity is part of the match too, so a sweep still has to pick one — tags/0 widens the tag, not the shape.

It is a snapshot, and it lags in both directions

A tag appears here once a shard exists for it, which is on its first out/1 — not when you first read from it. And a tag whose shard has just died stays listed until the registry processes the exit, so it can be reported for a moment after its tuples are gone, and a shard being started concurrently may not be listed yet.

Fine for introspection and debugging, which is what it is for. Do not build control flow on it.

@spec take(
  Tuplex.Template.template(),
  keyword()
) ::
  {:ok, Tuplex.Template.t()}
  | {:ok, Tuplex.Template.t(), lease_handle()}
  | {:error, :timeout}

An alias for in/2, for callers who would rather not work around the operator name.

@spec unwatch(reference()) :: :ok

Ends the subscription ref. Returns :ok even if it has already ended.

@spec watch(
  Tuplex.Template.template(),
  keyword()
) :: {:ok, reference()} | {:error, term()}

Subscribes the calling process to tuples matching template, as they happen.

Returns {:ok, ref}. Messages arrive as:

{Tuplex.Shard, :watch, ref, event, tuple}

Watching is observational, like rd/2 and unlike in/2: a watcher never consumes, and every matching watcher hears about a tuple even when a blocked in/2 takes it in the same instant. That works because a tuple is written to the table before anything is served, so what watchers are told about genuinely existed in the space.

Unlike a blocking read, a subscription is standing: it stays until unwatch/1, until the subscriber dies, or until the node does. It survives its shard crashing.

Options

• :events — which events to receive, default [:out]. Also available are :in, for tuples being consumed, and :requeue, for leased tuples returning to the space after their holder failed. :in roughly doubles the volume on a busy tag and most watchers do not want it, which is why it is off by default.

• :max_queue — drop threshold, default 10000. See below.

Watching is lossy, on purpose

A watcher is pushed at, never asked, so nothing about how fast it can keep up reaches the shard. An unbounded send to a subscriber slower than the space is an unbounded mailbox that the shard can neither see nor recover from, and the node dies of it.

So the subscriber's queue is measured before each send, and anything past :max_queue is dropped, with a [:tuplex, :watch, :dropped] telemetry event to say so. A debug console that quietly takes the node down with it is worse than one that misses events and tells you.

If you need every tuple, you need in/2 — a consumer that applies backpressure by existing — not a watcher.

Examples

{:ok, ref} = Tuplex.watch({:job, :_})
Tuplex.out({:job, 1})
#=> receives {Tuplex.Shard, :watch, ref, :out, {:job, 1}}