forked from siddontang/moonmq
-
Notifications
You must be signed in to change notification settings - Fork 0
/
channel.go
88 lines (69 loc) · 1.26 KB
/
channel.go
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
package client
import (
"errors"
"time"
)
var ErrChannelClosed = errors.New("channel has been closed")
type channelMsg struct {
ID string
Body []byte
}
type Channel struct {
c *Conn
queue string
routingKey string
noAck bool
msg chan *channelMsg
closed bool
lastId string
}
func newChannel(c *Conn, queue string, routingKey string, noAck bool) *Channel {
ch := new(Channel)
ch.c = c
ch.queue = queue
ch.routingKey = routingKey
ch.noAck = noAck
ch.msg = make(chan *channelMsg, c.cfg.MaxQueueSize)
ch.closed = false
return ch
}
func (c *Channel) Close() error {
c.closed = true
return c.c.unbind(c.queue)
}
func (c *Channel) Ack() error {
if c.closed {
return ErrChannelClosed
}
return c.c.ack(c.queue, c.lastId)
}
func (c *Channel) GetMsg() []byte {
if c.closed && len(c.msg) == 0 {
return nil
}
msg := <-c.msg
c.lastId = msg.ID
return msg.Body
}
func (c *Channel) WaitMsg(d time.Duration) []byte {
if c.closed && len(c.msg) == 0 {
return nil
}
select {
case <-time.After(d):
return nil
case msg := <-c.msg:
c.lastId = msg.ID
return msg.Body
}
}
func (c *Channel) pushMsg(msgId string, body []byte) {
for {
select {
case c.msg <- &channelMsg{msgId, body}:
return
default:
<-c.msg
}
}
}