pg writer: close confirms channel (#358)

provides support for acknowledged writes

adaptor/postgres/writer.go

@@ -34,10 +34,19 @@ func (w *Writer) Write(msg message.Msg) func(client.Session) (message.Msg, error
 	return func(s client.Session) (message.Msg, error) {
 		writeFunc, ok := w.writeMap[msg.OP()]
 		if !ok {
-			log.Infof("no function registered for operation, %s\n", msg.OP())
+			log.Infof("no function registered for operation, %s", msg.OP())
+			if msg.Confirms() != nil {
+				close(msg.Confirms())
+			}
 			return msg, nil
 		}
-		return msg, writeFunc(msg, s.(*Session).pqSession)
+		if err := writeFunc(msg, s.(*Session).pqSession); err != nil {
+			return nil, err
+		}
+		if msg.Confirms() != nil {
+			close(msg.Confirms())
+		}
+		return msg, nil
 	}
 }
 

adaptor/postgres/writer_test.go

@@ -3,6 +3,7 @@ package postgres
 import (
 	"fmt"
 	"math/rand"
+	"sync"
 	"testing"
 	"time"
 
@@ -47,15 +48,55 @@ func TestInsert(t *testing.T) {
 		t.Fatalf("unable to obtain session to postgres, %s", err)
 	}
 	for i := 0; i < 10; i++ {
-		msg := message.From(
-			ops.Insert,
-			fmt.Sprintf("public.%s", writerTestData.Table),
-			data.Data{"id": i, "colvar": "hello world", "coltimestamp": time.Now().UTC()})
-		if _, err := w.Write(msg)(s); err != nil {
+		if _, err := w.Write(
+			message.WithConfirms(
+				make(chan struct{}),
+				message.From(
+					ops.Insert,
+					fmt.Sprintf("public.%s", writerTestData.Table),
+					data.Data{"id": i, "colvar": "hello world", "coltimestamp": time.Now().UTC()}),
+			),
+		)(s); err != nil {
 			t.Errorf("unexpected Insert error, %s\n", err)
 		}
 	}
 
+	unusedConfirms := make(chan struct{})
+	var wg sync.WaitGroup
+	var confirmClosed bool
+	wg.Add(1)
+	go func(wg *sync.WaitGroup) {
+		select {
+		case <-time.After(10 * time.Second):
+		case <-unusedConfirms:
+			confirmClosed = true
+		}
+		wg.Done()
+	}(&wg)
+	if _, err := w.Write(message.WithConfirms(
+		unusedConfirms,
+		message.From(
+			ops.Command,
+			fmt.Sprintf("public.%s", writerTestData.Table),
+			map[string]interface{}{},
+		)),
+	)(s); err != nil {
+		t.Errorf("unexpected Command error, %s", err)
+	}
+
+	wg.Wait()
+	if !confirmClosed {
+		t.Errorf("confirms chan should have been closed but isn't")
+	}
+
+	if _, err := w.Write(message.From(
+		ops.Command,
+		fmt.Sprintf("public.%s", writerTestData.Table),
+		map[string]interface{}{},
+	))(s); err != nil {
+		t.Errorf("unexpected Command error, %s", err)
+	}
+
 	var (
 		id          int
 		stringValue string