Support master standby connections

.github/workflows/test.yml

@@ -7,8 +7,8 @@ on:
     branches: [ main ]
 
 jobs:
-  test:
-    name: Test
+  postgresql:
+    name: PostgreSQL
     strategy:
       matrix:
         postgres: [9, 10, 11, 12, 13]
@@ -42,3 +42,20 @@ jobs:
       - name: Codecov
         uses: codecov/codecov-action@v1
         if: matrix.postgres == 13
+
+  primaryreplica:
+    name: Primary Replica
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set up Go 1.x
+        uses: actions/setup-go@v2
+      - name: Check out code into the Go module directory
+        uses: actions/checkout@v2
+      - name: Start posgresql containers
+        run: docker-compose -f "docker-compose.yml" up -d --build
+      - name: Test
+        env:
+          TEST_PRIMARY_REPLICA: 'true'
+        run: |
+          sleep 15
+          go test -race -tags=primaryreplica ./...

README.md

@@ -35,6 +35,34 @@ func main() {
 }
 ```
 
+## Example Replication (Master/Standby)
+
+```go
+package main
+
+import (
+	"context"
+
+	"github.com/go-rel/primaryreplica"
+	_ "github.com/lib/pq"
+	"github.com/go-rel/postgres"
+	"github.com/go-rel/rel"
+)
+
+func main() {
+	// open postgres connections.
+	adapter := primaryreplica.New(
+		postgres.MustOpen("postgres://postgres@master/rel_test?sslmode=disable"),
+		postgres.MustOpen("postgres://postgres@standby/rel_test?sslmode=disable"),
+	)
+	defer adapter.Close()
+
+	// initialize REL's repo.
+	repo := rel.New(adapter)
+	repo.Ping(context.TODO())
+}
+```
+
 ## Supported Driver
 
 - github.com/lib/pq

docker-compose.yml

@@ -0,0 +1,29 @@
+version: '2'
+
+services:
+  postgresql-master:
+    image: docker.io/bitnami/postgresql:13
+    ports:
+      - '25432:5432'
+    environment:
+      - POSTGRESQL_REPLICATION_MODE=master
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_USERNAME=rel
+      - POSTGRESQL_PASSWORD=rel
+      - POSTGRESQL_DATABASE=rel_test
+      - ALLOW_EMPTY_PASSWORD=yes
+  postgresql-slave:
+    image: docker.io/bitnami/postgresql:13
+    ports:
+      - '25433:5432'
+    depends_on:
+      - postgresql-master
+    environment:
+      - POSTGRESQL_REPLICATION_MODE=slave
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_MASTER_HOST=postgresql-master
+      - POSTGRESQL_PASSWORD=rel
+      - POSTGRESQL_MASTER_PORT_NUMBER=5432
+      - ALLOW_EMPTY_PASSWORD=yes

go.mod

@@ -3,7 +3,8 @@ module github.com/go-rel/postgres
 go 1.17
 
 require (
-	github.com/go-rel/rel v0.27.0
+	github.com/go-rel/primaryreplica v0.1.0
+	github.com/go-rel/rel v0.28.0
 	github.com/go-rel/sql v0.5.0
 	github.com/lib/pq v1.10.3
 	github.com/stretchr/testify v1.7.0

go.sum

@@ -4,8 +4,11 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
-github.com/go-rel/rel v0.27.0 h1:Z9NOScAMwu54sxXmL/gET5+qKkisTS95MIk6uH/+r9Y=
+github.com/go-rel/primaryreplica v0.1.0 h1:ARt8QHa55Tmz7hJYtGunz4B3THQiFDOJb0yZBgnrzx4=
+github.com/go-rel/primaryreplica v0.1.0/go.mod h1:j/b9RmL4gdnuQMh4UPBeE1h6RpQCtiUIC9QHIvgjIs8=
 github.com/go-rel/rel v0.27.0/go.mod h1:zaIYPmM3AfJrh0xBmm7KoVKRgTNvr0cgZfcJ88gVA2U=
+github.com/go-rel/rel v0.28.0 h1:gRcQjNbwuFL35RxeHFMKSy3a/xV+WbtkA8wP24dWPEA=
+github.com/go-rel/rel v0.28.0/go.mod h1:zaIYPmM3AfJrh0xBmm7KoVKRgTNvr0cgZfcJ88gVA2U=
 github.com/go-rel/sql v0.5.0 h1:+TVS9JvEl06Q8rswwuWlY6VZ+gwSBX9um+vKuZ9gsyY=
 github.com/go-rel/sql v0.5.0/go.mod h1:2YwenlIaHpTqdD/KPVYG7Y5Ub1+sn1winlC8TrighRU=
 github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=

postgres.go

@@ -64,6 +64,13 @@ func Open(dsn string) (rel.Adapter, error) {
 	return New(database), err
 }
 
+// MustOpen postgres connection using dsn.
+func MustOpen(dsn string) rel.Adapter {
+	var database, err = db.Open("postgres", dsn)
+	check(err)
+	return New(database)
+}
+
 // Insert inserts a record to database and returns its id.
 func (p Postgres) Insert(ctx context.Context, query rel.Query, primaryField string, mutates map[string]rel.Mutate) (interface{}, error) {
 	var (
@@ -176,3 +183,9 @@ func columnMapper(column *rel.Column) (string, int, int) {
 
 	return typ, m, n
 }
+
+func check(err error) {
+	if err != nil {
+		panic(err)
+	}
+}

postgres_test.go

@@ -2,10 +2,12 @@ package postgres
 
 import (
 	"context"
+	"errors"
 	"os"
 	"testing"
 	"time"
 
+	"github.com/go-rel/primaryreplica"
 	"github.com/go-rel/rel"
 	"github.com/go-rel/rel/adapter/specs"
 	_ "github.com/lib/pq"
@@ -24,16 +26,10 @@ func dsn() string {
 		return os.Getenv("POSTGRESQL_DATABASE") + "?sslmode=disable&timezone=Asia/Jakarta"
 	}
 
-	return "postgres://rel@localhost:5432/rel_test?sslmode=disable&timezone=Asia/Jakarta"
+	return "postgres://rel:rel@localhost:25432/rel_test?sslmode=disable&timezone=Asia/Jakarta"
 }
 
-func TestAdapter_specs(t *testing.T) {
-	adapter, err := Open(dsn())
-	assert.Nil(t, err)
-	defer adapter.Close()
-
-	repo := rel.New(adapter)
-
+func AdapterSpecs(t *testing.T, repo rel.Repository) {
 	// Prepare tables
 	teardown := specs.Setup(t, repo)
 	defer teardown()
@@ -101,6 +97,36 @@ func TestAdapter_specs(t *testing.T) {
 	specs.CheckConstraintOnUpdate(t, repo)
 }
 
+func TestAdapter_specs(t *testing.T) {
+	if os.Getenv("TEST_PRIMARY_REPLICA") == "true" {
+		t.Log("Skipping single node specs")
+		return
+	}
+
+	adapter := MustOpen(dsn())
+	defer adapter.Close()
+
+	repo := rel.New(adapter)
+	AdapterSpecs(t, repo)
+}
+
+func TestAdapter_PrimaryReplica_specs(t *testing.T) {
+	if os.Getenv("TEST_PRIMARY_REPLICA") != "true" {
+		t.Log("Skipping primary replica specs")
+		return
+	}
+
+	adapter := primaryreplica.New(
+		MustOpen("postgres://rel:rel@localhost:25432/rel_test?sslmode=disable&timezone=Asia/Jakarta"),
+		MustOpen("postgres://rel:rel@localhost:25433/rel_test?sslmode=disable&timezone=Asia/Jakarta"),
+	)
+
+	defer adapter.Close()
+
+	repo := rel.New(adapter)
+	AdapterSpecs(t, repo)
+}
+
 func TestAdapter_Transaction_commitError(t *testing.T) {
 	adapter, err := Open(dsn())
 	assert.Nil(t, err)
@@ -125,3 +151,9 @@ func TestAdapter_Exec_error(t *testing.T) {
 	_, _, err = adapter.Exec(ctx, "error", nil)
 	assert.NotNil(t, err)
 }
+
+func TestCheck(t *testing.T) {
+	assert.Panics(t, func() {
+		check(errors.New("error"))
+	})
+}