# Implementation based on gb_trees defmodule TimeQueue do @moduledoc """ This is the single module of the TimeQueue library. """ require Record alias :gb_trees, as: Tree Record.defrecordp(:trec, tref: nil, val: nil) @timespec_units [ # :millisecond, # no single millisecond :ms, :second, :seconds, :minute, :minutes, :hour, :hours, :day, :days, :week, :weeks ] @type timespec_unit :: :ms | :second | :seconds | :minute | :minutes | :hour | :hours | :day | :days | :week | :weeks @type timespec :: {pos_integer, timespec_unit} @type ttl :: timespec | integer @opaque tref :: {pos_integer, integer} @opaque entry :: record(:trec, tref: tref, val: any) # @todo add values typing @opaque t :: Tree.tree(tref, any) defguardp is_timespec(timespec) when is_integer(elem(timespec, 0)) and elem(timespec, 1) in @timespec_units @doc """ Creates an empty time queue. iex> tq = TimeQueue.new() iex> TimeQueue.peek(tq) :empty """ @spec new :: t def new, do: Tree.empty() @doc """ Returns the next event of the queue with the current system time as `now`. See `peek/2`. """ @spec peek(t) :: :empty | {:delay, non_neg_integer} | {:ok, entry} def peek(tq), do: peek(tq, system_time()) @doc """ Returns the next event of the queue according to the given current time in milliseconds. Possible return values are: - `:empty` - `{:ok, entry}` if the timestamp of the first entry is `<=` to the given current time. - `{:delay, ms}` if the timestamp of the first entry is `>` to the given current time. The remaining amount of milliseconds is returned. ### Example iex> {:ok, _tref, tq} = TimeQueue.new() |> TimeQueue.enqueue(100, :hello, _now = 0) iex> TimeQueue.peek(tq, _now = 20) {:delay, 80} iex> {:ok, _} = TimeQueue.peek(tq, _now = 100) """ @spec peek(t, now_ms :: pos_integer) :: :empty | {:delay, non_neg_integer} | {:ok, entry} def peek(tq, now) do if Tree.is_empty(tq) do :empty else case Tree.smallest(tq) do {{ts, _} = tref, val} when ts <= now -> {:ok, trec(tref: tref, val: val)} {{ts, _}, _} -> {:delay, ts - now} end end end @doc """ Extracts the next event of the queue with the current system time as `now`. See `pop/2`. """ @spec pop(t) :: :empty | {:delay, non_neg_integer} | {:ok, TimeQueue.entry()} def pop(tq), do: pop(tq, system_time()) @doc """ Extracts the next event of the queue according to the given current time in milliseconds. Possible return values are: - `:empty` - `{:ok, entry, new_queue}` if the timestamp of the first entry is `<=` to the given current time. The entry is deleted from `new_queue`. - `{:delay, ms}` if the timestamp of the first entry is `>` to the given current time. The remaining amount of milliseconds is returned. ### Example iex> {:ok, _tref, tq} = TimeQueue.new() |> TimeQueue.enqueue(100, :hello, _now = 0) iex> TimeQueue.pop(tq, _now = 20) {:delay, 80} iex> {:ok, _, _} = TimeQueue.pop(tq, _now = 100) """ @spec pop(t, now_ms :: pos_integer) :: :empty | {:delay, non_neg_integer} | {:ok, entry()} def pop(tq, now) do if Tree.is_empty(tq) do :empty else case Tree.take_smallest(tq) do {{ts, _} = tref, val, tq2} when ts <= now -> {:ok, trec(tref: tref, val: val), tq2} {{ts, _}, _, _} -> {:delay, ts - now} end end end @doc """ Deletes an entry from the queue and returns the new queue. The function does not fail if the entry was not found and simply returns the queue as-is. """ @spec delete(t, entry) :: t def delete(tq, trec(tref: tref)), do: Tree.delete(tref, tq) @doc """ Adds a new entry to the queue with a TTL and the current system time as `now`. See `enqueue/4`. """ @spec enqueue(t, ttl, any) :: {:ok, tref, t} def enqueue(tq, ttl, val), do: enqueue(tq, ttl, val, system_time()) @doc """ Adds a new entry to the queue with a TTL relative to the given timestamp in milliseconds. Returns `{:ok, tref, new_queue}` where `tref` is a timer reference (not used yed). """ @spec enqueue(t, ttl, any, now :: integer) :: {:ok, tref, t} def enqueue(tq, ttl, val, now_ms) def enqueue(tq, ttl, val, now) when is_timespec(ttl), do: enqueue_abs(tq, timespec_add(ttl, now), val) def enqueue(tq, ttl, val, now) when is_integer(ttl), do: enqueue_abs(tq, now + ttl, val) @doc """ Adds a new entry to the queue with an absolute timestamp. Returns `{:ok, tref, new_queue}` where `tref` is a timer reference (not used yed). """ @spec enqueue_abs(t, end_time :: integer, value :: any) :: {:ok, tref, t} def enqueue_abs(tq, ts, val) do tref = {ts, :erlang.unique_integer()} tq = Tree.insert(tref, val, tq) {:ok, tref, tq} end @doc """ Returns the value of an queue entry. iex> tq = TimeQueue.new() iex> {:ok, _, tq} = TimeQueue.enqueue(tq, 10, :my_value) iex> Process.sleep(10) iex> {:ok, entry} = TimeQueue.peek(tq) iex> TimeQueue.value(entry) :my_value """ @spec value(entry) :: any def value(trec(val: val)), do: val defp system_time, do: :erlang.system_time(:millisecond) defp ttl_to_milliseconds({n, :ms}) when is_integer(n) and n > 0, do: n defp ttl_to_milliseconds({_, _} = ttl) when is_timespec(ttl), do: ttl_to_seconds(ttl) * 1000 defp ttl_to_seconds({seconds, unit}) when unit in [:second, :seconds], do: seconds defp ttl_to_seconds({minutes, unit}) when unit in [:minute, :minutes], do: minutes * 60 defp ttl_to_seconds({hours, unit}) when unit in [:hour, :hours], do: hours * 60 * 60 defp ttl_to_seconds({days, unit}) when unit in [:day, :days], do: days * 24 * 60 * 60 defp ttl_to_seconds({weeks, unit}) when unit in [:week, :weeks], do: weeks * 7 * 24 * 60 * 60 defp ttl_to_seconds({_, unit}), do: raise("Unknown TTL unit: #{unit}") defp timespec_add(ttl, int), do: ttl_to_milliseconds(ttl) + int end