OLD | NEW |
1 // Copyright 2015 The Chromium Authors. All rights reserved. | 1 // Copyright 2015 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 package memory | 5 package memory |
6 | 6 |
7 import ( | 7 import ( |
8 "errors" | 8 "errors" |
9 "fmt" | 9 "fmt" |
10 "net/http" | 10 "net/http" |
(...skipping 106 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
117 } | 117 } |
118 | 118 |
119 var tqOkMethods = map[string]struct{}{ | 119 var tqOkMethods = map[string]struct{}{ |
120 "GET": {}, | 120 "GET": {}, |
121 "POST": {}, | 121 "POST": {}, |
122 "HEAD": {}, | 122 "HEAD": {}, |
123 "PUT": {}, | 123 "PUT": {}, |
124 "DELETE": {}, | 124 "DELETE": {}, |
125 } | 125 } |
126 | 126 |
127 func (t *taskQueueData) prepTask(c context.Context, ns string, task *tq.Task, qu
eueName string) (*tq.Task, string, error) { | 127 func (t *taskQueueData) prepTask(c context.Context, ns string, task *tq.Task, qu
eueName string) (*tq.Task, error) { |
128 » queueName, err := t.getQueueName(queueName) | 128 » toSched := task.Duplicate() |
129 » if err != nil { | |
130 » » return nil, "", err | |
131 » } | |
132 | |
133 » toSched := dupTask(task) | |
134 | 129 |
135 if toSched.Path == "" { | 130 if toSched.Path == "" { |
136 toSched.Path = "/_ah/queue/" + queueName | 131 toSched.Path = "/_ah/queue/" + queueName |
137 } | 132 } |
138 | 133 |
139 if toSched.ETA.IsZero() { | 134 if toSched.ETA.IsZero() { |
140 toSched.ETA = clock.Now(c).Add(toSched.Delay) | 135 toSched.ETA = clock.Now(c).Add(toSched.Delay) |
141 } else if toSched.Delay != 0 { | 136 } else if toSched.Delay != 0 { |
142 panic("taskqueue: both Delay and ETA are set") | 137 panic("taskqueue: both Delay and ETA are set") |
143 } | 138 } |
144 toSched.Delay = 0 | 139 toSched.Delay = 0 |
145 | 140 |
146 if toSched.Method == "" { | 141 if toSched.Method == "" { |
147 toSched.Method = "POST" | 142 toSched.Method = "POST" |
148 } | 143 } |
149 if _, ok := tqOkMethods[toSched.Method]; !ok { | 144 if _, ok := tqOkMethods[toSched.Method]; !ok { |
150 » » return nil, "", fmt.Errorf("taskqueue: bad method %q", toSched.M
ethod) | 145 » » return nil, fmt.Errorf("taskqueue: bad method %q", toSched.Metho
d) |
151 } | 146 } |
152 if toSched.Method != "POST" && toSched.Method != "PUT" { | 147 if toSched.Method != "POST" && toSched.Method != "PUT" { |
153 toSched.Payload = nil | 148 toSched.Payload = nil |
154 } | 149 } |
155 | 150 |
156 if _, ok := toSched.Header[currentNamespace]; !ok { | 151 if _, ok := toSched.Header[currentNamespace]; !ok { |
157 if ns != "" { | 152 if ns != "" { |
158 if toSched.Header == nil { | 153 if toSched.Header == nil { |
159 toSched.Header = http.Header{} | 154 toSched.Header = http.Header{} |
160 } | 155 } |
161 toSched.Header[currentNamespace] = []string{ns} | 156 toSched.Header[currentNamespace] = []string{ns} |
162 } | 157 } |
163 } | 158 } |
164 // TODO(riannucci): implement DefaultNamespace | 159 // TODO(riannucci): implement DefaultNamespace |
165 | 160 |
166 if toSched.Name == "" { | 161 if toSched.Name == "" { |
167 toSched.Name = mkName(c, "", t.named[queueName]) | 162 toSched.Name = mkName(c, "", t.named[queueName]) |
168 } else { | 163 } else { |
169 if !validTaskName.MatchString(toSched.Name) { | 164 if !validTaskName.MatchString(toSched.Name) { |
170 » » » return nil, "", errors.New("INVALID_TASK_NAME") | 165 » » » return nil, errors.New("INVALID_TASK_NAME") |
171 } | 166 } |
172 } | 167 } |
173 | 168 |
174 » return toSched, queueName, nil | 169 » return toSched, nil |
175 } | 170 } |
176 | 171 |
177 /////////////////////////////// txnTaskQueueData /////////////////////////////// | 172 /////////////////////////////// txnTaskQueueData /////////////////////////////// |
178 | 173 |
179 type txnTaskQueueData struct { | 174 type txnTaskQueueData struct { |
180 lock sync.Mutex | 175 lock sync.Mutex |
181 | 176 |
182 // boolean 0 or 1, use atomic.*Int32 to access. | 177 // boolean 0 or 1, use atomic.*Int32 to access. |
183 closed int32 | 178 closed int32 |
184 anony tq.AnonymousQueueData | 179 anony tq.AnonymousQueueData |
185 parent *taskQueueData | 180 parent *taskQueueData |
186 } | 181 } |
187 | 182 |
188 var ( | 183 var ( |
189 _ = memContextObj((*txnTaskQueueData)(nil)) | 184 _ = memContextObj((*txnTaskQueueData)(nil)) |
190 _ = tq.Testable((*txnTaskQueueData)(nil)) | 185 _ = tq.Testable((*txnTaskQueueData)(nil)) |
191 ) | 186 ) |
192 | 187 |
193 func (t *txnTaskQueueData) canApplyTxn(obj memContextObj) bool { return
false } | 188 func (t *txnTaskQueueData) canApplyTxn(obj memContextObj) bool { return
false } |
194 func (t *txnTaskQueueData) applyTxn(context.Context, memContextObj) { panic("
impossible") } | 189 func (t *txnTaskQueueData) applyTxn(context.Context, memContextObj) { panic("
impossible") } |
195 func (t *txnTaskQueueData) mkTxn(*ds.TransactionOptions) memContextObj { panic("
impossible") } | 190 func (t *txnTaskQueueData) mkTxn(*ds.TransactionOptions) memContextObj { panic("
impossible") } |
196 | 191 |
197 func (t *txnTaskQueueData) endTxn() { | 192 func (t *txnTaskQueueData) endTxn() { |
198 if atomic.LoadInt32(&t.closed) == 1 { | 193 if atomic.LoadInt32(&t.closed) == 1 { |
199 panic("cannot end transaction twice") | 194 panic("cannot end transaction twice") |
200 } | 195 } |
201 atomic.StoreInt32(&t.closed, 1) | 196 atomic.StoreInt32(&t.closed, 1) |
202 } | 197 } |
203 | 198 |
204 func (t *txnTaskQueueData) run(f func() error) error { | |
205 // Slightly different from the SDK... datastore and taskqueue each imple
ment | |
206 // this here, where in the SDK only datastore.transaction.Call does. | |
207 if atomic.LoadInt32(&t.closed) == 1 { | |
208 return fmt.Errorf("taskqueue: transaction context has expired") | |
209 } | |
210 return f() | |
211 } | |
212 | |
213 func (t *txnTaskQueueData) ResetTasks() { | 199 func (t *txnTaskQueueData) ResetTasks() { |
214 t.Lock() | 200 t.Lock() |
215 defer t.Unlock() | 201 defer t.Unlock() |
216 | 202 |
217 for queuename := range t.anony { | 203 for queuename := range t.anony { |
218 t.anony[queuename] = nil | 204 t.anony[queuename] = nil |
219 } | 205 } |
220 t.parent.resetTasksWithLock() | 206 t.parent.resetTasksWithLock() |
221 } | 207 } |
222 | 208 |
223 func (t *txnTaskQueueData) Lock() { | 209 func (t *txnTaskQueueData) Lock() { |
224 t.lock.Lock() | 210 t.lock.Lock() |
225 t.parent.Lock() | 211 t.parent.Lock() |
226 } | 212 } |
227 func (t *txnTaskQueueData) Unlock() { | 213 func (t *txnTaskQueueData) Unlock() { |
228 t.parent.Unlock() | 214 t.parent.Unlock() |
229 t.lock.Unlock() | 215 t.lock.Unlock() |
230 } | 216 } |
231 | 217 |
232 func (t *txnTaskQueueData) GetTransactionTasks() tq.AnonymousQueueData { | 218 func (t *txnTaskQueueData) GetTransactionTasks() tq.AnonymousQueueData { |
233 t.Lock() | 219 t.Lock() |
234 defer t.Unlock() | 220 defer t.Unlock() |
235 | 221 |
236 ret := make(tq.AnonymousQueueData, len(t.anony)) | 222 ret := make(tq.AnonymousQueueData, len(t.anony)) |
237 for k, vs := range t.anony { | 223 for k, vs := range t.anony { |
238 ret[k] = make([]*tq.Task, len(vs)) | 224 ret[k] = make([]*tq.Task, len(vs)) |
239 for i, v := range vs { | 225 for i, v := range vs { |
240 » » » tsk := dupTask(v) | 226 » » » tsk := v.Duplicate() |
241 tsk.Name = "" | 227 tsk.Name = "" |
242 ret[k][i] = tsk | 228 ret[k][i] = tsk |
243 } | 229 } |
244 } | 230 } |
245 | 231 |
246 return ret | 232 return ret |
247 } | 233 } |
248 | 234 |
249 func (t *txnTaskQueueData) GetTombstonedTasks() tq.QueueData { | 235 func (t *txnTaskQueueData) GetTombstonedTasks() tq.QueueData { |
250 return t.parent.GetTombstonedTasks() | 236 return t.parent.GetTombstonedTasks() |
251 } | 237 } |
252 | 238 |
253 func (t *txnTaskQueueData) GetScheduledTasks() tq.QueueData { | 239 func (t *txnTaskQueueData) GetScheduledTasks() tq.QueueData { |
254 return t.parent.GetScheduledTasks() | 240 return t.parent.GetScheduledTasks() |
255 } | 241 } |
256 | 242 |
257 func (t *txnTaskQueueData) CreateQueue(queueName string) { | 243 func (t *txnTaskQueueData) CreateQueue(queueName string) { |
258 t.parent.CreateQueue(queueName) | 244 t.parent.CreateQueue(queueName) |
259 } | 245 } |
OLD | NEW |