Updated

Author: djthorpe

Diff for: pkg/pgqueue/opt.go

@@ -12,6 +12,7 @@ import (
 
 type opt struct {
 	worker string
+	schema string
 }
 
 // Opt represents a function that modifies the options
@@ -39,10 +40,20 @@ func applyOpts(opts ...Opt) (*opt, error) {
 
 func OptWorker(worker string) Opt {
 	return func(o *opt) error {
-		if worker = strings.TrimSpace(worker); worker == "" {
+		if v = strings.TrimSpace(v); v == "" {
 			return httpresponse.ErrBadRequest.With("empty worker name")
 		}
-		o.worker = worker
+		o.worker = v
+		return nil
+	}
+}
+
+func OptSchema(v string) Opt {
+	return func(o *opt) error {
+		if v = strings.TrimSpace(v); v == "" {
+			return httpresponse.ErrBadRequest.With("empty schema name")
+		}
+		o.schema = v
 		return nil
 	}
 }

Diff for: pkg/pgqueue/pgqueue.go

@@ -37,6 +37,7 @@ func New(ctx context.Context, conn pg.PoolConn, opt ...Opt) (*Client, error) {
 		return nil, httpresponse.ErrInternalError.Withf("Cannot create listener")
 	} else {
 		self.listener = listener
+		self.conn = conn.With("schema", schema.SchemaName).(pg.PoolConn)
 		self.topics = []string{schema.TopicQueueInsert}
 	}
 

Diff for: plugin.go

@@ -82,5 +82,5 @@ type Logger interface {
 
 type PG interface {
 	Task
-	pg.PoolConn
+	Conn() pg.PoolConn
 }

Diff for: plugin/pg/config.go

@@ -47,7 +47,7 @@ func (c Config) New(ctx context.Context) (server.Task, error) {
 	} else if err := pool.Ping(ctx); err != nil {
 		return nil, err
 	} else {
-		return taskWithConn(pool), nil
+		return taskWith(pool), nil
 	}
 }
 

Diff for: plugin/pg/task.go

@@ -12,21 +12,25 @@ import (
 // TYPES
 
 type pgpool struct {
-	pg.PoolConn
+	conn pg.PoolConn
 }
 
 var _ server.Task = (*pgpool)(nil)
 
 ////////////////////////////////////////////////////////////////////////////////
 // LIFECYCLE
 
-func taskWithConn(conn pg.PoolConn) *pgpool {
+func taskWith(conn pg.PoolConn) *pgpool {
 	return &pgpool{conn}
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 // PUBLIC METHODS
 
+func (pg *pgpool) Conn() pg.PoolConn {
+	return pg.conn
+}
+
 func (*pgpool) Run(ctx context.Context) error {
 	<-ctx.Done()
 	return nil

Diff for: plugin/pgqueue/config.go

@@ -34,7 +34,7 @@ func (c Config) New(ctx context.Context) (server.Task, error) {
 	}
 
 	// Create a new client
-	client, err := pgqueue.New(ctx, c.Pool, opts...)
+	client, err := pgqueue.New(ctx, c.Pool.Conn(), opts...)
 	if err != nil {
 		return nil, err
 	}