-
Notifications
You must be signed in to change notification settings - Fork 18
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(destination): implement task handler using etcd adapter, separat…
…e workers from task coordinators
- Loading branch information
1 parent
1f47321
commit 1001d91
Showing
22 changed files
with
277 additions
and
160 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -27,6 +27,7 @@ logs/ | |
| mise.log | ||
|
|
||
| destination/logs.json | ||
| destination/worker_log.json | ||
|
|
||
| temp | ||
| .env | ||
|
|
||
109 changes: 109 additions & 0 deletions
109
cmd/destination/delivery_workers/webhook_delivery_worker.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log" | ||
| "log/slog" | ||
| "os" | ||
| "os/signal" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/ormushq/ormus/adapter/etcd" | ||
| "github.com/ormushq/ormus/adapter/redis" | ||
| "github.com/ormushq/ormus/config" | ||
| "github.com/ormushq/ormus/destination/taskdelivery" | ||
| "github.com/ormushq/ormus/destination/taskdelivery/adapters/fakedeliveryhandler" | ||
| "github.com/ormushq/ormus/destination/taskmanager/adapter/rabbitmqtaskmanager" | ||
| "github.com/ormushq/ormus/destination/taskservice" | ||
| "github.com/ormushq/ormus/destination/taskservice/adapter/idempotency/redistaskidempotency" | ||
| "github.com/ormushq/ormus/destination/taskservice/adapter/repository/inmemorytaskrepo" | ||
| "github.com/ormushq/ormus/destination/worker" | ||
| "github.com/ormushq/ormus/logger" | ||
| ) | ||
|
|
||
| const waitingAfterShutdownInSeconds = 2 | ||
|
|
||
| func main() { | ||
| done := make(chan bool) | ||
| wg := sync.WaitGroup{} | ||
|
|
||
| //----------------- Setup Logger -----------------// | ||
|
|
||
| fileMaxSizeInMB := 10 | ||
| fileMaxAgeInDays := 30 | ||
|
|
||
| cfg := logger.Config{ | ||
| FilePath: "./destination/worker_log.json", | ||
| UseLocalTime: false, | ||
| FileMaxSizeInMB: fileMaxSizeInMB, | ||
| FileMaxAgeInDays: fileMaxAgeInDays, | ||
| } | ||
|
|
||
| logLevel := slog.LevelInfo | ||
| if config.C().Destination.DebugMode { | ||
| logLevel = slog.LevelDebug | ||
| } | ||
|
|
||
| opt := slog.HandlerOptions{ | ||
| // todo should level debug be read from config? | ||
| Level: logLevel, | ||
| } | ||
| l := logger.New(cfg, &opt) | ||
| slog.SetDefault(l) | ||
|
|
||
| //----------------- Setup Task Service -----------------// | ||
|
|
||
| redisAdapter, err := redis.New(config.C().Redis) | ||
| if err != nil { | ||
| log.Panicf("error in new redis") | ||
| } | ||
| taskIdempotency := redistaskidempotency.New(redisAdapter, "tasks:", 30*24*time.Hour) | ||
|
|
||
| taskRepo := inmemorytaskrepo.New() | ||
|
|
||
| // Set up etcd as distributed locker. | ||
| distributedLocker, err := etcd.New(config.C().Etcd) | ||
| if err != nil { | ||
| log.Panicf("Error on new etcd") | ||
| } | ||
|
|
||
| taskService := taskservice.New(taskIdempotency, taskRepo, distributedLocker) | ||
|
|
||
| // Register delivery handlers | ||
| // each destination type can have specific delivery handler | ||
| fakeTaskDeliveryHandler := fakedeliveryhandler.New() | ||
| taskdelivery.Register("webhook", fakeTaskDeliveryHandler) | ||
|
|
||
| //----------------- Consume ProcessEvents -----------------// | ||
|
|
||
| taskConsumerConf := config.C().Destination.RabbitMQTaskManagerConnection | ||
| webhookTaskConsumer := rabbitmqtaskmanager.NewTaskConsumer(taskConsumerConf, "webhook_tasks_queue") | ||
|
|
||
| processedEvents, err := webhookTaskConsumer.Consume(done, &wg) | ||
| if err != nil { | ||
| log.Panicf("Error on consuming tasks.") | ||
| } | ||
|
|
||
| w1 := worker.NewWorker(processedEvents, taskService) | ||
|
|
||
| w1Err := w1.Run(done, &wg) | ||
| if w1Err != nil { | ||
| log.Panicf("%s: %s", "Error on webhook worker", err) | ||
| } | ||
|
|
||
| //----------------- Handling graceful shutdown -----------------// | ||
|
|
||
| quit := make(chan os.Signal, 1) | ||
| signal.Notify(quit, os.Interrupt) | ||
| <-quit | ||
|
|
||
| fmt.Println("Received interrupt signal, shutting down gracefully...") | ||
| done <- true | ||
|
|
||
| close(done) | ||
|
|
||
| // todo use config for waiting time after graceful shutdown | ||
| time.Sleep(waitingAfterShutdownInSeconds * time.Second) | ||
| wg.Wait() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,12 +1,11 @@ | ||
| package taskrouter | ||
| package taskcoordinator | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/ormushq/ormus/event" | ||
| ) | ||
|
|
||
| // Coordinator is responsible for setup task managers and publish coming process events using suitable task publishers. | ||
| type Coordinator interface { | ||
| Start(processedEvents <-chan *event.ProcessedEvent, done <-chan bool, wg *sync.WaitGroup) | ||
| Start(processedEvents <-chan event.ProcessedEvent, done <-chan bool, wg *sync.WaitGroup) | ||
| } |
Oops, something went wrong.