refactoring: Improves the manual start of modules by removing default implementations.

Author: VictorGaiva

lib/consumer/behaviour.ex

@@ -0,0 +1,121 @@
+defmodule RabbitMQStream.Consumer.Behaviour do
+  @doc """
+  The callback that is invoked when a chunk is received.
+
+  The server sends us messages in chunks, which of each contains many messages. But if we are
+  consuming from an offset inside of a chunk, or if we have enabled the `filter_value` parameter,
+  some of those messages might not be passed to `handle_message/1` callback.
+
+  You can use use `handle_chunk/1` to access the whole chunk for some extra logic.
+
+  Implemeting `handle_chunk/1` doesn't prevent `handle_message/1` from being called.
+
+  Optionally if you implement `handle_chunk/2`, it also passes the current
+  state of the consumer. It can be used to access the `private` field
+  passed to `start_link/1`, or the `stream_name` itself.
+
+  """
+  @callback handle_chunk(chunk :: RabbitMQStream.OsirisChunk.t()) :: term()
+  @callback handle_chunk(chunk :: RabbitMQStream.OsirisChunk.t(), state :: RabbitMQStream.Consumer.t()) :: term()
+
+  @doc """
+  Callback invoked on each message received from the stream.
+
+  This is the main way of consuming messages, and it applies any filtering and decoding necessary
+  to the message before being invoked.
+  """
+  @callback handle_message(message :: binary() | term()) :: term()
+  @callback handle_message(message :: binary() | term(), state :: RabbitMQStream.Consumer.t()) :: term()
+  @callback handle_message(
+              message :: binary() | term(),
+              chunk :: RabbitMQStream.OsirisChunk.t(),
+              state :: RabbitMQStream.Consumer.t()
+            ) ::
+              term()
+
+  @doc """
+  If the consumer has been defined with the 'single-active-consumer' parameter,
+  this callback is invoked when the consumer is being upgraded to being the
+  active one, or when downgraded to being an inactive one.
+
+  When the flag parameter is set to ':upgrade', it means that the consumer is being
+  upgraded to active and it must return the offset for where it wants to start
+  consuming from the stream.
+
+  When being downgraded, the offset returned by the callback is also sent
+  to the server but, at the moment, is not being used in any way, and is only
+  sent because the API requires. But this is actually a good moment to store
+  the offset so that it can be retrieved by the other consumer that is being
+  upgraded.
+  """
+  @callback handle_update(consumer :: RabbitMQStream.Consumer.t(), action :: :upgrade | :downgrade) ::
+              {:ok, RabbitMQStream.Connection.offset()} | {:error, any()}
+
+  @doc """
+  Callback invoked on each message inside of a chunk.
+
+  It can be used to decode the message from a binary format into a Map,
+  or to use GZIP to decompress the content.
+
+  You can also globally define a 'Serializer' module, that must implement
+  the 'decode!/1' callback, at compile-time configuration so it is added
+  to as the default callback.
+  """
+  @callback decode!(message :: String.t()) :: term()
+
+  @doc """
+  Callback invoked right before subscribing a consumer to the stream.
+  Might be usefull for setup logic, like creating a stream if it doesn't yet exists.
+  """
+  @callback before_start(RabbitMQStream.Consumer.opts(), RabbitMQStream.Consumer.t()) :: RabbitMQStream.Consumer.t()
+
+  @doc """
+  Send a command to add the provided amount of credits to the consumer.
+
+  The credits are tracked by the Server, but it is also stored internally
+  on the Consumer state, which then can be retreived by calling 'get_credits/0'.
+
+  Always returns :ok, and any errors when adding credits to a consumer are logged.
+  """
+  @callback credit(amount :: non_neg_integer()) :: :ok
+
+  @doc """
+  Returns the internally tracked amount of credits for the Consumer.
+  """
+  @callback get_credits() :: non_neg_integer()
+
+  @doc """
+  Persists the consumer's latests offset into the stream.
+
+  Be aware that it does not reset any tracking strategy.
+  """
+  @callback store_offset() :: :ok
+
+  @doc """
+  Computes the `filter_value` for a decoded messages, used for filtering incoming messages.
+
+  Must follow the same implementation you define at your `c:RabbitMQStream.Producer.filter_value/1`
+  callback.
+
+  Required when passing either `:filter` or `:match_unfiltered` properties when declaring the consumer.
+
+  Since the server sends the data in chunks, each of which is guaranted to have at least one message
+  that match the consumer filter, but it might have some that don't. We need declare this callback to
+  do additional client side filtering, so that the `handle_message/1` callback only receives those
+  messages it is really interested in.
+  """
+  @callback filter_value(term()) :: binary() | nil
+
+  @optional_callbacks handle_chunk: 1,
+                      handle_chunk: 2,
+                      handle_message: 1,
+                      handle_message: 2,
+                      handle_message: 3,
+                      decode!: 1,
+                      handle_update: 2,
+                      before_start: 2,
+                      get_credits: 0,
+                      store_offset: 0,
+                      filter_value: 1,
+                      credit: 1
+end

