1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
// Copyright 2020 The Gitea Authors. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
package queue
import (
"fmt"
"sync"
"time"
)
// WrappedUniqueQueueType is the type for a wrapped delayed starting queue
const WrappedUniqueQueueType Type = "unique-wrapped"
// WrappedUniqueQueueConfiguration is the configuration for a WrappedUniqueQueue
type WrappedUniqueQueueConfiguration struct {
Underlying Type
Timeout time.Duration
MaxAttempts int
Config interface{}
QueueLength int
Name string
}
// WrappedUniqueQueue wraps a delayed starting unique queue
type WrappedUniqueQueue struct {
*WrappedQueue
table map[Data]bool
tlock sync.Mutex
ready bool
}
// NewWrappedUniqueQueue will attempt to create a unique queue of the provided type,
// but if there is a problem creating this queue it will instead create
// a WrappedUniqueQueue with delayed startup of the queue instead and a
// channel which will be redirected to the queue
//
// Please note that this Queue does not guarantee that a particular
// task cannot be processed twice or more at the same time. Uniqueness is
// only guaranteed whilst the task is waiting in the queue.
func NewWrappedUniqueQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error) {
configInterface, err := toConfig(WrappedUniqueQueueConfiguration{}, cfg)
if err != nil {
return nil, err
}
config := configInterface.(WrappedUniqueQueueConfiguration)
queue, err := NewQueue(config.Underlying, handle, config.Config, exemplar)
if err == nil {
// Just return the queue there is no need to wrap
return queue, nil
}
if IsErrInvalidConfiguration(err) {
// Retrying ain't gonna make this any better...
return nil, ErrInvalidConfiguration{cfg: cfg}
}
wrapped := &WrappedUniqueQueue{
WrappedQueue: &WrappedQueue{
channel: make(chan Data, config.QueueLength),
exemplar: exemplar,
delayedStarter: delayedStarter{
cfg: config.Config,
underlying: config.Underlying,
timeout: config.Timeout,
maxAttempts: config.MaxAttempts,
name: config.Name,
},
},
table: map[Data]bool{},
}
// wrapped.handle is passed to the delayedStarting internal queue and is run to handle
// data passed to
wrapped.handle = func(data ...Data) (unhandled []Data) {
for _, datum := range data {
wrapped.tlock.Lock()
if !wrapped.ready {
delete(wrapped.table, data)
// If our table is empty all of the requests we have buffered between the
// wrapper queue starting and the internal queue starting have been handled.
// We can stop buffering requests in our local table and just pass Push
// direct to the internal queue
if len(wrapped.table) == 0 {
wrapped.ready = true
}
}
wrapped.tlock.Unlock()
if u := handle(datum); u != nil {
unhandled = append(unhandled, u...)
}
}
return unhandled
}
_ = GetManager().Add(queue, WrappedUniqueQueueType, config, exemplar)
return wrapped, nil
}
// Push will push the data to the internal channel checking it against the exemplar
func (q *WrappedUniqueQueue) Push(data Data) error {
return q.PushFunc(data, nil)
}
// PushFunc will push the data to the internal channel checking it against the exemplar
func (q *WrappedUniqueQueue) PushFunc(data Data, fn func() error) error {
if !assignableTo(data, q.exemplar) {
return fmt.Errorf("unable to assign data: %v to same type as exemplar: %v in %s", data, q.exemplar, q.name)
}
q.tlock.Lock()
if q.ready {
// ready means our table is empty and all of the requests we have buffered between the
// wrapper queue starting and the internal queue starting have been handled.
// We can stop buffering requests in our local table and just pass Push
// direct to the internal queue
q.tlock.Unlock()
return q.internal.(UniqueQueue).PushFunc(data, fn)
}
locked := true
defer func() {
if locked {
q.tlock.Unlock()
}
}()
if _, ok := q.table[data]; ok {
return ErrAlreadyInQueue
}
// FIXME: We probably need to implement some sort of limit here
// If the downstream queue blocks this table will grow without limit
q.table[data] = true
if fn != nil {
err := fn()
if err != nil {
delete(q.table, data)
return err
}
}
locked = false
q.tlock.Unlock()
q.channel <- data
return nil
}
// Has checks if the data is in the queue
func (q *WrappedUniqueQueue) Has(data Data) (bool, error) {
q.tlock.Lock()
defer q.tlock.Unlock()
if q.ready {
return q.internal.(UniqueQueue).Has(data)
}
_, has := q.table[data]
return has, nil
}
// IsEmpty checks whether the queue is empty
func (q *WrappedUniqueQueue) IsEmpty() bool {
q.tlock.Lock()
if len(q.table) > 0 {
q.tlock.Unlock()
return false
}
if q.ready {
q.tlock.Unlock()
return q.internal.IsEmpty()
}
q.tlock.Unlock()
return false
}
func init() {
queuesMap[WrappedUniqueQueueType] = NewWrappedUniqueQueue
}
|
