aboutsummaryrefslogtreecommitdiffstats
path: root/modules/queue/queue.go
diff options
context:
space:
mode:
Diffstat (limited to 'modules/queue/queue.go')
-rw-r--r--modules/queue/queue.go43
1 files changed, 20 insertions, 23 deletions
diff --git a/modules/queue/queue.go b/modules/queue/queue.go
index d458a7d506..094699d4af 100644
--- a/modules/queue/queue.go
+++ b/modules/queue/queue.go
@@ -6,9 +6,8 @@ package queue
import (
"context"
- "encoding/json"
"fmt"
- "reflect"
+ "time"
)
// ErrInvalidConfiguration is called when there is invalid configuration for a queue
@@ -53,8 +52,11 @@ type Named interface {
Name() string
}
-// Queue defines an interface to save an issue indexer queue
+// Queue defines an interface of a queue-like item
+//
+// Queues will handle their own contents in the Run method
type Queue interface {
+ Flushable
Run(atShutdown, atTerminate func(context.Context, func()))
Push(Data) error
}
@@ -71,32 +73,27 @@ func NewDummyQueue(handler HandlerFunc, opts, exemplar interface{}) (Queue, erro
type DummyQueue struct {
}
-// Run starts to run the queue
+// Run does nothing
func (b *DummyQueue) Run(_, _ func(context.Context, func())) {}
-// Push pushes data to the queue
+// Push fakes a push of data to the queue
func (b *DummyQueue) Push(Data) error {
return nil
}
-func toConfig(exemplar, cfg interface{}) (interface{}, error) {
- if reflect.TypeOf(cfg).AssignableTo(reflect.TypeOf(exemplar)) {
- return cfg, nil
- }
+// Flush always returns nil
+func (b *DummyQueue) Flush(time.Duration) error {
+ return nil
+}
- configBytes, ok := cfg.([]byte)
- if !ok {
- configStr, ok := cfg.(string)
- if !ok {
- return nil, ErrInvalidConfiguration{cfg: cfg}
- }
- configBytes = []byte(configStr)
- }
- newVal := reflect.New(reflect.TypeOf(exemplar))
- if err := json.Unmarshal(configBytes, newVal.Interface()); err != nil {
- return nil, ErrInvalidConfiguration{cfg: cfg, err: err}
- }
- return newVal.Elem().Interface(), nil
+// FlushWithContext always returns nil
+func (b *DummyQueue) FlushWithContext(context.Context) error {
+ return nil
+}
+
+// IsEmpty asserts that the queue is empty
+func (b *DummyQueue) IsEmpty() bool {
+ return true
}
var queuesMap = map[Type]NewQueueFunc{DummyQueueType: NewDummyQueue}
@@ -123,7 +120,7 @@ func RegisteredTypesAsString() []string {
return types
}
-// NewQueue takes a queue Type and HandlerFunc some options and possibly an exemplar and returns a Queue or an error
+// NewQueue takes a queue Type, HandlerFunc, some options and possibly an exemplar and returns a Queue or an error
func NewQueue(queueType Type, handlerFunc HandlerFunc, opts, exemplar interface{}) (Queue, error) {
newFn, ok := queuesMap[queueType]
if !ok {