-
Notifications
You must be signed in to change notification settings - Fork 143
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* add support for in-memory queue initialization * register workers using cfg and queue * add in-memory queue provider * add storage client * add storage client feature * use cfg to create worker * add in-memory queue * add local storage * update requirements * memqueue-check stop call * change eventdelivery declaration * Fix: merge conflicts * fix: merge conflicts * use QueueOptions to define structure * wrap queue.Type check in Queuer interface * fix pointer to redisclient * restructure toggle between memqueue and redisqueue * fix CI issues * generate queue mocks * check to start consumer * add test scripts * fix CI lint * fix: remove test consumer starts * verbose error on newclient function * add redis client step * fix typo and remove queue type check * set redis dsn on env, change port in testdata conovy.json * print verbose error messages * fix redis version * test: add dsn for mongo * single step for integration tests * use test table * pull from upstream * restructure test table * build tags * add build tags
- Loading branch information
Showing
17 changed files
with
577 additions
and
31 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 |
|---|---|---|
|
|
@@ -13,6 +13,7 @@ jobs: | |
| go-version: [1.16.x, 1.17.x] | ||
| os: [ubuntu-latest, macos-latest] | ||
| mongodb-version: ["4.0", "4.2", "4.4"] | ||
| redis-version: ["6.2.6"] | ||
|
|
||
| runs-on: ubuntu-latest | ||
| steps: | ||
|
|
@@ -21,6 +22,12 @@ jobs: | |
| with: | ||
| mongodb-version: ${{ matrix.mongodb-version }} | ||
|
|
||
| - name: Start Redis v${{ matrix.redis-version }} | ||
| uses: supercharge/[email protected] | ||
| with: | ||
| redis-version: ${{ matrix.redis-version }} | ||
| redis-port: 6379 | ||
|
|
||
| - name: Get the version | ||
| id: get_version | ||
| run: echo ::set-output name=tag::$(echo ${GITHUB_SHA:8}) | ||
|
|
@@ -49,8 +56,9 @@ jobs: | |
| - name: Go vet | ||
| run: go vet ./... | ||
|
|
||
| - name: Run Mongo integration tests | ||
| - name: Run integration tests | ||
| run: go test -tags integration -v ./... | ||
| env: | ||
| TEST_BOLT_DSN: "../../test.db" | ||
| TEST_MONGO_DSN: "mongodb://localhost:27017/testdb" | ||
| TEST_REDIS_DSN: "redis://localhost:6379" | ||
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
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
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,75 @@ | ||
| package memqueue | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "time" | ||
|
|
||
| "github.com/frain-dev/convoy" | ||
| "github.com/frain-dev/convoy/config" | ||
| "github.com/frain-dev/convoy/datastore" | ||
| "github.com/frain-dev/convoy/queue" | ||
| "github.com/vmihailenco/taskq/v3" | ||
| "github.com/vmihailenco/taskq/v3/memqueue" | ||
| ) | ||
|
|
||
| type MemQueue struct { | ||
| Name string | ||
| queue *memqueue.Queue | ||
| inner taskq.Factory | ||
| closeChan chan struct{} | ||
| } | ||
|
|
||
| func NewClient(cfg config.Configuration) (queue.Storage, taskq.Factory, error) { | ||
| if cfg.Queue.Type != config.InMemoryQueueProvider { | ||
| return nil, nil, errors.New("please select the in-memory queue in your config") | ||
| } | ||
|
|
||
| qFn := memqueue.NewFactory() | ||
|
|
||
| storage := queue.NewLocalStorage() | ||
|
|
||
| return storage, qFn, nil | ||
| } | ||
|
|
||
| func NewQueue(opts queue.QueueOptions) queue.Queuer { | ||
| q := opts.Factory.RegisterQueue(&taskq.QueueOptions{ | ||
| Name: opts.Name, | ||
| Storage: opts.Storage, | ||
| }) | ||
|
|
||
| return &MemQueue{ | ||
| Name: opts.Name, | ||
| inner: opts.Factory, | ||
| queue: q.(*memqueue.Queue), | ||
| } | ||
| } | ||
|
|
||
| func (q *MemQueue) Close() error { | ||
| q.closeChan <- struct{}{} | ||
| return q.inner.Close() | ||
| } | ||
|
|
||
| func (q *MemQueue) Write(ctx context.Context, name convoy.TaskName, e *datastore.EventDelivery, delay time.Duration) error { | ||
| job := &queue.Job{ | ||
| ID: e.UID, | ||
| } | ||
|
|
||
| m := &taskq.Message{ | ||
| Ctx: ctx, | ||
| TaskName: string(name), | ||
| Args: []interface{}{job}, | ||
| Delay: delay, | ||
| } | ||
|
|
||
| err := q.queue.Add(m) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (q *MemQueue) Consumer() taskq.QueueConsumer { | ||
| return q.queue.Consumer() | ||
| } |
Oops, something went wrong.