Return PurgeStatus from queuePurge

It only contains the number of messages deleted from the queue, but
there may be more information in the future.

Fixes #151

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/Management.java

@@ -71,8 +71,9 @@ public interface Management extends AutoCloseable {
    * Purge (delete all messages) from a queue.
    *
    * @param queue queue to delete messages from
+   * @return the status of the purge operation
    */
-  void queuePurge(String queue);
+  PurgeStatus queuePurge(String queue);
 
   /**
    * Start exchange specification.
@@ -948,4 +949,14 @@ interface QueueInfo {
      */
     int consumerCount();
   }
+
+  interface PurgeStatus {
+
+    /**
+     * The number of messages purged from the queue.
+     *
+     * @return the number of messages purged
+     */
+    long messageCount();
+  }
 }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagement.java

@@ -183,11 +183,13 @@ public UnbindSpecification unbind() {
   }
 
   @Override
-  public void queuePurge(String queue) {
+  public PurgeStatus queuePurge(String queue) {
     Map<String, Object> responseBody = delete(queueLocation(queue) + "/messages", CODE_200);
-    if (!responseBody.containsKey("message_count")) {
+    if (!responseBody.containsKey("message_count")
+        && !(responseBody.get("message_count") instanceof Number)) {
       throw new AmqpException("Response body should contain message_count");
     }
+    return new DefaultPurgeStatus(((Number) responseBody.get("message_count")).longValue());
   }
 
   void setToken(String token) {
@@ -856,4 +858,18 @@ enum State {
     UNAVAILABLE,
     CLOSED
   }
+
+  private static final class DefaultPurgeStatus implements PurgeStatus {
+
+    private final long messageCount;
+
+    private DefaultPurgeStatus(long messageCount) {
+      this.messageCount = messageCount;
+    }
+
+    @Override
+    public long messageCount() {
+      return this.messageCount;
+    }
+  }
 }

src/test/java/com/rabbitmq/client/amqp/impl/AmqpTest.java

@@ -725,7 +725,8 @@ void queuePurgeShouldRemoveAllMessages(TestInfo info) {
         .forEach(ignored -> publisher.publish(publisher.message(), callback));
     assertThat(publishSync).completes();
     assertThat(management.queueInfo(q)).hasMessageCount(messageCount);
-    management.queuePurge(q);
+    Management.PurgeStatus purgeStatus = management.queuePurge(q);
+    org.assertj.core.api.Assertions.assertThat(purgeStatus.messageCount()).isEqualTo(messageCount);
     assertThat(management.queueInfo(q)).isEmpty();
   }