defmodule Redix.Stream do @moduledoc """ Documentation for Redix.Stream. """ @type redix :: pid() | atom() @type t :: String.t @doc """ Produces a new single message in a Redis stream. ## Examples iex> Redix.Stream.produce(:redix, "topic", "temperature", 55) """ @spec produce(redix, t, String.t, any()) :: {:ok, String.t} | {:error, any()} def produce(redix, stream, key, value) do case Redix.command(redix, ["XADD", stream, "*", key, value]) do {:ok, id} when is_binary(id) -> {:ok, id} {:error, error} -> {:error, error} end end @doc """ Provides a supervisable specification for a consumer which consumes from the given topic or topics. ## Examples iex> Redix.Stream.consumer(:redix, "topic", fn msg -> msg end) iex> Redix.Stream.consumer(:redix, "topic", {Module, :function, [:arg1, :arg2]}) iex> Redix.Stream.consumer(:redix, "topic", {Module, :function, [:arg1, :arg2]}, tracker: "my_stream_tracker") """ @spec consumer(redix, t, function() | mfa(), keyword()) :: Supervisor.Spec.spec def consumer(redix, stream, callback, opts \\ []) do Supervisor.Spec.worker(Redix.Stream.Consumer, [redix, stream, callback, opts]) end end