feat: Attempt to store the offset before shutting down

VictorGaiva · Jan 14, 2024 · 1d7832d · 1d7832d
1 parent 5095ae7
commit 1d7832d
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/lib/consumer/lifecycle.ex b/lib/consumer/lifecycle.ex
@@ -60,6 +60,9 @@ defmodule RabbitMQStream.Consumer.LifeCycle do
   def terminate(_reason, %{id: nil}), do: :ok
 
   def terminate(_reason, state) do
+    # While not guaranteed, we attempt to store the offset when terminating. Useful for when performing
+    # upgrades, and in a 'single-active-consumer' scenario.
+    state.connection.store_offset(state.stream_name, state.offset_reference, state.last_offset)
     state.connection.unsubscribe(state.id)
     :ok
   end