want signals that an acquiree wants clearance to query for a job with the given dKey.
(organization uuid.UUID, pt []database.ProvisionerType, tags Tags, clearance chan<- struct{})
| 155 | |
| 156 | // want signals that an acquiree wants clearance to query for a job with the given dKey. |
| 157 | func (a *Acquirer) want(organization uuid.UUID, pt []database.ProvisionerType, tags Tags, clearance chan<- struct{}) { |
| 158 | dk := domainKey(organization, pt, tags) |
| 159 | a.mu.Lock() |
| 160 | defer a.mu.Unlock() |
| 161 | cleared := false |
| 162 | d, ok := a.q[dk] |
| 163 | if !ok { |
| 164 | ctx, cancel := context.WithCancel(a.ctx) |
| 165 | d = domain{ |
| 166 | ctx: ctx, |
| 167 | cancel: cancel, |
| 168 | a: a, |
| 169 | key: dk, |
| 170 | pt: pt, |
| 171 | tags: tags, |
| 172 | organizationID: organization, |
| 173 | acquirees: make(map[chan<- struct{}]*acquiree), |
| 174 | } |
| 175 | a.q[dk] = d |
| 176 | go d.poll(a.backupPollDuration) |
| 177 | // this is a new request for this dKey, so is cleared. |
| 178 | cleared = true |
| 179 | } |
| 180 | w, ok := d.acquirees[clearance] |
| 181 | if !ok { |
| 182 | w = &acquiree{clearance: clearance} |
| 183 | d.acquirees[clearance] = w |
| 184 | } |
| 185 | // pending means that we got a job posting for this dKey while we were |
| 186 | // querying, so we should clear this acquiree to retry another time. |
| 187 | if w.pending { |
| 188 | cleared = true |
| 189 | w.pending = false |
| 190 | } |
| 191 | w.inProgress = cleared |
| 192 | if cleared { |
| 193 | // this won't block because clearance is buffered. |
| 194 | clearance <- struct{}{} |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | // cancel signals that an acquiree no longer wants clearance to query. Any error returned is a serious internal error |
| 199 | // indicating that integrity of the internal state is corrupted by a code bug. |