Added ability to ack specific messages

Author: luneo7

README.md

@@ -2,20 +2,17 @@
 
 Dump messages from a RabbitMQ queue to files, without affecting the queue.
 
-## Installation
-
-### Download a release
+This work is derivated from https://github.com/dubek/rabbitmq-dump-queue
 
-Precompiled binary packages can be found on the
-[releases](https://github.com/dubek/rabbitmq-dump-queue/releases) page.
+## Installation
 
 ### Compile from source
 
 If you have [Go](https://golang.org/doc/install) installed, you can install
 rabbitmq-dump-queue from source by running:
 
 ```
-go get github.com/dubek/rabbitmq-dump-queue
+go get github.com/luneo7/rabbitmq-dump-queue
 ```
 
 The `rabbitmq-dump-queue` executable will be created in the `$GOPATH/bin`
@@ -28,41 +25,57 @@ To dump the first 50 messages of queue `incoming_1` to `/tmp`:
 
     rabbitmq-dump-queue -uri="amqp://user:[email protected]:5672/" -queue=incoming_1 -max-messages=50 -output-dir=/tmp
 
-This will create the files `/tmp/msg-0000`, `/tmp/msg-0001`, and so on.
+This will create the files `/tmp/msg-0000.json`, `/tmp/msg-0001.json`, and so on.
 
 The output filenames are printed one per line to the standard output; this
 allows piping the output of rabbitmq-dump-queue to `xargs` or similar utilities
 in order to perform further processing on each message (e.g. decompressing,
 decoding, etc.).
 
-To include the AMQP headers and properties in the output, add the `-full`
-option to the command-line.  This will create the following files:
+By default, it will not acknowledge messages, so they will be requeued.
+Acknowledging messages using the `-ack=true` switch will *remove* them from the
+queue, allowing the user to process new messages (see implementation details).
 
-    /tmp/msg-0000
-    /tmp/msg-0000-headers+properties.json
-    /tmp/msg-0001
-    /tmp/msg-0001-headers+properties.json
-    ...
+    rabbitmq-dump-queue -uri="amqp://user:[email protected]:5672/" -queue=incoming_1 -max-messages=50 -output-dir=/tmp -ack=true
 
-The JSON files have the following structure:
+If you want to acknowledge a list of specific messages you can create a JSON file
+with the following structure
 
     {
-      "headers": {
-        "x-my-private-header": "my-value"
-      },
-      "properties": {
-        "correlation_id": "XYZ-9876",
-        "delivery_mode": 0,
-        "priority": 5
-      }
+        "messages": [
+            {
+                "routingKey" : "testEvent",
+                "containsString" : "[email protected]",
+                "contains" : [
+                    {
+                        "key": "fieldKeyInContent1",
+                        "value": "fieldValueInContent"
+                    },
+                    {
+                        "key": "fieldKeyInContent2",
+                        "value": 12345
+                    },
+                    {
+                        "key": "fieldKeyInContent3",
+                        "value": true
+                    }
+                ]
+            }
+        ]
     }
 
+In the above JSON `routingKey` is mandatory, `containsString` and `contains`
+are optional. So you can have only `containsString` or `contains` with the
+proper values. To use the acknowledge json file you should use the
+`-messages-to-ack` parameter:
 
-By default, it will not acknowledge messages, so they will be requeued.
-Acknowledging messages using the `-ack=true` switch will *remove* them from the
-queue, allowing the user to process new messages (see implementation details).
+    rabbitmq-dump-queue -uri="amqp://user:[email protected]:5672/" -queue=incoming_1 -max-messages=50 -output-dir=/tmp -messages-to-ack=/tmp/ackmessages.json
 
-    rabbitmq-dump-queue -uri="amqp://user:[email protected]:5672/" -queue=incoming_1 -max-messages=50 -output-dir=/tmp -ack=true
+
+`rabbitmq-dump-queue` is expecting a JSON response in the content of the message
+if it's not the case you can disable it with `json-content` paramater
+
+    rabbitmq-dump-queue -uri="amqp://user:[email protected]:5672/" -queue=incoming_1 -max-messages=50 -output-dir=/tmp -json-content=false
 
 Running `rabbitmq-dump-queue -help` will list the available command-line
 options.

ackmessageexample1.json

@@ -0,0 +1,8 @@
+{
+  "messages": [
+    {
+      "routingKey" : "test",
+      "containsString" : "[email protected]"
+    }
+  ]
+}

ackmessageexample2.json

@@ -0,0 +1,13 @@
+{
+  "messages": [
+    {
+      "routingKey" : "test",
+      "contains" : [
+        {
+          "key": "id",
+          "value": "6276b5b4-81d5-4e04-b58f-f1afd15a9c5d"
+        }
+      ]
+    }
+  ]
+}

ackmessageexample3.json

@@ -0,0 +1,22 @@
+{
+  "messages": [
+    {
+      "routingKey" : "test",
+      "containsString" : "[email protected]",
+      "contains" : [
+        {
+          "key": "id",
+          "value": "6276b5b4-81d5-4e04-b58f-f1afd15a9c5d"
+        },
+        {
+          "key": "checked",
+          "value": true
+        },
+        {
+          "key": "number",
+          "value": 123456
+        }
+      ]
+    }
+  ]
+}

main.go

@@ -8,30 +8,47 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
+	"strconv"
 	"strings"
 
 	"github.com/streadway/amqp"
 )
 
 var (
-	uri          = flag.String("uri", "amqp://guest:guest@localhost:5672/", "AMQP URI")
-	insecure_tls = flag.Bool("insecure-tls", false, "Insecure TLS mode: don't check certificates")
-	queue        = flag.String("queue", "", "AMQP queue name")
-	ack          = flag.Bool("ack", false, "Acknowledge messages")
-	maxMessages  = flag.Uint("max-messages", 1000, "Maximum number of messages to dump")
-	outputDir    = flag.String("output-dir", ".", "Directory in which to save the dumped messages")
-	full         = flag.Bool("full", false, "Dump the message, its properties and headers")
-	verbose      = flag.Bool("verbose", false, "Print progress")
+	uri               = flag.String("uri", "amqp://guest:guest@localhost:5672/", "AMQP URI")
+	insecure_tls      = flag.Bool("insecure-tls", false, "Insecure TLS mode: don't check certificates")
+	queue             = flag.String("queue", "", "AMQP queue name")
+	ack               = flag.Bool("ack", false, "Acknowledge messages")
+	maxMessages       = flag.Uint("max-messages", 1000, "Maximum number of messages to dump")
+	outputDir         = flag.String("output-dir", ".", "Directory in which to save the dumped messages")
+	jsonContent       = flag.Bool("json-content", true, "If the content of the message is a JSON object")
+	verbose           = flag.Bool("verbose", false, "Print progress")
+	messagesToAckFile = flag.String("messages-to-ack", "", "File with messages to ack")
 )
 
+type Messages struct {
+	Messages []Message `json:"messages"`
+}
+
+type Message struct {
+	RoutingKey     string    `json:"routingKey"`
+	Contains       []Contain `json:"contains"`
+	ContainsString string    `json:"containsString"`
+}
+
+type Contain struct {
+	Key   string      `json:"key"`
+	Value interface{} `json:"value"`
+}
+
 func main() {
 	flag.Parse()
 	if flag.NArg() > 0 {
 		fmt.Fprintf(os.Stderr, "Error: Unused command line arguments detected.\n")
 		flag.Usage()
 		os.Exit(2)
 	}
-	err := DumpMessagesFromQueue(*uri, *queue, *maxMessages, *outputDir)
+	err := DumpMessagesFromQueue(*uri, *queue, *maxMessages, *outputDir, *messagesToAckFile, *jsonContent)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "%s\n", err)
 		os.Exit(1)
@@ -50,7 +67,28 @@ func dial(amqpURI string) (*amqp.Connection, error) {
 	return conn, err
 }
 
-func DumpMessagesFromQueue(amqpURI string, queueName string, maxMessages uint, outputDir string) error {
+func DumpMessagesFromQueue(amqpURI string, queueName string, maxMessages uint, outputDir string, messagesToAckFilePath string, isContentJSON bool) error {
+	var messages Messages
+	var ackMessage bool
+	searchToAck := false
+
+	if messagesToAckFilePath != "" {
+		jsonFile, err := os.Open(messagesToAckFilePath)
+		if err != nil {
+			return fmt.Errorf("Opening messages to ack file: %s", err)
+		} else {
+			byteValue, err := ioutil.ReadAll(jsonFile)
+			if err != nil {
+				return fmt.Errorf("Reding sessages to ack file: %s", err)
+			}
+
+			json.Unmarshal(byteValue, &messages)
+			jsonFile.Close()
+			byteValue = nil
+			searchToAck = true
+		}
+	}
+
 	if queueName == "" {
 		return fmt.Errorf("Must supply queue name")
 	}
@@ -72,8 +110,9 @@ func DumpMessagesFromQueue(amqpURI string, queueName string, maxMessages uint, o
 
 	VerboseLog(fmt.Sprintf("Pulling messages from queue %q", queueName))
 	for messagesReceived := uint(0); messagesReceived < maxMessages; messagesReceived++ {
+		ackMessage = *ack
 		msg, ok, err := channel.Get(queueName,
-			*ack, // autoAck
+			ackMessage, // autoAck
 		)
 		if err != nil {
 			return fmt.Errorf("Queue get: %s", err)
@@ -84,31 +123,54 @@ func DumpMessagesFromQueue(amqpURI string, queueName string, maxMessages uint, o
 			break
 		}
 
-		err = SaveMessageToFile(msg.Body, outputDir, messagesReceived)
-		if err != nil {
-			return fmt.Errorf("Save message: %s", err)
-		}
-
-		if *full {
-			err = SavePropsAndHeadersToFile(msg, outputDir, messagesReceived)
-			if err != nil {
-				return fmt.Errorf("Save props and headers: %s", err)
+		if searchToAck {
+			message_content := string(msg.Body[:])
+			for i := len(messages.Messages) - 1; i >= 0; i-- {
+				if msg.RoutingKey == messages.Messages[i].RoutingKey {
+					containsNeeded := len(messages.Messages[i].Contains)
+					containsFound := 0
+					for j := 0; j < containsNeeded; j++ {
+						search := ""
+						switch messages.Messages[i].Contains[j].Value.(type) {
+						case string:
+							search = "\"" + messages.Messages[i].Contains[j].Key + "\":\"" + messages.Messages[i].Contains[j].Value.(string) + "\""
+						case float64:
+							val := messages.Messages[i].Contains[j].Value.(float64)
+							fmtp := "%." + strconv.Itoa(NumDecPlaces(val)) + "f"
+							search = "\"" + messages.Messages[i].Contains[j].Key + "\":" + fmt.Sprintf(fmtp, val)
+						case bool:
+							search = "\"" + messages.Messages[i].Contains[j].Key + "\":" + fmt.Sprintf("%t", messages.Messages[i].Contains[j].Value.(bool))
+						}
+
+						if search != "" && strings.Contains(message_content, search) {
+							containsFound++
+						}
+					}
+
+					if messages.Messages[i].ContainsString != "" {
+						containsNeeded++
+						if strings.Contains(message_content, messages.Messages[i].ContainsString) {
+							containsFound++
+						}
+					}
+
+					if containsNeeded == containsFound {
+						msg.Ack(true)
+						fmt.Printf("Acked msg-%04d\n", messagesReceived)
+						ackMessage = true
+						messages.Messages = append(messages.Messages[:i], messages.Messages[i+1:]...)
+					}
+				}
 			}
 		}
-	}
 
-	return nil
-}
+		err = SaveMessageToFile(msg, outputDir, messagesReceived, isContentJSON, ackMessage)
+		if err != nil {
+			return fmt.Errorf("Save message: %s", err)
+		}
 
-func SaveMessageToFile(body []byte, outputDir string, counter uint) error {
-	filePath := GenerateFilePath(outputDir, counter)
-	err := ioutil.WriteFile(filePath, body, 0644)
-	if err != nil {
-		return err
 	}
 
-	fmt.Println(filePath)
-
 	return nil
 }
 
@@ -125,6 +187,8 @@ func GetProperties(msg amqp.Delivery) map[string]interface{} {
 		"reply_to":         msg.ReplyTo,
 		"type":             msg.Type,
 		"user_id":          msg.UserId,
+		"exchange":         msg.Exchange,
+		"routing_key":      msg.RoutingKey,
 	}
 
 	if !msg.Timestamp.IsZero() {
@@ -140,17 +204,25 @@ func GetProperties(msg amqp.Delivery) map[string]interface{} {
 	return props
 }
 
-func SavePropsAndHeadersToFile(msg amqp.Delivery, outputDir string, counter uint) error {
+func SaveMessageToFile(msg amqp.Delivery, outputDir string, counter uint, isContentJSON bool, wasAcked bool) error {
 	extras := make(map[string]interface{})
 	extras["properties"] = GetProperties(msg)
 	extras["headers"] = msg.Headers
+	extras["acked"] = wasAcked
+	if isContentJSON {
+		var content interface{}
+		json.Unmarshal(msg.Body, &content)
+		extras["content"] = content
+	} else {
+		extras["content"] = string(msg.Body[:])
+	}
 
 	data, err := json.MarshalIndent(extras, "", "  ")
 	if err != nil {
 		return err
 	}
 
-	filePath := GenerateFilePath(outputDir, counter) + "-headers+properties.json"
+	filePath := GenerateFilePath(outputDir, counter)
 	err = ioutil.WriteFile(filePath, data, 0644)
 	if err != nil {
 		return err
@@ -162,11 +234,20 @@ func SavePropsAndHeadersToFile(msg amqp.Delivery, outputDir string, counter uint
 }
 
 func GenerateFilePath(outputDir string, counter uint) string {
-	return path.Join(outputDir, fmt.Sprintf("msg-%04d", counter))
+	return path.Join(outputDir, fmt.Sprintf("msg-%04d.json", counter))
 }
 
 func VerboseLog(msg string) {
 	if *verbose {
 		fmt.Println("*", msg)
 	}
 }
+
+func NumDecPlaces(v float64) int {
+	s := strconv.FormatFloat(v, 'f', -1, 64)
+	i := strings.IndexByte(s, '.')
+	if i > -1 {
+		return len(s) - i - 1
+	}
+	return 0
+}