services/ws/server.go

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