workerCommands defines the "workers" command to start worker processes. The workers listen to various queues such as transaction processing, indexing, and inflight expiry.
(b *ledgerforgeInstance)
| 422 | // workerCommands defines the "workers" command to start worker processes. |
| 423 | // The workers listen to various queues such as transaction processing, indexing, and inflight expiry. |
| 424 | func workerCommands(b *ledgerforgeInstance) *cobra.Command { |
| 425 | cmd := &cobra.Command{ |
| 426 | Use: "workers", |
| 427 | Short: "start ledgerforge workers", |
| 428 | Run: func(cmd *cobra.Command, args []string) { |
| 429 | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) |
| 430 | defer stop() |
| 431 | |
| 432 | conf, err := config.Fetch() |
| 433 | if err != nil { |
| 434 | logrus.Fatal("Error fetching config:", err) |
| 435 | } |
| 436 | |
| 437 | phClient, shutdown, err := initializeTelemetryAndObservability(context.Background(), conf) |
| 438 | if err != nil { |
| 439 | logrus.Fatal(err) |
| 440 | } |
| 441 | if shutdown != nil { |
| 442 | defer func() { |
| 443 | tctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 444 | defer cancel() |
| 445 | if err := shutdown(tctx); err != nil { |
| 446 | logrus.Errorf("Error during shutdown: %v", err) |
| 447 | } |
| 448 | }() |
| 449 | } |
| 450 | if phClient != nil { |
| 451 | defer phClient.Close() |
| 452 | } |
| 453 | |
| 454 | srv, hotSrv, webhookSrv, mux, webhookMux, err := setupWorkerServers(b, conf) |
| 455 | if err != nil { |
| 456 | logrus.Fatal(err) |
| 457 | } |
| 458 | |
| 459 | monitoringSrv := startMonitoringServer(conf) |
| 460 | |
| 461 | if err := srv.Start(mux); err != nil { |
| 462 | logrus.Fatalf("could not start transaction worker server: %v", err) |
| 463 | } |
| 464 | if hotSrv != nil { |
| 465 | if err := hotSrv.Start(mux); err != nil { |
| 466 | logrus.Fatalf("could not start hot transaction worker server: %v", err) |
| 467 | } |
| 468 | } |
| 469 | if err := webhookSrv.Start(webhookMux); err != nil { |
| 470 | logrus.Fatalf("could not start webhook worker server: %v", err) |
| 471 | } |
| 472 | |
| 473 | recoveryProcessor := ledgerforge.NewQueuedTransactionRecoveryProcessor(b.ledgerforge) |
| 474 | recoveryProcessor.Start(ctx) |
| 475 | |
| 476 | logrus.Info("Workers started.") |
| 477 | |
| 478 | // Wait for SIGINT/SIGTERM. |
| 479 | <-ctx.Done() |
| 480 | |
| 481 | logrus.Errorf("Shutdown signal received. Shutting down...") |
no test coverage detected