8460: Return recent events if last_log_id given.

[arvados.git] / services / ws / pg.go

diff --git a/services/ws/pg.go b/services/ws/pg.go
index f89fd075b5cee1822ae73c932d3258de0d074a73..08fbee1d584c229e3fce106947311af5f4d165c3 100644 (file)
--- a/services/ws/pg.go
+++ b/services/ws/pg.go
@@ -2,6 +2,7 @@ package main
 
 import (
        "database/sql"
+       "fmt"
        "log"
        "strconv"
        "strings"
@@ -27,42 +28,49 @@ func (c pgConfig) ConnectionString() string {
 }
 
 type pgEventSource struct {
-       PgConfig  pgConfig
-       QueueSize int
+       DataSource string
+       QueueSize  int
 
+       db         *sql.DB
        pqListener *pq.Listener
        sinks      map[*pgEventSink]bool
        setupOnce  sync.Once
        mtx        sync.Mutex
+       shutdown   chan error
 }
 
 func (ps *pgEventSource) setup() {
+       ps.shutdown = make(chan error, 1)
        ps.sinks = make(map[*pgEventSink]bool)
-       go ps.run()
-}
 
-func (ps *pgEventSource) run() {
-       db, err := sql.Open("postgres", ps.PgConfig.ConnectionString())
+       db, err := sql.Open("postgres", ps.DataSource)
        if err != nil {
-               log.Fatal(err)
+               log.Fatalf("sql.Open: %s", err)
        }
+       if err = db.Ping(); err != nil {
+               log.Fatalf("db.Ping: %s", err)
+       }
+       ps.db = db
 
-       listener := pq.NewListener(ps.PgConfig.ConnectionString(), time.Second, time.Minute, func(ev pq.ListenerEventType, err error) {
+       ps.pqListener = pq.NewListener(ps.DataSource, time.Second, time.Minute, func(ev pq.ListenerEventType, err error) {
                if err != nil {
                        // Until we have a mechanism for catching up
                        // on missed events, we cannot recover from a
                        // dropped connection without breaking our
                        // promises to clients.
-                       log.Fatalf("pgEventSource listener problem: %s", err)
+                       ps.shutdown <- fmt.Errorf("pgEventSource listener problem: %s", err)
                }
        })
-       err = listener.Listen("logs")
+       err = ps.pqListener.Listen("logs")
        if err != nil {
                log.Fatal(err)
        }
-
        debugLogf("pgEventSource listening")
 
+       go ps.run()
+}
+
+func (ps *pgEventSource) run() {
        eventQueue := make(chan *event, ps.QueueSize)
 
        go func() {
@@ -87,11 +95,18 @@ func (ps *pgEventSource) run() {
        defer ticker.Stop()
        for {
                select {
+               case err, ok := <-ps.shutdown:
+                       if ok {
+                               debugLogf("shutdown on error: %s", err)
+                       }
+                       close(eventQueue)
+                       return
+
                case <-ticker.C:
                        debugLogf("pgEventSource listener ping")
-                       listener.Ping()
+                       ps.pqListener.Ping()
 
-               case pqEvent, ok := <-listener.Notify:
+               case pqEvent, ok := <-ps.pqListener.Notify:
                        if !ok {
                                close(eventQueue)
                                return
@@ -109,7 +124,7 @@ func (ps *pgEventSource) run() {
                                LogID:    logID,
                                Received: time.Now(),
                                Serial:   serial,
-                               db:       db,
+                               db:       ps.db,
                        }
                        debugLogf("event %d %+v", e.Serial, e)
                        eventQueue <- e
@@ -137,6 +152,11 @@ func (ps *pgEventSource) NewSink() eventSink {
        return sink
 }
 
+func (ps *pgEventSource) DB() *sql.DB {
+       ps.setupOnce.Do(ps.setup)
+       return ps.db
+}
+
 type pgEventSink struct {
        channel chan *event
        source  *pgEventSource