anycable / anycable-go / 4737922678

func NewRunner(c *config.Config, options []Option) (*Runner, error) {

        r := &Runner{

                options:            options,

                config:             c,

                shutdownables:      []Shutdownable{},

                websocketEndpoints: make(map[string]websocketHandler),

                errChan:            make(chan error),

        }

        err := r.checkAndSetDefaults()

        if err != nil {

        return r, nil

func (r *Runner) checkAndSetDefaults() error {

        for _, o := range r.options {

                err := o(r)

                if err != nil {

        err := utils.InitLogger(r.config.LogFormat, r.config.LogLevel)

        if err != nil {

        r.log = log.WithFields(log.Fields{"context": "main"})

        err = r.config.LoadPresets()

        if err != nil {

        server.SSL = &r.config.SSL

        server.Host = r.config.Host

        server.MaxConn = r.config.MaxConn

        if r.name == "" {

        if r.controllerFactory == nil {

        if r.brokerFactory == nil {

        if r.subscriberFactory == nil {

        if r.disconnectorFactory == nil {

                r.disconnectorFactory = r.defaultDisconnector

        }

        if r.websocketHandlerFactory == nil {

                r.websocketHandlerFactory = r.defaultWebSocketHandler

        }

        return nil

func (r *Runner) Run() error {

        numProcs := r.setMaxProcs()

        r.announceDebugMode()

        mrubySupport := r.initMRuby()

        r.log.Infof("Starting %s %s%s (pid: %d, open file limit: %s, gomaxprocs: %d)", r.name, version.Version(), mrubySupport, os.Getpid(), utils.OpenFileLimit(), numProcs)

        metrics, err := r.initMetrics(&r.config.Metrics)

        if err != nil {

        r.shutdownables = append(r.shutdownables, metrics)

        controller, err := r.newController(metrics)

        if err != nil {

        appNode := node.NewNode(controller, metrics, &r.config.App)

        subscriber, err := r.subscriberFactory(appNode, r.config)

        if err != nil {

        appBroker, err := r.brokerFactory(subscriber, r.config)

        if err != nil {

        if appBroker != nil {

                r.log.Infof(appBroker.Announce())

                appNode.SetBroker(appBroker)

        }

        disconnector, err := r.disconnectorFactory(appNode, r.config)

        if err != nil {

        go disconnector.Run() // nolint:errcheck

        appNode.SetDisconnector(disconnector)

        err = appNode.Start()

        if err != nil {

        if r.config.EmbedNats {

                service, enatsErr := r.embedNATS(&r.config.EmbeddedNats)

                if enatsErr != nil {

                desc := service.Description()

                if desc != "" {

                        desc = fmt.Sprintf(" (%s)", desc)

                }

                r.log.Infof("Embedded NATS server started: %s%s", r.config.EmbeddedNats.ServiceAddr, desc)

                r.shutdownables = append(r.shutdownables, service)

        err = subscriber.Start(r.errChan)

        if err != nil {

        r.shutdownables = append(r.shutdownables, subscriber)

        if r.broadcasters != nil {

                for _, broadcasterFactory := range r.broadcasters {

                        broadcaster, berr := broadcasterFactory(appNode, r.config)

                        if berr != nil {

                        if broadcaster.IsFanout() && subscriber.IsMultiNode() {

                        if !broadcaster.IsFanout() && !subscriber.IsMultiNode() {

                                r.log.Warnf("Using a non-distributed broadcaster without a pub/sub enabled; each broadcasted message is only processed by a single node")

                        }

                        err = broadcaster.Start(r.errChan)

                        if err != nil {

                        r.shutdownables = append(r.shutdownables, broadcaster)

        err = controller.Start()

        if err != nil {

        wsServer, err := server.ForPort(strconv.Itoa(r.config.Port))

        if err != nil {

        wsHandler, err := r.websocketHandlerFactory(appNode, r.config)

        if err != nil {

        for _, path := range r.config.Path {

                wsServer.SetupHandler(path, wsHandler)

                r.log.Infof("Handle WebSocket connections at %s%s", wsServer.Address(), path)

        }

        for path, handlerFactory := range r.websocketEndpoints {

        wsServer.SetupHandler(r.config.HealthPath, http.HandlerFunc(server.HealthHandler))

        r.log.Infof("Handle health requests at %s%s", wsServer.Address(), r.config.HealthPath)

        go r.startWSServer(wsServer)

        go r.startMetrics(metrics)

        r.shutdownables = append(r.shutdownables, appBroker, appNode)

        r.announceGoPools()

        r.setupSignalHandlers()

        // Wait for an error (or none)

        return <-r.errChan

func (r *Runner) setMaxProcs() int {

        // See https://github.com/uber-go/automaxprocs/issues/18

        nopLog := func(string, ...interface{}) {}

        maxprocs.Set(maxprocs.Logger(nopLog)) // nolint:errcheck

        return runtime.GOMAXPROCS(0)

func (r *Runner) announceDebugMode() {

        if r.config.Debug {

                r.log.Debug("🔧 🔧 🔧 Debug mode is on 🔧 🔧 🔧")

        }

func (r *Runner) initMetrics(c *metricspkg.Config) (*metricspkg.Metrics, error) {

        m, err := metricspkg.NewFromConfig(c)

        if err != nil {

        if c.Statsd.Enabled() {

        return m, nil

func (r *Runner) newController(metrics *metricspkg.Metrics) (node.Controller, error) {

        controller, err := r.controllerFactory(metrics, r.config)

        if err != nil {

        if r.config.JWT.Enabled() {

        if !r.Router().Empty() {

        return controller, nil

func (r *Runner) startWSServer(wsServer *server.HTTPServer) {

        go func() {

                err := wsServer.StartAndAnnounce("WebSocket server")

                if err != nil {

func (r *Runner) startMetrics(metrics *metricspkg.Metrics) {

        err := metrics.Run()

        if err != nil {

func (r *Runner) defaultDisconnector(n *node.Node, c *config.Config) (node.Disconnector, error) {

        if c.DisconnectorDisabled {

                return node.NewNoopDisconnector(), nil

        }

        return node.NewDisconnectQueue(n, &c.DisconnectQueue), nil

func (r *Runner) defaultWebSocketHandler(n *node.Node, c *config.Config) (http.Handler, error) {

        extractor := ws.DefaultHeadersExtractor{Headers: c.Headers, Cookies: c.Cookies}

        return ws.WebsocketHandler(common.ActionCableProtocols(), &extractor, &c.WS, func(wsc *websocket.Conn, info *ws.RequestInfo, callback func()) error {

                wrappedConn := ws.NewConnection(wsc)

                session := node.NewSession(n, wrappedConn, info.URL, info.Headers, info.UID)

                _, err := n.Authenticate(session)

                if err != nil {

                return session.Serve(callback)

func (r *Runner) initMRuby() string {

        if mrb.Supported() {

                var mrbv string

                mrbv, err := mrb.Version()

                if err != nil {

                } else {

                        return " (with " + mrbv + ")"

                }

func (r *Runner) Router() *router.RouterController {

        if r.router == nil {

                r.SetRouter(r.defaultRouter())

        }

        return r.router

func (r *Runner) SetRouter(router *router.RouterController) {

        r.router = router

}

func (r *Runner) defaultRouter() *router.RouterController {

        router := router.NewRouterController(nil)

        if r.config.Rails.TurboRailsKey != "" {

        if r.config.Rails.CableReadyKey != "" {

        return router

func (r *Runner) announceGoPools() {

        configs := make([]string, 0)

        pools := utils.AllPools()

        for _, pool := range pools {

                configs = append(configs, fmt.Sprintf("%s: %d", pool.Name(), pool.Size()))

        }

        log.WithField("context", "main").Debugf("Go pools initialized (%s)", strings.Join(configs, ", "))

func (r *Runner) setupSignalHandlers() {

        t := tebata.New(syscall.SIGINT, syscall.SIGTERM)

        t.Reserve(func() { // nolint:errcheck

                log.Infof("Shutting down... (hit Ctrl-C to stop immediately)")

                go func() {

                        termSig := make(chan os.Signal, 1)

                        signal.Notify(termSig, syscall.SIGINT, syscall.SIGTERM)

                        <-termSig

                        log.Warnf("Immediate termination requested. Stopped")

                        r.errChan <- nil

                }()

        for _, shutdownable := range r.shutdownables {

                t.Reserve(shutdownable.Shutdown) // nolint:errcheck

        }

        t.Reserve(func() { r.errChan <- nil }) // nolint:errcheck

func (r *Runner) embedNATS(c *enats.Config) (*enats.Service, error) {

        service := enats.NewService(c)

        err := service.Start()

        if err != nil {

        return service, nil