POC to check AMQP performances with che and axe

cmd/engine-axe/engine.go

@@ -20,6 +20,7 @@ type EngineAxe struct {
 	canopsis.DefaultEngine
 	Options    Options
 	References References
+	Messages   int
 }
 
 func (e *EngineAxe) ConsumerChan() (<-chan amqp.Delivery, error) {
@@ -185,11 +186,15 @@ func (e *EngineAxe) PeriodicalProcess() {
 }
 
 func (e *EngineAxe) ackMsg(msg amqp.Delivery) error {
-	err := e.References.ChannelSub.Ack(msg.DeliveryTag, false)
-	if err != nil {
-		e.DebugPrint("ack error: %+v", err)
+	e.Messages++
+	if e.Messages%10 == 0 {
+		err := e.References.ChannelSub.Ack(msg.DeliveryTag, true)
+		if err != nil {
+			e.DebugPrint("ack error: %+v", err)
+		}
+		return err
 	}
-	return err
+	return nil
 }
 
 func (e *EngineAxe) processWorkerError(err error, msg amqp.Delivery) error {

cmd/engine-che/engine.go

@@ -19,6 +19,7 @@ type EngineChe struct {
 	canopsis.DefaultEngine
 	Options    Options
 	References References
+	Messages   int
 }
 
 func (e *EngineChe) ConsumerChan() (<-chan amqp.Delivery, error) {
@@ -87,11 +88,15 @@ func (e *EngineChe) PeriodicalProcess() {
 }
 
 func (e *EngineChe) ackMsg(msg amqp.Delivery) error {
-	err := e.References.ChannelSub.Ack(msg.DeliveryTag, false)
-	if err != nil {
-		e.DebugPrint("ack error: %+v", err)
+	e.Messages++
+	if e.Messages%10 == 0 {
+		err := e.References.ChannelSub.Ack(msg.DeliveryTag, true)
+		if err != nil {
+			e.DebugPrint("ack error: %+v", err)
+		}
+		return err
 	}
-	return err
+	return nil
 }
 
 // processWorkerError read errors, ack amqp messages and stop the engine if needed