services/ws/server.go

   1 package main
   2
   3 import (
   4         "net"
   5         "net/http"
   6         "sync"
   7         "time"
   8
   9         "github.com/coreos/go-systemd/daemon"
  10 )
  11
  12 type server struct {
  13         httpServer  *http.Server
  14         listener    net.Listener
  15         wsConfig    *wsConfig
  16         eventSource *pgEventSource
  17         setupOnce   sync.Once
  18 }
  19
  20 func (srv *server) Close() {
  21         srv.WaitReady()
  22         srv.eventSource.Close()
  23         srv.listener.Close()
  24 }
  25
  26 func (srv *server) WaitReady() {
  27         srv.setupOnce.Do(srv.setup)
  28         srv.eventSource.WaitReady()
  29 }
  30
  31 func (srv *server) Run() error {
  32         srv.setupOnce.Do(srv.setup)
  33         return srv.httpServer.Serve(srv.listener)
  34 }
  35
  36 func (srv *server) setup() {
  37         log := logger(nil)
  38
  39         ln, err := net.Listen("tcp", srv.wsConfig.Listen)
  40         if err != nil {
  41                 log.WithField("Listen", srv.wsConfig.Listen).Fatal(err)
  42         }
  43         log.WithField("Listen", ln.Addr().String()).Info("listening")
  44
  45         srv.listener = ln
  46         srv.eventSource = &pgEventSource{
  47                 DataSource: srv.wsConfig.Postgres.ConnectionString(),
  48                 QueueSize:  srv.wsConfig.ServerEventQueue,
  49         }
  50         srv.httpServer = &http.Server{
  51                 Addr:           srv.wsConfig.Listen,
  52                 ReadTimeout:    time.Minute,
  53                 WriteTimeout:   time.Minute,
  54                 MaxHeaderBytes: 1 << 20,
  55                 Handler: &router{
  56                         Config:         srv.wsConfig,
  57                         eventSource:    srv.eventSource,
  58                         newPermChecker: func() permChecker { return newPermChecker(srv.wsConfig.Client) },
  59                 },
  60         }
  61
  62         go func() {
  63                 srv.eventSource.Run()
  64                 log.Info("event source stopped")
  65                 srv.Close()
  66         }()
  67
  68         if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
  69                 log.WithError(err).Warn("error notifying init daemon")
  70         }
  71 }