-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* feat: add otkafka/processor * Update go.yml * feat: add otkafka processor * test: use wait group for graceful shutdown * test: add lock for data * refactor: rename some variables and modify comment * fix: hound check * chore: update go.mod kafka-go * refactor: restructure and fix some bug * fix: single Handler's message missing commit Co-authored-by: 谷溪 <guxi99@gmail.com>
- Loading branch information
Showing
8 changed files
with
937 additions
and
1 deletion.
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 |
|---|---|---|
|
|
@@ -4,6 +4,7 @@ on: | |
| branches: | ||
| - master | ||
| pull_request: | ||
| workflow_dispatch: | ||
| jobs: | ||
| build: | ||
| strategy: | ||
|
|
||
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,80 @@ | ||
| package processor | ||
|
|
||
| import "time" | ||
|
|
||
| // Info the info of BatchHandler. | ||
| // | ||
| // Note: | ||
| // If sequence is necessary, make sure that per worker count is one. | ||
| // Multiple goroutines cannot guarantee the order in which data is processed. | ||
| type Info struct { | ||
| // used to get reader from otkafka.ReaderMaker. | ||
| // default: "default" | ||
| Name string | ||
| // reader workers count. | ||
| // default: 1 | ||
| ReadWorker int | ||
| // batch workers count. | ||
| // default: 1 | ||
| BatchWorker int | ||
| // data size for batch processing. | ||
| // default: 1 | ||
| BatchSize int | ||
| // handler workers count. | ||
| HandleWorker int | ||
| // the size of the data channel. | ||
| // default: 100 | ||
| ChanSize int | ||
| // run the batchFunc automatically at specified intervals, avoid not executing without reaching BatchSize | ||
| // default: 30s | ||
| AutoBatchInterval time.Duration | ||
| } | ||
|
|
||
| func (i *Info) name() string { | ||
| if i.Name == "" { | ||
| return "default" | ||
| } | ||
| return i.Name | ||
| } | ||
|
|
||
| func (i *Info) readWorker() int { | ||
| if i.ReadWorker <= 0 { | ||
| return 1 | ||
| } | ||
| return i.ReadWorker | ||
| } | ||
|
|
||
| func (i *Info) batchWorker() int { | ||
| if i.BatchWorker <= 0 { | ||
| return 1 | ||
| } | ||
| return i.BatchWorker | ||
| } | ||
|
|
||
| func (i *Info) batchSize() int { | ||
| if i.BatchSize <= 0 { | ||
| return 1 | ||
| } | ||
| return i.BatchSize | ||
| } | ||
|
|
||
| func (i *Info) handleWorker() int { | ||
| if i.HandleWorker <= 0 { | ||
| return 1 | ||
| } | ||
| return i.HandleWorker | ||
| } | ||
|
|
||
| func (i *Info) chanSize() int { | ||
| if i.ChanSize <= 0 { | ||
| return 100 | ||
| } | ||
| return i.ChanSize | ||
| } | ||
|
|
||
| func (i *Info) autoBatchInterval() time.Duration { | ||
| if i.AutoBatchInterval < 10 { | ||
| return 30 * time.Second | ||
| } | ||
| return i.AutoBatchInterval | ||
| } |
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,43 @@ | ||
| package processor | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestInfo(t *testing.T) { | ||
| i := Info{ | ||
| Name: "", | ||
| ReadWorker: 0, | ||
| BatchWorker: 0, | ||
| BatchSize: 0, | ||
| HandleWorker: 0, | ||
| ChanSize: 0, | ||
| } | ||
| assert.Equal(t, "default", i.name()) | ||
| assert.Equal(t, 1, i.readWorker()) | ||
| assert.Equal(t, 1, i.batchWorker()) | ||
| assert.Equal(t, 1, i.batchSize()) | ||
| assert.Equal(t, 1, i.handleWorker()) | ||
| assert.Equal(t, 100, i.chanSize()) | ||
| assert.Equal(t, 30*time.Second, i.autoBatchInterval()) | ||
|
|
||
| j := Info{ | ||
| Name: "test", | ||
| ReadWorker: 2, | ||
| BatchWorker: 2, | ||
| BatchSize: 10, | ||
| HandleWorker: 2, | ||
| ChanSize: 10, | ||
| AutoBatchInterval: 10 * time.Second, | ||
| } | ||
| assert.Equal(t, j.Name, j.name()) | ||
| assert.Equal(t, j.ReadWorker, j.readWorker()) | ||
| assert.Equal(t, j.BatchWorker, j.batchWorker()) | ||
| assert.Equal(t, j.BatchSize, j.batchSize()) | ||
| assert.Equal(t, j.HandleWorker, j.handleWorker()) | ||
| assert.Equal(t, j.ChanSize, j.chanSize()) | ||
| assert.Equal(t, j.AutoBatchInterval, j.autoBatchInterval()) | ||
| } |
Oops, something went wrong.