Packages

gnat

1.10.1
1.16.0 1.15.2 1.15.1 1.15.0 1.14.1 1.14.0 1.14.0-rc1 1.14.0-rc0 1.13.1 1.13.0 1.12.1 1.12.0 1.11.1 1.11.0 1.10.2 1.10.2-rc1 1.10.2-rc0 1.10.1 1.10.0 1.9.1 1.9.0 1.8.5 1.8.4 1.8.3 1.8.2 1.8.1 1.7.1 1.7.0 1.6.0 1.5.2 1.5.1 1.5.0 1.4.0 1.3.0 1.3.0-rc0 1.2.1 1.2.0 1.1.3 1.1.2 1.1.1 1.1.0 1.0.1 1.0.0 0.7.0 0.6.1 0.6.0 0.5.5 0.5.4 0.5.3 0.5.2 0.5.1 0.5.0 0.4.2 0.4.1 0.4.0 0.3.4 0.3.3 0.3.2 0.3.1 0.3.0 0.2.0 0.1.3 0.1.2 0.1.1 0.1.0

A nats client in pure elixir. Resilience, Performance, Ease-of-Use.

HexDocs

Current section

Files

Jump to
Readme
65 Versions
6 Dependencies
7 Dependants
Files
Activity
gnat lib gnat jetstream api message.ex
Raw

lib/gnat/jetstream/api/message.ex 

defmodule Gnat.Jetstream.API.Message do

  @moduledoc false



  # Based on:

  # https://github.com/nats-io/nats.py/blob/d9f24b4beae541b7723873ba0a786ea7c0ecb3d5/nats/aio/msg.py#L182



  defmodule Metadata do

    @type t :: %__MODULE__{

            :stream_seq => integer(),

            :consumer_seq => integer(),

            :num_pending => integer(),

            :num_delivered => integer(),

            :timestamp => DateTime.t(),

            :stream => String.t(),

            :consumer => String.t(),

            :domain => String.t() | nil

          }



    defstruct [

      :stream_seq,

      :consumer_seq,

      :num_pending,

      :num_delivered,

      :timestamp,

      :stream,

      :consumer,

      :domain

    ]

  end



  @spec metadata(message :: Gnat.age()) :: {:ok, Metadata.t()} | {:error, term()}

  def metadata(%{reply_to: "$JS.ACK." <> ack_topic}),

    do: decode_reply_to(String.split(ack_topic, "."))



  def metadata(_), do: {:error, :no_jetstream_message}



  # # Subject without domain:

  # $JS.ACK.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>

  defp decode_reply_to([

         stream,

         consumer,

         num_delivered,

         stream_seq,

         consumer_seq,

         ts,

         num_pending

       ]) do

    with {ts, ""} <- Integer.parse(ts),

         {:ok, ts} <- DateTime.from_unix(ts, :nanosecond),

         {stream_seq, ""} <- Integer.parse(stream_seq),

         {consumer_seq, ""} <- Integer.parse(consumer_seq),

         {num_delivered, ""} <- Integer.parse(num_delivered),

         {num_pending, ""} <- Integer.parse(num_pending) do

      {:ok,

       %Metadata{

         stream_seq: stream_seq,

         consumer_seq: consumer_seq,

         num_delivered: num_delivered,

         num_pending: num_pending,

         timestamp: ts,

         stream: stream,

         consumer: consumer

       }}

    else

      _ ->

        {:error, :invalid_ack_reply_to}

    end

  end



  # Subject with domain:

  # $JS.ACK.<domain>.<account hash>.<stream>.<consumer>.<delivered>.<sseq>.

  #   <cseq>.<tm>.<pending>.<a token with a random value>

  defp decode_reply_to([

         domain,

         _account_hash,

         stream,

         consumer,

         num_delivered,

         stream_seq,

         consumer_seq,

         ts,

         num_pending,

         _random_value

       ]) do

    case decode_reply_to([

           stream,

           consumer,

           num_delivered,

           stream_seq,

           consumer_seq,

           ts,

           num_pending

         ]) do

      {:ok, metadata} ->

        domain = if domain == "_", do: nil, else: domain

        {:ok, %{metadata | domain: domain}}



      err = {:error, _} ->

        err

    end

  end



  defp decode_reply_to(_), do: {:error, :invalid_ack_reply_to}

end