Merge pull request dubek#7 from msteggink/master

Add support for `-ack=true` (messages are acked and removed from the queue).

Author: dubek

README.md

@@ -57,10 +57,18 @@ The JSON files have the following structure:
       }
     }
 
+
+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 -ack=true
+
 Running `rabbitmq-dump-queue -help` will list the available command-line
 options.
 
 
+
 ## Message requeuing implementation details
 
 In order to fetch messages from the queue and later return them in the original

main.go

@@ -17,6 +17,7 @@ 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")
@@ -27,7 +28,7 @@ func main() {
 	flag.Parse()
 	if flag.NArg() > 0 {
 		fmt.Fprintf(os.Stderr, "Error: Unused command line arguments detected.\n")
-		flag.Usage();
+		flag.Usage()
 		os.Exit(2)
 	}
 	err := DumpMessagesFromQueue(*uri, *queue, *maxMessages, *outputDir)
@@ -72,7 +73,7 @@ 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++ {
 		msg, ok, err := channel.Get(queueName,
-			false, // autoAck
+			*ack, // autoAck
 		)
 		if err != nil {
 			return fmt.Errorf("Queue get: %s", err)

main_test.go

@@ -122,6 +122,37 @@ func verifyFileContent(t *testing.T, filename, expectedContent string) {
 	}
 }
 
+func TestAcknowledge(t *testing.T) {
+	os.MkdirAll("tmp-test", 0775)
+	defer os.RemoveAll("tmp-test")
+	populateTestQueue(t, 10)
+	defer deleteTestQueue(t)
+	output, err := exec.Command("./rabbitmq-dump-queue", "-uri="+TEST_AMQP_URI, "-queue="+TEST_QUEUE_NAME, "-max-messages=3", "-output-dir=tmp-test", "-ack=true").CombinedOutput()
+	if err != nil {
+		t.Fatalf("run: %s: %s", err, string(output))
+	}
+	expectedOutput := "tmp-test/msg-0000\n" +
+		"tmp-test/msg-0001\n" +
+		"tmp-test/msg-0002\n"
+	if string(output) != expectedOutput {
+		t.Errorf("Wrong output: expected '%s' but got '%s'", expectedOutput, output)
+	}
+	output2, err2 := exec.Command("./rabbitmq-dump-queue", "-uri="+TEST_AMQP_URI, "-queue="+TEST_QUEUE_NAME, "-max-messages=10", "-output-dir=tmp-test", "-ack=true").CombinedOutput()
+	if err2 != nil {
+		t.Fatalf("run: %s: %s", err, string(output))
+	}
+	expectedOutput2 := "tmp-test/msg-0000\n" +
+		"tmp-test/msg-0001\n" +
+		"tmp-test/msg-0002\n" +
+		"tmp-test/msg-0003\n" +
+		"tmp-test/msg-0004\n" +
+		"tmp-test/msg-0005\n" +
+		"tmp-test/msg-0006\n"
+	if string(output2) != expectedOutput2 {
+		t.Errorf("Wrong output: expected '%s' but got '%s'", expectedOutput2, output2)
+	}
+}
+
 func TestNormal(t *testing.T) {
 	os.MkdirAll("tmp-test", 0775)
 	defer os.RemoveAll("tmp-test")