GOMODULES
/
pgx
mirror of https://github.com/jackc/pgx.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Replication partially working

pull/483/head
Jack Christensen 2019-02-02 12:15:48 -06:00
parent 08c8b49fe4
commit 2e3738f0a9
2 changed files with 22 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
replication.go
View File

@ -9,6 +9,7 @@ import (
"github.com/pkg/errors"
"github.com/jackc/pgx/pgconn"
"github.com/jackc/pgx/pgio"
"github.com/jackc/pgx/pgproto3"
"github.com/jackc/pgx/pgtype"
@ -445,14 +446,25 @@ func (rc *ReplicationConn) CreateReplicationSlot(slotName, outputPlugin string)
// Create the replication slot, using the given name and output plugin, and return the consistent_point and snapshot_name values.
func (rc *ReplicationConn) CreateReplicationSlotEx(slotName, outputPlugin string) (consistentPoint string, snapshotName string, err error) {
var dummy string
var rows *Rows
rows, err = rc.sendReplicationModeQuery(fmt.Sprintf("CREATE_REPLICATION_SLOT %s LOGICAL %s", slotName, outputPlugin))
defer rows.Close()
for rows.Next() {
rows.Scan(&dummy, &consistentPoint, &snapshotName, &dummy)
var results []*pgconn.Result
results, err = rc.c.pgConn.Exec(context.TODO(), fmt.Sprintf("CREATE_REPLICATION_SLOT %s LOGICAL %s", slotName, outputPlugin)).ReadAll()
if err != nil {
return "", "", err
}
return
if len(results) != 1 {
return "", "", errors.Errorf("expected 1 result got: %d", len(results))
}
if len(results[0].Rows) != 1 {
return "", "", errors.Errorf("expected 1 row got: %d", len(results[0].Rows))
}
if len(results[0].Rows[0]) != 4 {
return "", "", errors.Errorf("expected 4 columns got: %d", len(results[0].Rows[0]))
}
return string(results[0].Rows[0][1]), string(results[0].Rows[0][2]), nil
}
// Drop the replication slot for the given name

3
replication_test.go
View File

@ -230,6 +230,7 @@ func TestReplicationConn_DropReplicationSlot(t *testing.T) {
}
func TestIdentifySystem(t *testing.T) {
t.Skipf("TODO")
connString := os.Getenv("PGX_TEST_REPLICATION_CONN_STRING")
if connString == "" {
t.Skipf("Skipping due to missing environment variable %v", "PGX_TEST_REPLICATION_CONN_STRING")
@ -284,6 +285,8 @@ func getCurrentTimeline(t *testing.T, rc *pgx.ReplicationConn) int {
}
func TestGetTimelineHistory(t *testing.T) {
t.Skipf("TODO")
connString := os.Getenv("PGX_TEST_REPLICATION_CONN_STRING")
if connString == "" {
t.Skipf("Skipping due to missing environment variable %v", "PGX_TEST_REPLICATION_CONN_STRING")