handle plugins without opts

Author: filipecabaco

example/inspect_calls/lib/handler.ex

@@ -8,7 +8,6 @@ defmodule Handler do
     message
     |> PostgresReplication.Protocol.parse()
     |> PostgresReplication.Decoder.decode_message()
-    |> IO.inspect()
 
     :noreply
   end

example/inspect_calls/lib/inspect_calls.ex

@@ -34,8 +34,7 @@ defmodule InspectCalls do
     query = "INSERT INTO random_values (value) VALUES ('Random Text 1') RETURNING id"
 
     %{rows: [[id]]} =
-      Postgrex.query!(Process.whereis(WalReplication.DB1), query, []) |> IO.inspect()
-
+      Postgrex.query!(Process.whereis(WalReplication.DB1), query, [])
     id
   end
 

example/wal_replication/lib/wal_replication/handler.ex

@@ -185,7 +185,7 @@ defmodule WalReplication.Handler do
         query =
           "DELETE FROM #{schema}.#{table} WHERE #{conditions}"
 
-        Postgrex.query!(pid, query, values) |> IO.inspect()
+        Postgrex.query!(pid, query, values)
 
       :update ->
         column_names = Enum.map(state.values, &elem(&1, 0))

lib/postgres_replication.ex

@@ -177,11 +177,12 @@ defmodule PostgresReplication do
     %__MODULE__{
       replication_slot_name: replication_slot_name,
       publication_name: publication_name,
+      output_plugin: output_plugin,
       output_plugin_options: output_plugin_options
     } = state
 
     Logger.info(
-      "Starting stream replication for slot #{replication_slot_name} using plugins options: #{inspect(output_plugin_options)}"
+      "Starting stream replication for slot #{replication_slot_name} using the #{output_plugin} plugin with the options: #{inspect(output_plugin_options)}"
     )
 
     output_plugin_options =
@@ -191,9 +192,14 @@ defmodule PostgresReplication do
         {k, v} -> "#{k} '#{v}'"
       end)
       |> String.trim()
+      |> then(fn
+        "" -> ""
+        config -> " (#{config})"
+      end)
 
     query =
-      "START_REPLICATION SLOT #{replication_slot_name} LOGICAL 0/0 (#{output_plugin_options})"
+      "START_REPLICATION SLOT #{replication_slot_name} LOGICAL 0/0#{output_plugin_options}"
+      |> IO.inspect()
 
     {:stream, query, [], %{state | step: :streaming}}
   end

lib/postgres_replication/decoder.ex renamed to lib/postgres_replication/plugin/pgoutput/decoder.ex

@@ -1,4 +1,4 @@
-defmodule PostgresReplication.Decoder do
+defmodule PostgresReplication.Plugin.Pgoutput.Decoder do
   @moduledoc """
   Functions for decoding different types of logical replication messages.
   """
@@ -149,7 +149,7 @@ defmodule PostgresReplication.Decoder do
 
   alias PostgresReplication.Protocol.Write
 
-  alias PostgresReplication.OidDatabase
+  alias PostgresReplication.Plugin.Pgoutput.OidDatabase
 
   @doc """
   Parses logical replication messages from Postgres

lib/postgres_replication/oid_database.ex renamed to lib/postgres_replication/plugin/pgoutput/oid_database.ex

@@ -15,7 +15,7 @@
 #  AND typtype = 'b' -- Only basic types
 #  AND typisdefined -- Ignore undefined types
 # credo:disable-for-this-file
-defmodule PostgresReplication.OidDatabase do
+defmodule PostgresReplication.Plugin.Pgoutput.OidDatabase do
   @moduledoc "This module maps a numeric PostgreSQL type ID to a descriptive string."
 
   @doc """

test/postgres_replication_test.exs

@@ -1,5 +1,6 @@
 defmodule PostgresReplicationTest do
   use ExUnit.Case
+  alias PostgresReplication.Plugin.Pgoutput.Decoder
 
   test "handles database connection and receives WAL changes" do
     opts = %PostgresReplication{
@@ -27,14 +28,14 @@ defmodule PostgresReplicationTest do
       []
     )
 
-    assert_receive %PostgresReplication.Decoder.Messages.Begin{}
-    assert_receive %PostgresReplication.Decoder.Messages.Relation{}
+    assert_receive %Decoder.Messages.Begin{}
+    assert_receive %Decoder.Messages.Relation{}
 
-    assert_receive %PostgresReplication.Decoder.Messages.Insert{
+    assert_receive %Decoder.Messages.Insert{
       tuple_data: {_, "Random Text 1"}
     }
 
-    assert_receive %PostgresReplication.Decoder.Messages.Commit{}
+    assert_receive %Decoder.Messages.Commit{}
   end
 
   test "handles database connection, receives WAL changes and you can set plugin options" do
@@ -69,22 +70,80 @@ defmodule PostgresReplicationTest do
       []
     )
 
-    assert_receive %PostgresReplication.Decoder.Messages.Begin{}
-    assert_receive %PostgresReplication.Decoder.Messages.Relation{}
+    assert_receive %Decoder.Messages.Begin{}
+    assert_receive %Decoder.Messages.Relation{}
 
-    assert_receive %PostgresReplication.Decoder.Messages.Insert{
+    assert_receive %Decoder.Messages.Insert{
       tuple_data: {_, "Random Text 1"}
     }
 
-    assert_receive %PostgresReplication.Decoder.Messages.Commit{}
+    assert_receive %Decoder.Messages.Commit{}
+  end
+
+  test "handles database connection, receives WAL changes and can use plugins without options" do
+    opts = %PostgresReplication{
+      connection_opts: [
+        hostname: "localhost",
+        username: "postgres",
+        password: "postgres",
+        database: "postgres",
+        parameters: [
+          application_name: "PostgresReplication"
+        ]
+      ],
+      table: :all,
+      opts: [name: __MODULE__],
+      publication_name: "test_publication",
+      output_plugin: "test_decoding",
+      output_plugin_options: [],
+      handler_module: PostgresReplicationTest.TestDecodingHandler,
+      metadata: %{pid: self()}
+    }
+
+    {:ok, conn} = Postgrex.start_link(opts.connection_opts)
+    PostgresReplication.start_link(opts)
+
+    Postgrex.query!(
+      conn,
+      "INSERT INTO random_values (value) VALUES ('Random Text 1')",
+      []
+    )
+
+    assert_receive :ok
+    assert_receive :ok
+    assert_receive :ok
+  end
+
+  defmodule TestDecodingHandler do
+    @behaviour PostgresReplication.Handler
+
+    @impl true
+    def call(<<?w, _header::192, message::binary>>, %{metadata: %{pid: pid}}) do
+      send(pid, :ok)
+      :noreply
+    end
+
+    # Handles keep alive messages
+    def call(<<?k, wal_end::64, _clock::64, reply>>, _) do
+      messages =
+        case reply do
+          1 -> [<<?r, wal_end + 1::64, wal_end + 1::64, wal_end + 1::64, current_time()::64, 0>>]
+          0 -> []
+        end
+
+      {:reply, messages}
+    end
+
+    @epoch DateTime.to_unix(~U[2000-01-01 00:00:00Z], :microsecond)
+    defp current_time, do: System.os_time(:microsecond) - @epoch
   end
 
   defmodule PgoutputHandler do
     @behaviour PostgresReplication.Handler
 
     @impl true
     def call(<<?w, _header::192, message::binary>>, %{metadata: %{pid: pid}}) do
-      message |> PostgresReplication.Decoder.decode_message() |> then(&send(pid, &1))
+      message |> Decoder.decode_message() |> then(&send(pid, &1))
       :noreply
     end