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
  11         "git.arvados.org/arvados.git/lib/cmd"
  12         "git.arvados.org/arvados.git/lib/service"
  13         "git.arvados.org/arvados.git/sdk/go/arvados"
  14         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  15         "github.com/prometheus/client_golang/prometheus"
  16 )
  17
  18 var testMode = false
  19
  20 var Command cmd.Handler = service.Command(arvados.ServiceNameWebsocket, newHandler)
  21
  22 func newHandler(ctx context.Context, cluster *arvados.Cluster, token string, reg *prometheus.Registry) service.Handler {
  23         client, err := arvados.NewClientFromConfig(cluster)
  24         if err != nil {
  25                 return service.ErrorHandler(ctx, cluster, fmt.Errorf("error initializing client from cluster config: %s", err))
  26         }
  27         eventSource := &pgEventSource{
  28                 DataSource:   cluster.PostgreSQL.Connection.String(),
  29                 MaxOpenConns: cluster.PostgreSQL.ConnectionPool,
  30                 QueueSize:    cluster.API.WebsocketServerEventQueue,
  31                 Logger:       ctxlog.FromContext(ctx),
  32                 Reg:          reg,
  33         }
  34         done := make(chan struct{})
  35         go func() {
  36                 eventSource.Run()
  37                 ctxlog.FromContext(ctx).Error("event source stopped")
  38                 close(done)
  39         }()
  40         eventSource.WaitReady()
  41         if err := eventSource.DBHealth(); err != nil {
  42                 return service.ErrorHandler(ctx, cluster, err)
  43         }
  44         rtr := &router{
  45                 cluster:        cluster,
  46                 client:         client,
  47                 eventSource:    eventSource,
  48                 newPermChecker: func() permChecker { return newPermChecker(*client) },
  49                 done:           done,
  50                 reg:            reg,
  51         }
  52         return rtr
  53 }