Implement queue purge

Fixes #136

Author: acogoluegnes

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

@@ -94,6 +94,8 @@ public interface Management extends AutoCloseable {
    */
   UnbindSpecification unbind();
 
+  void queuePurge(String queue);
+
   /** Close the management instance and release its resources. */
   @Override
   void close();

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

@@ -68,6 +68,7 @@ class AmqpManagement implements Management {
   private static final int CODE_200 = 200;
   private static final int CODE_201 = 201;
   private static final int CODE_204 = 204;
+  private static final int CODE_400 = 400;
   private static final int CODE_404 = 404;
   private static final int CODE_409 = 409;
 
@@ -173,6 +174,14 @@ public UnbindSpecification unbind() {
     return new AmqpBindingManagement.AmqpUnbindSpecification(this);
   }
 
+  @Override
+  public void queuePurge(String queue) {
+    Map<String, Object> responseBody = delete(queueLocation(queue) + "/messages", CODE_200);
+    if (!responseBody.containsKey("message_count")) {
+      throw new AmqpException("Response body should contain message_count");
+    }
+  }
+
   void setToken(String token) {
     if (!this.connection.setTokenSupported()) {
       throw new UnsupportedOperationException("Token renewal requires at least RabbitMQ 4.1.0");
@@ -477,18 +486,23 @@ private static void checkResponse(
     if (!requestId.equals(response.correlationId())) {
       throw new AmqpException("Unexpected correlation ID");
     }
-    int responseCode = request.mapResponse().code();
+    int responseCode = request.response().code();
+    String explanation =
+        request.response().body() instanceof String ? (String) request.response().body() : null;
     if (IntStream.of(expectedResponseCodes).noneMatch(c -> c == responseCode)) {
-      if (responseCode == CODE_404) {
-        throw new AmqpException.AmqpEntityDoesNotExistException("Entity does not exist");
+      if (responseCode == CODE_404
+          || (responseCode == CODE_400 && queueDoesNotExist(explanation))) {
+        explanation = explanation == null ? "Entity does not exist" : explanation;
+        throw new AmqpException.AmqpEntityDoesNotExistException(explanation);
       } else {
         String message =
             String.format(
-                "Unexpected response code: %d instead of %s",
+                "Unexpected response code: %d instead of %s%s",
                 responseCode,
                 IntStream.of(expectedResponseCodes)
                     .mapToObj(String::valueOf)
-                    .collect(Collectors.joining(", ")));
+                    .collect(Collectors.joining(", ")),
+                explanation != null ? " (message: '" + explanation : "')");
         try {
           LOGGER.info(
               "Management request failed: '{}'. Response body: '{}'",
@@ -502,6 +516,11 @@ private static void checkResponse(
     }
   }
 
+  private static boolean queueDoesNotExist(String explanation) {
+    return explanation != null
+        && (explanation.contains("no queue '") && explanation.contains("in vhost '"));
+  }
+
   void bind(Map<String, Object> body) {
     declare(body, "/bindings", POST, CODE_204);
   }
@@ -637,6 +656,10 @@ private <K, V> Response<Map<K, V>> mapResponse() {
       return (Response<Map<K, V>>) this.response.get();
     }
 
+    private Response<?> response() {
+      return this.response.get();
+    }
+
     @SuppressWarnings("unchecked")
     private <K, V> Map<K, V> responseBodyAsMap() throws ClientException {
       return (Map<K, V>) this.responseMessage.get().body();

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

@@ -707,6 +707,36 @@ void publishedMessageShouldBeRejectedWhenQueueLimitIsReached(TestInfo info) {
     }
   }
 
+  @Test
+  void queuePurgeShouldRemoveAllMessages(TestInfo info) {
+    Management management = connection.management();
+    String q = TestUtils.name(info);
+    management.queue(q).exclusive(true).declare();
+    int messageCount = 100;
+    Sync publishSync = sync(messageCount);
+    Publisher.Callback callback =
+        context -> {
+          if (context.status() == Publisher.Status.ACCEPTED) {
+            publishSync.down();
+          }
+        };
+    Publisher publisher = connection.publisherBuilder().queue(q).build();
+    IntStream.range(0, messageCount)
+        .forEach(ignored -> publisher.publish(publisher.message(), callback));
+    assertThat(publishSync).completes();
+    assertThat(management.queueInfo(q)).hasMessageCount(messageCount);
+    management.queuePurge(q);
+    assertThat(management.queueInfo(q)).isEmpty();
+  }
+
+  @Test
+  void queuePurgeOnNonExistingQueueShouldThrowException(TestInfo info) {
+    String q = TestUtils.name(info);
+    assertThatThrownBy(() -> connection.management().queuePurge(q))
+        .isInstanceOf(AmqpException.AmqpEntityDoesNotExistException.class)
+        .hasMessageContaining(q);
+  }
+
   private static String uuid() {
     return UUID.randomUUID().toString();
   }