lib/consumer/consumer.ex

@@ -194,7 +194,7 @@ defmodule RabbitMQStream.Consumer do
       @opts unquote(opts)
       require Logger
 
-      @behaviour RabbitMQStream.Consumer
+      @behaviour RabbitMQStream.Consumer.Behaviour
 
       def start_link(opts \\ []) do
         unless !Keyword.has_key?(opts, :serializer) do
@@ -215,27 +215,22 @@ defmodule RabbitMQStream.Consumer do
         %{id: __MODULE__, start: {__MODULE__, :start_link, [opts]}}
       end
 
+      def stop() do
+        GenServer.stop(__MODULE__)
+      end
+
       def credit(amount) do
-        GenServer.cast(__MODULE__, {:credit, amount})
+        RabbitMQStream.Consumer.credit(__MODULE__, amount)
       end
 
       def get_credits() do
-        GenServer.call(__MODULE__, :get_credits)
+        RabbitMQStream.Consumer.get_credits(__MODULE__)
       end
 
       def store_offset() do
-        GenServer.call(__MODULE__, :store_offset)
+        RabbitMQStream.Consumer.store_offset(__MODULE__)
       end
 
-      def before_start(_opts, state), do: state
-
-      def handle_chunk(_chunk), do: nil
-      def handle_chunk(chunk, _state), do: handle_chunk(chunk)
-
-      def handle_message(_message), do: nil
-      def handle_message(message, _state), do: handle_message(message)
-      def handle_message(message, _chunk, state), do: handle_message(message, state)
-
       unquote(
         if serializer != nil do
           quote do
@@ -248,7 +243,7 @@ defmodule RabbitMQStream.Consumer do
         end
       )
 
-      defoverridable RabbitMQStream.Consumer
+      defoverridable RabbitMQStream.Consumer.Behaviour
     end
   end
 
@@ -266,121 +261,20 @@ defmodule RabbitMQStream.Consumer do
     %{id: __MODULE__, start: {__MODULE__, :start_link, [opts]}}
   end
 
