services/ws/service.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package ws
   6
   7 import (
   8         "context"
   9         "fmt"
  10         "os"
  11
  12         "git.arvados.org/arvados.git/lib/cmd"
  13         "git.arvados.org/arvados.git/lib/service"
  14         "git.arvados.org/arvados.git/sdk/go/arvados"
  15         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  16         "github.com/prometheus/client_golang/prometheus"
  17 )
  18
  19 var testMode = false
  20
  21 var Command cmd.Handler = service.Command(arvados.ServiceNameWebsocket, newHandler)
  22
  23 func newHandler(ctx context.Context, cluster *arvados.Cluster, token string, reg *prometheus.Registry) service.Handler {
  24         client, err := arvados.NewClientFromConfig(cluster)
  25         if err != nil {
  26                 return service.ErrorHandler(ctx, cluster, fmt.Errorf("error initializing client from cluster config: %s", err))
  27         }
  28         eventSource := &pgEventSource{
  29                 DataSource:   cluster.PostgreSQL.Connection.String(),
  30                 MaxOpenConns: cluster.PostgreSQL.ConnectionPool,
  31                 QueueSize:    cluster.API.WebsocketServerEventQueue,
  32                 Logger:       ctxlog.FromContext(ctx),
  33         }
  34         go func() {
  35                 eventSource.Run()
  36                 ctxlog.FromContext(ctx).Error("event source stopped")
  37                 if !testMode {
  38                         os.Exit(1)
  39                 }
  40         }()
  41         eventSource.WaitReady()
  42         if err := eventSource.DBHealth(); err != nil {
  43                 return service.ErrorHandler(ctx, cluster, err)
  44         }
  45         rtr := &router{
  46                 cluster:        cluster,
  47                 client:         client,
  48                 eventSource:    eventSource,
  49                 newPermChecker: func() permChecker { return newPermChecker(*client) },
  50         }
  51         return rtr
  52 }