provide plugin options in a more flexible way

Author: filipecabaco

lib/postgres_replication.ex

@@ -11,7 +11,7 @@ defmodule PostgresReplication do
   * `publication_name` - The name of the publication to create. If not provided, it will use the schema and table name.
   * `replication_slot_name` - The name of the replication slot to create. If not provided, it will use the schema and table name.
   * `output_plugin` - The output plugin to use. Default is `pgoutput`.
-  * `proto_version` - The protocol version to use. Default is `1`.
+  * `output_plugin_options` - The options to pass to the output plugin If you use :publication_name as the value of the key, the lib will automatically set the generated or set value of `publication_name`. Default is [proto_version: "1", publication_names: :publication_name].
   * `handler_module` - The module that will handle the data received from the replication stream.
   * `target_pid` - The PID of the parent process that will receive the data.
 
@@ -40,7 +40,7 @@ defmodule PostgresReplication do
           publication_name: String.t(),
           replication_slot_name: String.t(),
           output_plugin: String.t(),
-          proto_version: integer(),
+          output_plugin_options: Keyword.t(),
           handler_module: Handler.t(),
           metadata: map()
         }
@@ -52,7 +52,7 @@ defmodule PostgresReplication do
             publication_name: nil,
             replication_slot_name: nil,
             output_plugin: "pgoutput",
-            proto_version: 1,
+            output_plugin_options: [proto_version: "1", publication_names: :publication_name],
             handler_module: nil,
             metadata: %{}
 
@@ -175,17 +175,25 @@ defmodule PostgresReplication do
         %__MODULE__{step: :start_replication_slot} = state
       ) do
     %__MODULE__{
-      proto_version: proto_version,
       replication_slot_name: replication_slot_name,
-      publication_name: publication_name
+      publication_name: publication_name,
+      output_plugin_options: output_plugin_options
     } = state
 
     Logger.info(
-      "Starting stream replication for slot #{replication_slot_name} using publication #{publication_name} and protocol version #{proto_version}"
+      "Starting stream replication for slot #{replication_slot_name} using plugins options: #{inspect(output_plugin_options)}"
     )
 
+    output_plugin_options =
+      output_plugin_options
+      |> Enum.map_join(", ", fn
+        {k, :publication_name} -> "#{k} '#{publication_name}'"
+        {k, v} -> "#{k} '#{v}'"
+      end)
+      |> String.trim()
+
     query =
-      "START_REPLICATION SLOT #{replication_slot_name} LOGICAL 0/0 (proto_version '#{proto_version}', publication_names '#{publication_name}')"
+      "START_REPLICATION SLOT #{replication_slot_name} LOGICAL 0/0 (#{output_plugin_options})"
 
     {:stream, query, [], %{state | step: :streaming}}
   end

lib/postgres_replication/decoder.ex

@@ -136,15 +136,15 @@ defmodule PostgresReplication.Decoder do
   alias Messages.{
     Begin,
     Commit,
+    Delete,
+    Insert,
     Origin,
     Relation,
     Relation.Column,
-    Insert,
-    Update,
-    Delete,
     Truncate,
     Type,
-    Unsupported
+    Unsupported,
+    Update
   }
 
   alias PostgresReplication.Protocol.Write
@@ -187,7 +187,6 @@ defmodule PostgresReplication.Decoder do
     }
   end
 
-  # TODO: Verify this is correct with real data from Postgres
   defp decode_message_impl(<<"O", lsn::binary-8, name::binary>>) do
     %Origin{
       origin_commit_lsn: decode_lsn(lsn),
@@ -201,7 +200,6 @@ defmodule PostgresReplication.Decoder do
       | [name | [<<replica_identity::binary-1, _number_of_columns::integer-16, columns::binary>>]]
     ] = String.split(rest, <<0>>, parts: 3)
 
-    # TODO: Handle case where pg_catalog is blank, we should still return the schema as pg_catalog
     friendly_replica_identity =
       case replica_identity do
         "d" -> :default

lib/postgres_replication/handler.ex

@@ -1,4 +1,7 @@
 defmodule PostgresReplication.Handler do
