-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
6d970b8
commit 5e70e29
Showing
18 changed files
with
583 additions
and
0 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
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 |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| package pipeline | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/google/uuid" | ||
| ) | ||
|
|
||
| type ConcurrentJob struct { | ||
| id string | ||
| name string | ||
| defaultValue interface{} | ||
| jobs []Job | ||
| config JobConfig | ||
| aggregator Aggregator | ||
| errorHandler ErrorHandler | ||
| summary Summary | ||
| } | ||
|
|
||
| func (concurrent ConcurrentJob) Do(ctx context.Context) JobResult { | ||
| start := time.Now() | ||
| r := concurrent.do(ctx) | ||
| elapsed := time.Since(start).Milliseconds() | ||
| concurrent.summary.summary(concurrent.id, concurrent.name, len(concurrent.jobs), concurrent.config, elapsed) | ||
| return r | ||
| } | ||
|
|
||
| func (concurrent ConcurrentJob) do(ctx context.Context) JobResult { | ||
| data := concurrent.defaultValue | ||
| ch := make(chan JobResult, concurrent.config.maxConcurrency) | ||
| length := len(concurrent.jobs) | ||
| for index, job := range concurrent.jobs { | ||
| go func(c context.Context, j Job) { | ||
| defer func() { | ||
| if re := recover(); re != nil { | ||
| ch <- FailureResult(fmt.Errorf("concurrent job %v unexpected error: %v", concurrent.id, re), "Unexpected failure") | ||
| } | ||
| }() | ||
| ch <- j.Do(c) | ||
| }(ctx, job) | ||
| if (index+1)%concurrent.config.maxConcurrency == 0 || index+1 == length { | ||
| for i := 0; i <= index%concurrent.config.maxConcurrency; i++ { | ||
| r := <-ch | ||
| if r.Success { | ||
| d, e := concurrent.aggregator(ctx, data, r.Data) | ||
| if e != nil { | ||
| if terminate := concurrent.errorHandler.handleError(concurrent.config, concurrent.name, concurrent.id, | ||
| fmt.Sprintf("Concurrent job %v aggregation error: %v", concurrent.id, e.Error()), e); terminate != nil { | ||
| return *terminate | ||
| } | ||
| } else { | ||
| data = d | ||
| } | ||
| } else { | ||
| if terminate := concurrent.errorHandler.handleError(concurrent.config, concurrent.name, concurrent.id, | ||
| fmt.Sprintf("Concurrent job %v error: %v", concurrent.id, r.Message), r.Error); terminate != nil { | ||
| return *terminate | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return SuccessResultWithData(data) | ||
| } | ||
|
|
||
| func NewDefaultConcurrentJob(name string, defaultValue interface{}, jobs []Job, aggregator Aggregator) ConcurrentJob { | ||
| return NewConcurrentJob(name, defaultValue, jobs, aggregator, DefaultJobConfig(), NewDefaultErrorHandler(), NewDefaultSummary()) | ||
| } | ||
|
|
||
| func NewConcurrentJob(name string, defaultValue interface{}, jobs []Job, aggregator Aggregator, config JobConfig, | ||
| errorHandler ErrorHandler, summary Summary) ConcurrentJob { | ||
| return ConcurrentJob{ | ||
| id: uuid.New().String(), | ||
| name: name, | ||
| defaultValue: defaultValue, | ||
| jobs: jobs, | ||
| config: config, | ||
| aggregator: aggregator, | ||
| errorHandler: errorHandler, | ||
| summary: summary, | ||
| } | ||
| } |
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,50 @@ | ||
| package pipeline | ||
|
|
||
| import "fmt" | ||
|
|
||
| const ( | ||
| DefaultMaxConcurrency = 40 | ||
| ) | ||
|
|
||
| type JobConfig struct { | ||
| allowError bool | ||
| logError bool | ||
| summary bool | ||
| maxConcurrency int | ||
| } | ||
|
|
||
| func (config JobConfig) WithAllowError(allowError bool) JobConfig { | ||
| config.allowError = allowError | ||
| return config | ||
| } | ||
|
|
||
| func (config JobConfig) WithLogError(logError bool) JobConfig { | ||
| config.logError = logError | ||
| return config | ||
| } | ||
|
|
||
| func (config JobConfig) WithMaxConcurrency(maxConcurrency int) JobConfig { | ||
| if maxConcurrency > 0 && maxConcurrency <= DefaultMaxConcurrency { | ||
| config.maxConcurrency = maxConcurrency | ||
| } | ||
| return config | ||
| } | ||
|
|
||
| func (config JobConfig) WithSummary(summary bool) JobConfig { | ||
| config.summary = summary | ||
| return config | ||
| } | ||
|
|
||
| func (config JobConfig) String() string { | ||
| return fmt.Sprintf("AllowError: %v\nLogError: %v\nSummary: %v\nMaxConcurrency: %v\n", | ||
| config.allowError, config.logError, config.summary, config.maxConcurrency) | ||
| } | ||
|
|
||
| func DefaultJobConfig() JobConfig { | ||
| return JobConfig{ | ||
| allowError: true, | ||
| logError: true, | ||
| summary: true, | ||
| maxConcurrency: DefaultMaxConcurrency, | ||
| } | ||
| } |
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,29 @@ | ||
| package pipeline | ||
|
|
||
| import "github.com/sirupsen/logrus" | ||
|
|
||
| type ErrorHandler interface { | ||
| name() string | ||
| handleError(config JobConfig, name, id, msg string, e error) *JobResult | ||
| } | ||
|
|
||
| type DefaultErrorHandler struct{} | ||
|
|
||
| func (handler DefaultErrorHandler) name() string { | ||
| return "DefaultErrorHandler" | ||
| } | ||
|
|
||
| func (handler DefaultErrorHandler) handleError(config JobConfig, name, id, msg string, e error) *JobResult { | ||
| if config.logError { | ||
| logrus.WithField("name", name).WithField("id", id).Error(msg) | ||
| } | ||
| if !config.allowError { | ||
| r := FailureResult(e, msg) | ||
| return &r | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func NewDefaultErrorHandler() DefaultErrorHandler { | ||
| return DefaultErrorHandler{} | ||
| } |
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,20 @@ | ||
| package pipeline | ||
|
|
||
| import "context" | ||
|
|
||
| type Job interface { | ||
| // Do returns job execution result | ||
| // | ||
| // @params ctx Execution context | ||
| // @return Job execution result | ||
| Do(ctx context.Context) JobResult | ||
| } | ||
|
|
||
| // Aggregator aggregates a job's result with an accumulated result, and returns this final result | ||
| // | ||
| // @param ctx Execution context | ||
| // @param prior Accumulated result | ||
| // @param current Current job's result, which can be diffrent from prior | ||
| // @return The first value is the job result to be returned, which should be the same type as prior. | ||
| // @return The second value is a potential error | ||
| type Aggregator func(ctx context.Context, prior, current interface{}) (interface{}, error) |
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,27 @@ | ||
| package pipeline | ||
|
|
||
| type JobResult struct { | ||
| Success bool `json:"success"` | ||
| Message string `json:"message"` | ||
| Error error `json:"-"` | ||
| Data interface{} `json:"data"` | ||
| } | ||
|
|
||
| func SuccessResultWithData(data interface{}) JobResult { | ||
| return JobResult{ | ||
| Success: true, | ||
| Data: data, | ||
| } | ||
| } | ||
|
|
||
| func SuccessResult() JobResult { | ||
| return SuccessResultWithData(nil) | ||
| } | ||
|
|
||
| func FailureResult(e error, msg string) JobResult { | ||
| return JobResult{ | ||
| Success: false, | ||
| Message: msg, | ||
| Error: e, | ||
| } | ||
| } |
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,86 @@ | ||
| package pipeline | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/google/uuid" | ||
| ) | ||
|
|
||
| type SequentialJob struct { | ||
| id string | ||
| name string | ||
| jobs []Job | ||
| config JobConfig | ||
| defaultValue interface{} | ||
| aggregator Aggregator | ||
| errorHandler ErrorHandler | ||
| summary Summary | ||
| } | ||
|
|
||
| func (sequential SequentialJob) Do(ctx context.Context) JobResult { | ||
| start := time.Now() | ||
| r := sequential.do(ctx) | ||
| elapsed := time.Since(start).Milliseconds() | ||
| sequential.summary.summary(sequential.id, sequential.name, len(sequential.jobs), sequential.config, elapsed) | ||
| return r | ||
| } | ||
|
|
||
| func (sequential SequentialJob) do(ctx context.Context) JobResult { | ||
| ch := make(chan JobResult) | ||
| go func(c context.Context, jobs []Job) { | ||
| var data = sequential.defaultValue | ||
| defer func() { | ||
| if re := recover(); re != nil { | ||
| if r := sequential.errorHandler.handleError(sequential.config, sequential.name, sequential.id, | ||
| "Unexpected failure", fmt.Errorf("sequential job %v unexpected error: %v", sequential.id, re)); r != nil { | ||
| ch <- *r | ||
| } else { | ||
| ch <- SuccessResultWithData(data) | ||
| } | ||
| } | ||
| }() | ||
| for _, job := range sequential.jobs { | ||
| r := job.Do(ctx) | ||
| if r.Success { | ||
| d, e := sequential.aggregator(ctx, data, r.Data) | ||
| if e != nil { | ||
| if terminate := sequential.errorHandler.handleError(sequential.config, sequential.name, sequential.id, | ||
| fmt.Sprintf("Sequential job %v aggregation error: %v", sequential.id, e.Error()), e); terminate != nil { | ||
| ch <- *terminate | ||
| return | ||
| } | ||
| } else { | ||
| data = d | ||
| } | ||
| } else { | ||
| if terminate := sequential.errorHandler.handleError(sequential.config, sequential.name, sequential.id, | ||
| fmt.Sprintf("Sequential job %v error: %v", sequential.id, r.Message), r.Error); terminate != nil { | ||
| ch <- *terminate | ||
| return | ||
| } | ||
| } | ||
| } | ||
| ch <- SuccessResultWithData(data) | ||
| }(ctx, sequential.jobs) | ||
| return <-ch | ||
| } | ||
|
|
||
| func NewDefaultSequentialJob(name string, defaultValue interface{}, jobs []Job, aggregator Aggregator) SequentialJob { | ||
| return NewSequentialJob(name, defaultValue, jobs, aggregator, DefaultJobConfig(), NewDefaultErrorHandler(), NewDefaultSummary()) | ||
| } | ||
|
|
||
| func NewSequentialJob(name string, defaultValue interface{}, jobs []Job, aggregator Aggregator, config JobConfig, | ||
| errorHandler ErrorHandler, summary Summary) SequentialJob { | ||
| return SequentialJob{ | ||
| id: uuid.New().String(), | ||
| name: name, | ||
| jobs: jobs, | ||
| config: config, | ||
| defaultValue: defaultValue, | ||
| aggregator: aggregator, | ||
| errorHandler: errorHandler, | ||
| summary: summary, | ||
| } | ||
| } |
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,23 @@ | ||
| package pipeline | ||
|
|
||
| import ( | ||
| "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| type Summary interface { | ||
| summary(id, name string, jobs int, config JobConfig, elapsed int64) | ||
| } | ||
|
|
||
| type DefaultSummary struct{} | ||
|
|
||
| func (summary DefaultSummary) summary(id, name string, jobs int, config JobConfig, elapsed int64) { | ||
| if config.summary { | ||
| logrus.WithField("id", id).WithField("name", name).WithField("jobs count", jobs). | ||
| WithField("config", config.String()).WithField("elapsed", elapsed). | ||
| Info("Job summary details") | ||
| } | ||
| } | ||
|
|
||
| func NewDefaultSummary() DefaultSummary { | ||
| return DefaultSummary{} | ||
| } |
File renamed without changes.
File renamed without changes.
Oops, something went wrong.