-  @doc """
-  The callback that is invoked when a chunk is received.
-
-  The server sends us messages in chunks, which of each contains many messages. But if we are
-  consuming from an offset inside of a chunk, or if we have enabled the `filter_value` parameter,
-  some of those messages might not be passed to `handle_message/1` callback.
-
-  You can use use `handle_chunk/1` to access the whole chunk for some extra logic.
-
-  Implemeting `handle_chunk/1` doesn't prevent `handle_message/1` from being called.
-
-  Optionally if you implement `handle_chunk/2`, it also passes the current
-  state of the consumer. It can be used to access the `private` field
-  passed to `start_link/1`, or the `stream_name` itself.
-
-  """
-  @callback handle_chunk(chunk :: RabbitMQStream.OsirisChunk.t()) :: term()
-  @callback handle_chunk(chunk :: RabbitMQStream.OsirisChunk.t(), state :: t()) :: term()
-
-  @doc """
-  Callback invoked on each message received from the stream.
-
-  This is the main way of consuming messages, and it applies any filtering and decoding necessary
-  to the message before being invoked.
-  """
-  @callback handle_message(message :: binary() | term()) :: term()
-  @callback handle_message(message :: binary() | term(), state :: t()) :: term()
-  @callback handle_message(message :: binary() | term(), chunk :: RabbitMQStream.OsirisChunk.t(), state :: t()) ::
-              term()
-
-  @doc """
-  If the consumer has been defined with the 'single-active-consumer' parameter,
-  this callback is invoked when the consumer is being upgraded to being the
-  active one, or when downgraded to being an inactive one.
-
-  When the flag parameter is set to ':upgrade', it means that the consumer is being
-  upgraded to active and it must return the offset for where it wants to start
-  consuming from the stream.
-
-  When being downgraded, the offset returned by the callback is also sent
-  to the server but, at the moment, is not being used in any way, and is only
-  sent because the API requires. But this is actually a good moment to store
-  the offset so that it can be retrieved by the other consumer that is being
-  upgraded.
-  """
-  @callback handle_update(consumer :: t(), action :: :upgrade | :downgrade) ::
-              {:ok, RabbitMQStream.Connection.offset()} | {:error, any()}
-
-  @doc """
-  Callback invoked on each message inside of a chunk.
-
-  It can be used to decode the message from a binary format into a Map,
-  or to use GZIP to decompress the content.
-
-  You can also globally define a 'Serializer' module, that must implement
-  the 'decode!/1' callback, at compile-time configuration so it is added
-  to as the default callback.
-  """
-  @callback decode!(message :: String.t()) :: term()
-
-  @doc """
-  Callback invoked right before subscribing a consumer to the stream.
-  Might be usefull for setup logic, like creating a stream if it doesn't yet exists.
-  """
-  @callback before_start(opts(), t()) :: t()
-
-  @doc """
-  Send a command to add the provided amount of credits to the consumer.
-
-  The credits are tracked by the Server, but it is also stored internally
-  on the Consumer state, which then can be retreived by calling 'get_credits/0'.
-
-  Always returns :ok, and any errors when adding credits to a consumer are logged.
-  """
-  @callback credit(amount :: non_neg_integer()) :: :ok
-
-  @doc """
-  Returns the internally tracked amount of credits for the Consumer.
-  """
-  @callback get_credits() :: non_neg_integer()
-
-  @doc """
-  Persists the consumer's latests offset into the stream.
-
-  Be aware that it does not reset any tracking strategy.
-  """
-  @callback store_offset() :: :ok
-
-  @doc """
-  Computes the `filter_value` for a decoded messages, used for filtering incoming messages.
-
-  Must follow the same implementation you define at your `c:RabbitMQStream.Producer.filter_value/1`
-  callback.
+  @spec credit(GenServer.server(), amount :: non_neg_integer()) :: :ok
+  def credit(server, amount) do
+    GenServer.cast(server, {:credit, amount})
+  end
 
-  Required when passing either `:filter` or `:match_unfiltered` properties when declaring the consumer.
+  @spec get_credits(GenServer.server()) :: non_neg_integer()
+  def get_credits(server) do
+    GenServer.call(server, :get_credits)
+  end
 
-  Since the server sends the data in chunks, each of which is guaranted to have at least one message
-  that match the consumer filter, but it might have some that don't. We need declare this callback to
-  do additional client side filtering, so that the `handle_message/1` callback only receives those
-  messages it is really interested in.
-  """
-  @callback filter_value(term()) :: binary() | nil
-
-  @optional_callbacks handle_chunk: 1,
-                      handle_chunk: 2,
-                      handle_message: 1,
-                      handle_message: 2,
-                      handle_message: 3,
-                      decode!: 1,
-                      handle_update: 2,
-                      before_start: 2,
-                      get_credits: 0,
-                      store_offset: 0,
-                      filter_value: 1,
-                      credit: 1
+  @spec store_offset(GenServer.server()) :: :ok
+  def store_offset(server) do
+    GenServer.call(server, :store_offset)
+  end
 
   defstruct [
     :offset_reference,

lib/consumer/lifecycle.ex

@@ -49,7 +49,12 @@ defmodule RabbitMQStream.Consumer.LifeCycle do
 
   @impl true
   def handle_continue({:init, opts}, state) do
-    state = apply(state.consumer_module, :before_start, [opts, state])
+    state =
+      if function_exported?(state.consumer_module, :before_start, 2) do
+        apply(state.consumer_module, :before_start, [opts, state])
+      else
+        state
+      end
 
     last_offset =
       case RabbitMQStream.Connection.query_offset(state.connection, state.stream_name, state.offset_reference) do
@@ -118,13 +123,31 @@ defmodule RabbitMQStream.Consumer.LifeCycle do
 
   @impl true
   def handle_info({:deliver, %DeliverData{osiris_chunk: %RabbitMQStream.OsirisChunk{} = chunk}}, state) do
-    # We could wrap the application with a `try do` clause, but it would break the "let it crash" filosophy
-    apply(state.consumer_module, :handle_chunk, [chunk, state])
+    if function_exported?(state.consumer_module, :handle_chunk, 2) do
+      apply(state.consumer_module, :handle_chunk, [chunk, state])
+    end
+
+    for message <- Enum.slice(chunk.data_entries, (state.last_offset - chunk.chunk_id)..chunk.num_entries) do
+      message =
+        if function_exported?(state.consumer_module, :decode!, 1) do
+          apply(state.consumer_module, :decode!, [message])
+        else
+          message
+        end
+
+      if filtered?(message, state) do
+        if function_exported?(state.consumer_module, :handle_message, 3) do
+          apply(state.consumer_module, :handle_message, [message, chunk, state])
+        end
 
-    for message <- Enum.slice(chunk.data_entries, (state.last_offset - chunk.chunk_id)..chunk.num_entries),
-        decoded = apply(state.consumer_module, :decode!, [message]),
-        filtered?(decoded, state) do
-      apply(state.consumer_module, :handle_message, [decoded, chunk, state])
+        if function_exported?(state.consumer_module, :handle_message, 2) do
+          apply(state.consumer_module, :handle_message, [message, state])
+        end
+
+        if function_exported?(state.consumer_module, :handle_message, 1) do
+          apply(state.consumer_module, :handle_message, [message])
+        end
+      end
     end
 
     # Based on the [Python implementation](https://github.com/qweeze/rstream/blob/a81176a5c7cf4accaee25ca7725bd7bd94bf0ce8/rstream/consumer.py#L327),
@@ -213,7 +236,12 @@ defmodule RabbitMQStream.Consumer.LifeCycle do
     match_unfiltered = state.properties[:match_unfiltered]
 
     if filter != nil or match_unfiltered != nil do
-      filter_value = apply(state.consumer_module, :filter_value, [decoded])
+      filter_value =
+        if function_exported?(state.consumer_module, :filter_value, 1) do
+          apply(state.consumer_module, :filter_value, [decoded])
+        else
+          nil
+        end
 
       case {filter_value, filter, match_unfiltered} do
         {nil, _, true} ->