A process that receives and sends PostgreSQL replication messages.

Note: this module is experimental and may be subject to changes in the future.

Logical replication

Let's see how to use this module for connecting to PostgreSQL for logical replication. First of all, you need to configure the wal level in PostgreSQL to logical. Run this inside your PostgreSQL shell/configuration:

ALTER SYSTEM SET wal_level='logical';
ALTER SYSTEM SET max_wal_senders='10';
ALTER SYSTEM SET max_replication_slots='10';

Then you must restart your server. Alternatively, you can set those values when starting "postgres". This is useful, for example, when running it from Docker:

services:
  postgres:
    image: postgres:14
    env:
      ...
    command: ["postgres", "-c", "wal_level=logical"]

For CI, GitHub Actions do not support setting command, so you can update and restart Postgres instead in a step:

- name: "Set PG settings"
  run: |
    docker exec ${{ job.services.postgres.id }} sh -c 'echo "wal_level=logical" >> /var/lib/postgresql/data/postgresql.conf'
    docker restart ${{ job.services.pg.id }}

Then you must create a publication to be replicated. This can be done in any session:

CREATE PUBLICATION postgrex_example FOR ALL TABLES;

You can also filter if you want to publish insert, update, delete or a subset of them:

# Skips updates (keeps inserts, deletes, begins, commits, etc)
create PUBLICATION postgrex_example FOR ALL TABLES WITH (publish = 'insert,delete');

# Skips inserts, updates, and deletes (keeps begins, commits, etc)
create PUBLICATION postgrex_example FOR ALL TABLES WITH (publish = '');

Now we are ready to create module that starts a replication slot and listens to our publication. Our example will use the pgoutput for logical replication and print all incoming messages to the terminal:

Mix.install([:postgrex])

defmodule Repl do
  use Postgrex.ReplicationConnection

  def start_link(opts) do
    # Automatically reconnect if we lose connection.
    extra_opts = [
      auto_reconnect: true
    ]

    Postgrex.ReplicationConnection.start_link(__MODULE__, :ok, extra_opts ++ opts)
  end

  @impl true
  def init(:ok) do
    {:ok, %{step: :disconnected}}
  end

  @impl true
  def handle_connect(state) do
    query = "CREATE_REPLICATION_SLOT postgrex TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT"
    {:query, query, %{state | step: :create_slot}}
  end

  @impl true
  def handle_result(results, %{step: :create_slot} = state) when is_list(results) do
    query = "START_REPLICATION SLOT postgrex LOGICAL 0/0 (proto_version '1', publication_names 'postgrex_example')"
    {:stream, query, [], %{state | step: :streaming}}
  end

  @impl true
  # https://www.postgresql.org/docs/14/protocol-replication.html
  def handle_data(<<?w, _wal_start::64, _wal_end::64, _clock::64, rest::binary>>, state) do
    IO.inspect(rest)
    {:noreply, state}
  end

  def handle_data(<<?k, wal_end::64, _clock::64, reply>>, state) 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

    {:noreply, messages, state}
  end

  @epoch DateTime.to_unix(~U[2000-01-01 00:00:00Z], :microsecond)
  defp current_time(), do: System.os_time(:microsecond) - @epoch
end

{:ok, pid} =
  Repl.start_link(
    host: "localhost",
    database: "demo_dev",
    username: "postgres",
  )

Process.sleep(:infinity)

use options

use Postgrex.ReplicationConnection accepts a list of options which configures the child specification and therefore how it runs under a supervisor. The generated child_spec/1 can be customized with the following options:

• :id - the child specification identifier, defaults to the current module
• :restart - when the child should be restarted, defaults to :permanent
• :shutdown - how to shut down the child, either immediately or by giving it time to shut down

For example:

use Postgrex.ReplicationConnection, restart: :transient, shutdown: 10_000

See the "Child specification" section in the Supervisor module for more detailed information. The @doc annotation immediately preceding use Postgrex.ReplicationConnection will be attached to the generated child_spec/1 function.

Name registration

A Postgrex.ReplicationConnection is bound to the same name registration rules as a GenServer. Read more about them in the GenServer docs.