BREAKING CHANGES: Add support for transactions and support for pgmq-pg17 docker image (#67)

Author: saberder

Makefile

@@ -1,7 +1,7 @@
 .PHONY: mockgen
 mockgen:
-	mockgen --destination mocks/row.go --package=mocks --build_flags=--mod=mod github.com/jackc/pgx/v5 Row
-	mockgen --source=pgmq.go --destination mocks/pgmq.go --package=mocks
+	go run go.uber.org/mock/mockgen --destination mocks/row.go --package=mocks --build_flags=--mod=mod github.com/jackc/pgx/v5 Row
+	go run go.uber.org/mock/mockgen --source=pgmq.go --destination mocks/pgmq.go --package=mocks
 
 .PHONY: test
 test: mockgen

README.md

@@ -9,6 +9,8 @@ A Go (Golang) client for
 [Postgres Message Queue](https://github.com/tembo-io/pgmq) (PGMQ). Based loosely
 on the [Rust client](https://github.com/tembo-io/pgmq/tree/main/pgmq-rs).
 
+`pgmq-go` works with [pgx](https://github.com/jackc/pgx). The second argument of most functions only needs to satisfy the [DB](https://pkg.go.dev/github.com/craigpastro/pgmq-go#DB) interface, which means it can take, among others, a `*pgx.Conn`, `*pgxpool.Pool`, or `pgx.Tx`.
+
 ## Usage
 
 Start a Postgres instance with the PGMQ extension installed:
@@ -32,33 +34,53 @@ import (
 func main() {
     ctx := context.Background()
 
-    q, err := pgmq.New(ctx, "postgres://postgres:password@localhost:5432/postgres")
+    pool, err := pgmq.NewPgxPool(ctx, "postgres://postgres:password@localhost:5432/postgres")
+    if err != nil {
+        panic(err)
+    }
+
+    err = pgmq.CreatePGMQExtension(ctx, pool)
+    if err != nil {
+        panic(err)
+    }
+
+    err = pgmq.CreateQueue(ctx, pool, "my_queue")
     if err != nil {
         panic(err)
     }
 
-    err = q.CreateQueue(ctx, "my_queue")
+    // We can perform various queue operations using a transaction.
+    tx, err := pool.Begin(ctx)
     if err != nil {
         panic(err)
     }
 
-    id, err := q.Send(ctx, "my_queue", json.RawMessage(`{"foo": "bar"}`))
+    id, err := pgmq.Send(ctx, tx, "my_queue", json.RawMessage(`{"foo": "bar"}`))
     if err != nil {
         panic(err)
     }
 
-    msg, err := q.Read(ctx, "my_queue", 30)
+    msg, err := pgmq.Read(ctx, tx, "my_queue", 30)
     if err != nil {
         panic(err)
     }
 
     // Archive the message by moving it to the "pgmq.a_<queue_name>" table.
     // Alternatively, you can `Delete` the message, or read and delete in one
     // call by using `Pop`.
-    _, err = q.Archive(ctx, "my_queue", id)
+    _, err = pgmq.Archive(ctx, tx, "my_queue", id)
     if err != nil {
         panic(err)
     }
+
+    // Commit the transaction.
+    err = tx.Commit(ctx)
+    if err != nil {
+        panic(err)
+    }
+
+    // Close the connection pool.
+    pool.Close()
 }
 ```
 

go.mod

@@ -63,11 +63,13 @@ require (
 	go.opentelemetry.io/otel/sdk v1.21.0 // indirect
 	go.opentelemetry.io/otel/trace v1.31.0 // indirect
 	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/mod v0.18.0 // indirect
 	golang.org/x/net v0.26.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
+	golang.org/x/tools v0.22.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect

go.sum

@@ -143,6 +143,8 @@ golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.18.0 h1:5+9lSbEzPSdWkH32vYPBwEpX8KwDbM52Ud9xBUvNlb0=
+golang.org/x/mod v0.18.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -176,6 +178,8 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.22.0 h1:gqSGLZqv+AI9lIQzniJ0nZDRG5GBPsSi+DRNHWNz6yA=
+golang.org/x/tools v0.22.0/go.mod h1:aCwcsjqvq7Yqt6TNyX7QMU2enbQ/Gt0bo6krSeEri+c=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=