@@ -14,7 +14,8 @@ import (
1414)
1515
1616// ErrNotAcked indicated that published messages was not acked by RabbitMQ
17- var ErrNotAcked = errors .New ("messge was not acked" )
17+ var ErrNotAcked = errors .New ("message was not acked" )
18+ var ErrTimedout = errors .New ("message was timed out" )
1819
1920type message struct {
2021 action string
@@ -23,20 +24,22 @@ type message struct {
2324
2425// producer holds a amqp connection and channel to publish messages to.
2526type producer struct {
26- m sync.Mutex
27- wg sync.WaitGroup
28- conn * connection
29- channel * amqplib.Channel
30- notifyConfirm chan amqplib.Confirmation
31- closeQueue chan bool
32- config ProducerConfig
27+ m sync.Mutex
28+ wg sync.WaitGroup
29+ conn * connection
30+ channel * amqplib.Channel
31+ notifyConfirm chan amqplib.Confirmation
32+ notifyChanClose chan * amqplib.Error
33+ closeQueue chan bool
34+ config ProducerConfig
3335
3436 internalQueue chan message
3537
3638 exchangeName string
3739
38- closed bool
39- closes []chan bool
40+ closed bool
41+ channelReady bool
42+ closes []chan bool
4043}
4144
4245// ProducerConfig to be used when creating a new producer.
@@ -59,7 +62,6 @@ func NewProducerConfig(c messaging.Connection, exchange string, config ProducerC
5962 config : config ,
6063 internalQueue : make (chan message , 2 ),
6164 exchangeName : exchange ,
62- notifyConfirm : make (chan amqplib.Confirmation ),
6365 closeQueue : make (chan bool ),
6466 }
6567
@@ -141,8 +143,14 @@ func (p *producer) Close() {
141143// and updates the channel listeners to reflect this.
142144func (p * producer ) changeChannel (channel * amqplib.Channel ) {
143145 p .channel = channel
146+
147+ p .notifyChanClose = make (chan * amqplib.Error )
148+ p .channel .NotifyClose (p .notifyChanClose )
149+
144150 p .notifyConfirm = make (chan amqplib.Confirmation )
145151 p .channel .NotifyPublish (p .notifyConfirm )
152+
153+ p .channelReady = true
146154}
147155
148156func (p * producer ) setupTopology () error {
@@ -197,14 +205,59 @@ func (p *producer) setupTopology() error {
197205 return nil
198206}
199207
208+ func (p * producer ) setChannelReady (ready bool ) {
209+ p .m .Lock ()
210+ defer p .m .Unlock ()
211+
212+ p .channelReady = ready
213+ }
214+
215+ func (p * producer ) isChannelReady () bool {
216+ p .m .Lock ()
217+ defer p .m .Unlock ()
218+
219+ return p .channelReady
220+ }
221+
222+ func (p * producer ) isConnected () bool {
223+ if ! p .conn .IsConnected () {
224+ return false
225+ }
226+
227+ return p .isChannelReady ()
228+ }
229+
230+ func (p * producer ) waitConnectionLost () bool {
231+ if ! p .isConnected () {
232+ return true
233+ }
234+
235+ defer p .setChannelReady (false )
236+
237+ select {
238+ case <- p .conn .NotifyConnectionClose ():
239+ log .Warn ("Producer connection closed" )
240+ return true
241+ case <- p .notifyChanClose :
242+ log .Warn ("Producer channel closed" )
243+ return false
244+ }
245+ }
246+
200247func (p * producer ) handleReestablishedConnnection () {
201248 rs := p .conn .NotifyReestablish ()
202249
203250 for ! p .isClosed () {
204- <- rs
251+ // true if connection is lot
252+ // false if channel connection is lost
253+ connectionLost := p .waitConnectionLost ()
205254
206- err := p .setupTopology ()
255+ if connectionLost {
256+ // Wait reconnect
257+ <- rs
258+ }
207259
260+ err := p .setupTopology ()
208261 if err != nil {
209262 log .WithFields (log.Fields {
210263 "type" : "goevents" ,
@@ -216,8 +269,8 @@ func (p *producer) handleReestablishedConnnection() {
216269}
217270
218271func (p * producer ) publishMessage (msg amqplib.Publishing , queue string ) (err error ) {
219- if ! p .conn . IsConnected () {
220- err = errors .New ("connection is not open" )
272+ if ! p .isConnected () {
273+ err = errors .New ("connection/channel is not open" )
221274 return
222275 }
223276
@@ -245,7 +298,12 @@ func (p *producer) publishMessage(msg amqplib.Publishing, queue string) (err err
245298 }
246299 }()
247300
248- err = p .channel .Publish (p .exchangeName , queue , false , false , msg )
301+ err = p .channel .Publish (
302+ p .exchangeName , // Exchange
303+ queue , // Routing key
304+ false , // Mandatory
305+ false , // Immediate
306+ msg )
249307
250308 if err != nil {
251309 return
@@ -260,7 +318,7 @@ func (p *producer) publishMessage(msg amqplib.Publishing, queue string) (err err
260318 err = ErrNotAcked
261319 return
262320 case <- time .After (p .config .publishInterval ):
263- err = ErrNotAcked
321+ err = ErrTimedout
264322 return
265323 }
266324
0 commit comments