+  @moduledoc """
+  A behaviour module for handling logical replication messages.
+  """
   @type t :: module()
   @doc """
   The `call/2` callback is called by the `PostgresReplication` module to send messages to the parent process. It also sends back to the server connection a message in return if the user wants to.

lib/postgres_replication/oid_database.ex

@@ -14,7 +14,7 @@
 #  WHERE typnamespace = (SELECT pgn.oid FROM pg_namespace pgn WHERE nspname = 'pg_catalog') -- Take only builting Postgres types with stable OID (extension types are not guaranteed to be stable)
 #  AND typtype = 'b' -- Only basic types
 #  AND typisdefined -- Ignore undefined types
-
+# credo:disable-for-this-file
 defmodule PostgresReplication.OidDatabase do
   @moduledoc "This module maps a numeric PostgreSQL type ID to a descriptive string."
 

lib/postgres_replication/protocol.ex

@@ -1,6 +1,9 @@
 defmodule PostgresReplication.Protocol do
-  alias PostgresReplication.Protocol.Write
+  @moduledoc """
+  Functions for parsing different types of WAL messages.
+  """
   alias PostgresReplication.Protocol.KeepAlive
+  alias PostgresReplication.Protocol.Write
 
   defguard is_write(value) when binary_part(value, 0, 1) == <<?w>>
   defguard is_keep_alive(value) when binary_part(value, 0, 1) == <<?k>>
@@ -97,9 +100,9 @@ defmodule PostgresReplication.Protocol do
   @doc """
   Message to send the server to not do any operation since the server can wait
   """
-  @spec hold() :: []
-  def hold(), do: []
+  @spec hold :: []
+  def hold, do: []
 
   @epoch DateTime.to_unix(~U[2000-01-01 00:00:00Z], :microsecond)
-  def current_time(), do: System.os_time(:microsecond) - @epoch
+  def current_time, do: System.os_time(:microsecond) - @epoch
 end

mix.exs

@@ -21,7 +21,9 @@ defmodule PostgresReplication.MixProject do
   # Run "mix help deps" to learn about dependencies.
   defp deps do
     [
-      {:postgrex, "~> 0.19.0"}
+      {:postgrex, "~> 0.19.0"},
+      {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
+      {:sobelow, "~> 0.13.0", only: [:dev, :test], runtime: false}
     ]
   end
 end

mix.lock

@@ -1,6 +1,11 @@
 %{
+  "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
+  "credo": {:hex, :credo, "1.7.11", "d3e805f7ddf6c9c854fd36f089649d7cf6ba74c42bc3795d587814e3c9847102", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "56826b4306843253a66e47ae45e98e7d284ee1f95d53d1612bb483f88a8cf219"},
   "db_connection": {:hex, :db_connection, "2.7.0", "b99faa9291bb09892c7da373bb82cba59aefa9b36300f6145c5f201c7adf48ec", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "dcf08f31b2701f857dfc787fbad78223d61a32204f217f15e881dd93e4bdd3ff"},
   "decimal": {:hex, :decimal, "2.3.0", "3ad6255aa77b4a3c4f818171b12d237500e63525c2fd056699967a3e7ea20f62", [:mix], [], "hexpm", "a4d66355cb29cb47c3cf30e71329e58361cfcb37c34235ef3bf1d7bf3773aeac"},
+  "file_system": {:hex, :file_system, "1.1.0", "08d232062284546c6c34426997dd7ef6ec9f8bbd090eb91780283c9016840e8f", [:mix], [], "hexpm", "bfcf81244f416871f2a2e15c1b515287faa5db9c6bcf290222206d120b3d43f6"},
+  "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
   "postgrex": {:hex, :postgrex, "0.19.3", "a0bda6e3bc75ec07fca5b0a89bffd242ca209a4822a9533e7d3e84ee80707e19", [:mix], [{:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "d31c28053655b78f47f948c85bb1cf86a9c1f8ead346ba1aa0d0df017fa05b61"},
+  "sobelow": {:hex, :sobelow, "0.13.0", "218afe9075904793f5c64b8837cc356e493d88fddde126a463839351870b8d1e", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "cd6e9026b85fc35d7529da14f95e85a078d9dd1907a9097b3ba6ac7ebbe34a0d"},
   "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"},
 }

test/postgres_replication_test.exs

@@ -13,7 +13,49 @@ defmodule PostgresReplicationTest do
         ]
       ],
       table: :all,
-      opts: [name: __MODULE__, auto_reconnect: true],
+      opts: [name: __MODULE__],
+      handler_module: PostgresReplicationTest.PgoutputHandler,
+      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 %PostgresReplication.Decoder.Messages.Begin{}
+    assert_receive %PostgresReplication.Decoder.Messages.Relation{}
+
+    assert_receive %PostgresReplication.Decoder.Messages.Insert{
+      tuple_data: {_, "Random Text 1"}
+    }
+
+    assert_receive %PostgresReplication.Decoder.Messages.Commit{}
+  end
+
+  test "handles database connection, receives WAL changes and you can set plugin 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: "pgoutput",
+      output_plugin_options: [
+        proto_version: "2",
+        publication_names: :publication_name
+      ],
       handler_module: PostgresReplicationTest.PgoutputHandler,
       metadata: %{pid: self()}
     }
@@ -58,6 +100,6 @@ defmodule PostgresReplicationTest do
     end
 
     @epoch DateTime.to_unix(~U[2000-01-01 00:00:00Z], :microsecond)
-    defp current_time(), do: System.os_time(:microsecond) - @epoch
+    defp current_time, do: System.os_time(:microsecond) - @epoch
   end
 end