Add examples and a new help function

Author: kyleconroy

README.md

@@ -5,52 +5,69 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"log"
+	"time"
 
-	"github.com/kyleconroy/pgoutput"
 	"github.com/jackc/pgx"
+	"github.com/kyleconroy/pgoutput"
 )
 
 func main() {
 	ctx := context.Background()
-	config := pgx.ConnConfig{Database: "db", User: "replicant"}
+	config := pgx.ConnConfig{Database: "opsdash", User: "replicant"}
 	conn, err := pgx.ReplicationConnect(config)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	err = conn.StartReplication("sub1", 0, -1, `("proto_version" '1', "publication_names" 'pub1')`)
-	if err != nil {
-		log.Fatalf("Failed to start replication: %v", err)
-	}
+  // Create a slot if it doesn't already exist
+	// if err := conn.CreateReplicationSlot("sub2", "pgoutput"); err != nil {
+	// 	log.Fatalf("Failed to create replication slot: %v", err)
+	// }
 
 	set := pgoutput.NewRelationSet()
-	for {
-		var message *pgx.ReplicationMessage
-		message, err = conn.WaitForReplicationMessage(ctx)
+
+	dump := func(relation uint32, row []pgoutput.Tuple) error {
+		values, err := set.Values(relation, row)
 		if err != nil {
-			log.Fatalf("Replication failed: %v %s", err)
-		}
-		if message.WalMessage == nil {
-			continue
+			return fmt.Errorf("error parsing values: %s", err)
 		}
-		m, err := pgoutput.Parse(message.WalMessage.WalData)
-		if err != nil {
-			log.Fatalf("error parsing waldata: %s", err)
+		for name, value := range values {
+			val := value.Get()
+			log.Printf("%s (%T): %#v", name, val, val)
 		}
+		return nil
+	}
+
+	handler := func(m pgoutput.Message) error {
 		switch v := m.(type) {
 		case pgoutput.Relation:
+			log.Printf("RELATION")
 			set.Add(v)
 		case pgoutput.Insert:
-			values, err := set.Values(v.RelationID, v.Row)
-			if err != nil {
-				log.Fatalf("error parsing values: %s", err)
-			}
-			for name, value := range values {
-				val := value.Get()
-				log.Printf("%s (%T): %#v", name, val, val)
-			}
+			log.Printf("INSERT")
+			return dump(v.RelationID, v.Row)
+		case pgoutput.Update:
+			log.Printf("UPDATE")
+			return dump(v.RelationID, v.Row)
+		case pgoutput.Delete:
+			log.Printf("DELETE")
+			return dump(v.RelationID, v.Row)
 		}
+		return nil
+	}
+
+	replication := pgoutput.LogicalReplication{
+		Subscription:  "sub2",
+		Publication:   "pub2",
+		WaitTimeout:   time.Second * 10,
+		StatusTimeout: time.Second * 10,
+		Handler:       handler,
+	}
+
+	if err := replication.Start(ctx, conn); err != nil {
+		log.Fatal(err)
 	}
 }
 ```

examples/replicate.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/jackc/pgx"
+	"github.com/kyleconroy/pgoutput"
+)
+
+func main() {
+	ctx := context.Background()
+	config := pgx.ConnConfig{Database: "opsdash", User: "replicant"}
+	conn, err := pgx.ReplicationConnect(config)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// if err := conn.CreateReplicationSlot("sub2", "pgoutput"); err != nil {
+	// 	log.Fatalf("Failed to create replication slot: %v", err)
+	// }
+
+	set := pgoutput.NewRelationSet()
+
+	dump := func(relation uint32, row []pgoutput.Tuple) error {
+		values, err := set.Values(relation, row)
+		if err != nil {
+			return fmt.Errorf("error parsing values: %s", err)
+		}
+		for name, value := range values {
+			val := value.Get()
+			log.Printf("%s (%T): %#v", name, val, val)
+		}
+		return nil
+	}
+
+	handler := func(m pgoutput.Message) error {
+		switch v := m.(type) {
+		case pgoutput.Relation:
+			log.Printf("RELATION")
+			set.Add(v)
+		case pgoutput.Insert:
+			log.Printf("INSERT")
+			return dump(v.RelationID, v.Row)
+		case pgoutput.Update:
+			log.Printf("UPDATE")
+			return dump(v.RelationID, v.Row)
+		case pgoutput.Delete:
+			log.Printf("DELETE")
+			return dump(v.RelationID, v.Row)
+		}
+		return nil
+	}
+
+	replication := pgoutput.LogicalReplication{
+		Subscription:  "sub2",
+		Publication:   "pub2",
+		WaitTimeout:   time.Second * 10,
+		StatusTimeout: time.Second * 10,
+		Handler:       handler,
+	}
+
+	// Create slot
+
+	if err := replication.Start(ctx, conn); err != nil {
+		log.Fatal(err)
+	}
+}

replicate.go

@@ -0,0 +1,74 @@
+package pgoutput
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/jackc/pgx"
+)
+
+type LogicalReplication struct {
+	Subscription  string
+	Publication   string
+	WaitTimeout   time.Duration
+	StatusTimeout time.Duration
+	Handler       func(Message) error
+}
+
+func pluginArgs(version, publication string) string {
+	return fmt.Sprintf(`("proto_version" '%s', "publication_names" '%s')`, version, publication)
+}
+
+func (lr *LogicalReplication) Start(ctx context.Context, conn *pgx.ReplicationConn) error {
+	// TODO: Struct Validation here
+	err := conn.StartReplication(lr.Subscription, 0, -1, pluginArgs("1", lr.Publication))
+	if err != nil {
+		return fmt.Errorf("Failed to start replication: %v", err)
+	}
+
+	var maxWal uint64
+	tick := time.NewTicker(lr.StatusTimeout).C
+	for {
+		select {
+		case <-tick:
+			log.Println("pub status")
+			if maxWal == 0 {
+				continue
+			}
+			k, err := pgx.NewStandbyStatus(maxWal)
+			if err != nil {
+				return fmt.Errorf("Create new status failed: %s", err)
+			}
+			if err := conn.SendStandbyStatus(k); err != nil {
+				return fmt.Errorf("Sending standy status failed: %s", err)
+			}
+		default:
+			var message *pgx.ReplicationMessage
+			wctx, cancel := context.WithTimeout(ctx, lr.WaitTimeout)
+			message, err = conn.WaitForReplicationMessage(wctx)
+			cancel()
+			if err == context.DeadlineExceeded {
+				continue
+			}
+			if err != nil {
+				return fmt.Errorf("Replication failed: %s", err)
+			}
+			if message.WalMessage == nil {
+				log.Println("nil wal message")
+				continue
+			}
+			if message.WalMessage.WalStart > maxWal {
+				maxWal = message.WalMessage.WalStart
+			}
+			logmsg, err := Parse(message.WalMessage.WalData)
+			if err != nil {
+				return fmt.Errorf("invalid pgoutput message: %s", err)
+			}
+			if err := lr.Handler(logmsg); err != nil {
+				return fmt.Errorf("error handling waldata: %s", err)
+			}
+		}
+	}
+}