Merge pull request #1 from edadeal/delivery_wrapper

wrap delivery methods with concurrent safe alternatives

Author: bbrodriges

lepus.go

@@ -174,7 +174,7 @@ func (c *Channel) ConsumeMessages(queue, consumer string, autoAck, exclusive, no
 	go func() {
 		for msg := range d {
 			msg.Acknowledger = c
-			wd <- Delivery{msg}
+			wd <- Delivery{msg, 0}
 		}
 	}()
 
@@ -184,12 +184,49 @@ func (c *Channel) ConsumeMessages(queue, consumer string, autoAck, exclusive, no
 // Delivery is a superset of amqp.Delivery
 type Delivery struct {
 	amqp.Delivery
+	acked int32
+}
+
+// Nack is a concurrent safe wrapper around standart AMQP Nack
+func (d *Delivery) Nack(multiple, requeue bool) error {
+	if atomic.CompareAndSwapInt32(&d.acked, 0, 1) {
+		err := d.Delivery.Nack(multiple, requeue)
+		if err != nil {
+			atomic.StoreInt32(&d.acked, 0)
+		}
+		return err
+	}
+	return nil
+}
+
+// Ack is a concurrent safe wrapper around standart AMQP Ack
+func (d *Delivery) Ack(multiple bool) error {
+	if atomic.CompareAndSwapInt32(&d.acked, 0, 1) {
+		err := d.Delivery.Ack(multiple)
+		if err != nil {
+			atomic.StoreInt32(&d.acked, 0)
+		}
+		return err
+	}
+	return nil
+}
+
+// Reject is a concurrent safe wrapper around standart AMQP Reject
+func (d *Delivery) Reject(requeue bool) error {
+	if atomic.CompareAndSwapInt32(&d.acked, 0, 1) {
+		err := d.Delivery.Reject(requeue)
+		if err != nil {
+			atomic.StoreInt32(&d.acked, 0)
+		}
+		return err
+	}
+	return nil
 }
 
 // NackDelayed nacks message without requeue and publishes it again
 // without modification back to tail of queue
 func (d *Delivery) NackDelayed(multiple, mandatory, immediate bool) (State, error) {
-	ch, ok := d.Acknowledger.(*Channel)
+	ch, ok := d.Delivery.Acknowledger.(*Channel)
 	if !ok {
 		return StateUnknown, errors.New("Acknowledger is not of type *lepus.Channel")
 	}
@@ -199,20 +236,20 @@ func (d *Delivery) NackDelayed(multiple, mandatory, immediate bool) (State, erro
 		return StateUnknown, err
 	}
 
-	return ch.PublishAndWait(d.Exchange, d.RoutingKey, mandatory, immediate, amqp.Publishing{
-		Headers:         d.Headers,
-		ContentType:     d.ContentType,
-		ContentEncoding: d.ContentEncoding,
-		DeliveryMode:    d.DeliveryMode,
-		Priority:        d.Priority,
-		CorrelationId:   d.CorrelationId,
-		ReplyTo:         d.ReplyTo,
-		Expiration:      d.Expiration,
-		MessageId:       d.MessageId,
-		Timestamp:       d.Timestamp,
-		Type:            d.Type,
-		UserId:          d.UserId,
-		AppId:           d.AppId,
-		Body:            d.Body,
+	return ch.PublishAndWait(d.Delivery.Exchange, d.Delivery.RoutingKey, mandatory, immediate, amqp.Publishing{
+		Headers:         d.Delivery.Headers,
+		ContentType:     d.Delivery.ContentType,
+		ContentEncoding: d.Delivery.ContentEncoding,
+		DeliveryMode:    d.Delivery.DeliveryMode,
+		Priority:        d.Delivery.Priority,
+		CorrelationId:   d.Delivery.CorrelationId,
+		ReplyTo:         d.Delivery.ReplyTo,
+		Expiration:      d.Delivery.Expiration,
+		MessageId:       d.Delivery.MessageId,
+		Timestamp:       d.Delivery.Timestamp,
+		Type:            d.Delivery.Type,
+		UserId:          d.Delivery.UserId,
+		AppId:           d.Delivery.AppId,
+		Body:            d.Delivery.Body,
 	})
 }