/usr/share/gocode/src/github.com/streadway/amqp/confirms.go is in golang-github-streadway-amqp-dev 0.0~git20150820.0.f4879ba-6.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | package amqp
import "sync"
// confirms resequences and notifies one or multiple publisher confirmation listeners
type confirms struct {
m sync.Mutex
listeners []chan Confirmation
sequencer map[uint64]Confirmation
published uint64
expecting uint64
}
// newConfirms allocates a confirms
func newConfirms() *confirms {
return &confirms{
sequencer: map[uint64]Confirmation{},
published: 0,
expecting: 1,
}
}
func (c *confirms) Listen(l chan Confirmation) {
c.m.Lock()
defer c.m.Unlock()
c.listeners = append(c.listeners, l)
}
// publish increments the publishing counter
func (c *confirms) Publish() uint64 {
c.m.Lock()
defer c.m.Unlock()
c.published++
return c.published
}
// confirm confirms one publishing, increments the expecting delivery tag, and
// removes bookkeeping for that delivery tag.
func (c *confirms) confirm(confirmation Confirmation) {
delete(c.sequencer, c.expecting)
c.expecting++
for _, l := range c.listeners {
l <- confirmation
}
}
// resequence confirms any out of order delivered confirmations
func (c *confirms) resequence() {
for c.expecting <= c.published {
sequenced, found := c.sequencer[c.expecting]
if !found {
return
}
c.confirm(sequenced)
}
}
// one confirms one publishing and all following in the publishing sequence
func (c *confirms) One(confirmed Confirmation) {
c.m.Lock()
defer c.m.Unlock()
if c.expecting == confirmed.DeliveryTag {
c.confirm(confirmed)
} else {
c.sequencer[confirmed.DeliveryTag] = confirmed
}
c.resequence()
}
// multiple confirms all publishings up until the delivery tag
func (c *confirms) Multiple(confirmed Confirmation) {
c.m.Lock()
defer c.m.Unlock()
for c.expecting <= confirmed.DeliveryTag {
c.confirm(Confirmation{c.expecting, confirmed.Ack})
}
}
// Close closes all listeners, discarding any out of sequence confirmations
func (c *confirms) Close() error {
c.m.Lock()
defer c.m.Unlock()
for _, l := range c.listeners {
close(l)
}
c.listeners = nil
return nil
}
|