Delete Container support akka#3253

1. Request builder for delete container
2. Implementation of delete container and Scala and Java API
3. Tests for delete container
4. Documentation

azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/impl/AzureStorageStream.scala

@@ -30,6 +30,7 @@ import akka.stream.alpakka.azure.storage.requests.{
   ClearFileRange,
   CreateContainer,
   CreateFile,
+  DeleteContainer,
   GetProperties,
   RequestBuilder,
   UpdateFileRange
@@ -132,6 +133,13 @@ object AzureStorageStream {
                   objectPath = objectPath,
                   requestBuilder = requestBuilder)
 
+  private[storage] def deleteContainer(objectPath: String,
+                                       requestBuilder: DeleteContainer): Source[Option[ObjectMetadata], NotUsed] =
+    handleRequest(successCode = Accepted,
+                  storageType = BlobType,
+                  objectPath = objectPath,
+                  requestBuilder = requestBuilder)
+
   /**
    * Common function to handle all requests where we don't expect response body.
    *

azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/javadsl/BlobService.scala

@@ -12,6 +12,7 @@ import akka.http.scaladsl.model.HttpEntity
 import akka.stream.alpakka.azure.storage.impl.AzureStorageStream
 import akka.stream.alpakka.azure.storage.requests.{
   CreateContainer,
+  DeleteContainer,
   DeleteFile,
   GetBlob,
   GetProperties,
@@ -133,4 +134,15 @@ object BlobService {
    */
   def createContainer(objectPath: String, requestBuilder: CreateContainer): Source[Optional[ObjectMetadata], NotUsed] =
     AzureStorageStream.createContainer(objectPath, requestBuilder).map(opt => Optional.ofNullable(opt.orNull)).asJava
+
+  /**
+   * Delete container.
+   *
+   * @param objectPath name of the container
+   * @param requestBuilder builder to build deleteContainer request
+   * @return A [[akka.stream.scaladsl.Source Source]] containing an [[scala.Option]] of
+   *         [[akka.stream.alpakka.azure.storage.ObjectMetadata]], will be [[scala.None]] in case the object does not exist
+   */
+  def deleteContainer(objectPath: String, requestBuilder: DeleteContainer): Source[Optional[ObjectMetadata], NotUsed] =
+    AzureStorageStream.deleteContainer(objectPath, requestBuilder).map(opt => Optional.ofNullable(opt.orNull)).asJava
 }

azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/requests/DeleteContainer.scala

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka
+package azure
+package storage
+package requests
+
+import akka.http.scaladsl.model.{HttpHeader, HttpMethod, HttpMethods}
+import akka.stream.alpakka.azure.storage.headers.ServerSideEncryption
+import akka.stream.alpakka.azure.storage.impl.StorageHeaders
+
+final class DeleteContainer(val leaseId: Option[String] = None,
+                            override val sse: Option[ServerSideEncryption] = None,
+                            override val additionalHeaders: Seq[HttpHeader] = Seq.empty)
+    extends RequestBuilder {
+
+  override protected val method: HttpMethod = HttpMethods.DELETE
+
+  override protected val queryParams: Map[String, String] = super.queryParams ++ Map("restype" -> "container")
+
+  def withLeaseId(leaseId: String): DeleteContainer = copy(leaseId = Option(leaseId))
+
+  override def withServerSideEncryption(sse: ServerSideEncryption): DeleteContainer = copy(sse = Option(sse))
+
+  override def addHeader(httpHeader: HttpHeader): DeleteContainer =
+    copy(additionalHeaders = additionalHeaders :+ httpHeader)
+
+  override protected def getHeaders: Seq[HttpHeader] =
+    StorageHeaders()
+      .witServerSideEncryption(sse)
+      .withLeaseIdHeader(leaseId)
+      .withAdditionalHeaders(additionalHeaders)
+      .headers
+
+  private def copy(leaseId: Option[String] = leaseId,
+                   sse: Option[ServerSideEncryption] = sse,
+                   additionalHeaders: Seq[HttpHeader] = additionalHeaders) =
+    new DeleteContainer(leaseId = leaseId, sse = sse, additionalHeaders = additionalHeaders)
+}
+
+object DeleteContainer {
+
+  /*
+   * Scala API
+   */
+  def apply(): DeleteContainer = new DeleteContainer()
+
+  /*
+   * Java API
+   */
+  def create(): DeleteContainer = DeleteContainer()
+}

azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/scaladsl/BlobService.scala

@@ -13,6 +13,7 @@ import akka.stream.alpakka.azure.storage.impl.AzureStorageStream
 import akka.stream.alpakka.azure.storage.requests.{
   CreateContainer,
   DeleteBlob,
+  DeleteContainer,
   GetBlob,
   GetProperties,
   PutAppendBlock,
@@ -110,4 +111,15 @@ object BlobService {
    */
   def createContainer(objectPath: String, requestBuilder: CreateContainer): Source[Option[ObjectMetadata], NotUsed] =
     AzureStorageStream.createContainer(objectPath, requestBuilder)
+
+  /**
+   * Delete container.
+   *
+   * @param objectPath name of the container
+   * @param requestBuilder builder to build deleteContainer request
+   * @return A [[akka.stream.scaladsl.Source Source]] containing an [[scala.Option]] of
+   *         [[akka.stream.alpakka.azure.storage.ObjectMetadata]], will be [[scala.None]] in case the object does not exist
+   */
+  def deleteContainer(objectPath: String, requestBuilder: DeleteContainer): Source[Option[ObjectMetadata], NotUsed] =
+    AzureStorageStream.deleteContainer(objectPath, requestBuilder)
 }

azure-storage/src/test/java/docs/javadsl/StorageTest.java

@@ -14,6 +14,7 @@
 import akka.stream.alpakka.azure.storage.requests.ClearFileRange;
 import akka.stream.alpakka.azure.storage.requests.CreateContainer;
 import akka.stream.alpakka.azure.storage.requests.CreateFile;
+import akka.stream.alpakka.azure.storage.requests.DeleteContainer;
 import akka.stream.alpakka.azure.storage.requests.DeleteFile;
 import akka.stream.alpakka.azure.storage.requests.GetBlob;
 import akka.stream.alpakka.azure.storage.requests.GetFile;
@@ -82,6 +83,22 @@ public void createContainer() throws Exception {
         Assert.assertEquals(objectMetadata.getETag().get(), ETagRawValue());
     }
 
+    @Test
+    public void deleteContainer() throws Exception {
+        mockDeleteContainer();
+
+        //#delete-container
+        final Source<Optional<ObjectMetadata>, NotUsed> source = BlobService.deleteContainer(containerName(), DeleteContainer.create());
+
+        final CompletionStage<Optional<ObjectMetadata>> optionalCompletionStage = source.runWith(Sink.head(), system);
+        //#delete-container
+
+        final var optionalObjectMetadata = optionalCompletionStage.toCompletableFuture().get();
+        Assert.assertTrue(optionalObjectMetadata.isPresent());
+        final var objectMetadata = optionalObjectMetadata.get();
+        Assert.assertEquals(objectMetadata.getContentLength(), 0L);
+    }
+
 
     // TODO: There are couple of issues, firstly there are two `Content-Length` headers being added, one by `putBlob`
     // function and secondly by, most likely, by WireMock. Need to to figure out how to tell WireMock not to add `Content-Length`

azure-storage/src/test/scala/akka/stream/alpakka/azure/storage/scaladsl/AzuriteIntegrationSpec.scala

@@ -12,9 +12,6 @@ import akka.stream.Attributes
 import com.dimafeng.testcontainers.ForAllTestContainer
 import org.scalatest.Ignore
 
-import scala.concurrent.duration._
-import scala.concurrent.Await
-
 // TODO: investigate how Azurite works, it is not even working with pure Java API
 // `putBlob` operations fails with "Premature end of file." error with BadRequest.
 @Ignore
@@ -27,12 +24,5 @@ class AzuriteIntegrationSpec extends StorageIntegrationSpec with ForAllTestConta
   protected lazy val blobSettings: StorageSettings =
     StorageExt(system).settings("azurite").withEndPointUrl(container.getBlobHostAddress)
 
-  override protected def beforeAll(): Unit = {
-    super.beforeAll()
-
-    val eventualDone = createContainer(defaultContainerName)
-    Await.result(eventualDone, 10.seconds)
-  }
-
   override protected def getDefaultAttributes: Attributes = StorageAttributes.settings(blobSettings)
 }

azure-storage/src/test/scala/akka/stream/alpakka/azure/storage/scaladsl/StorageIntegrationSpec.scala

@@ -13,7 +13,14 @@ import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.ContentTypes
 import akka.http.scaladsl.model.headers.ByteRange
 import akka.stream.Attributes
-import akka.stream.alpakka.azure.storage.requests.{CreateContainer, DeleteBlob, GetBlob, GetProperties, PutBlockBlob}
+import akka.stream.alpakka.azure.storage.requests.{
+  CreateContainer,
+  DeleteBlob,
+  DeleteContainer,
+  GetBlob,
+  GetProperties,
+  PutBlockBlob
+}
 import akka.stream.alpakka.testkit.scaladsl.LogCapturing
 import akka.stream.scaladsl.{Flow, Framing, Keep, Sink, Source}
 import akka.testkit.TestKit
@@ -56,6 +63,19 @@ trait StorageIntegrationSpec
   protected def getDefaultAttributes: Attributes = StorageAttributes.settings(StorageSettings())
 
   "BlobService" should {
+    "create container" in {
+      val maybeObjectMetadata =
+        BlobService
+          .createContainer(objectPath = defaultContainerName, requestBuilder = CreateContainer())
+          .withAttributes(getDefaultAttributes)
+          .runWith(Sink.head)
+          .futureValue
+
+      maybeObjectMetadata shouldBe defined
+      val objectMetadata = maybeObjectMetadata.get
+      objectMetadata.contentLength shouldBe 0L
+    }
+
     "put blob" in {
       val maybeObjectMetadata =
         BlobService
@@ -142,13 +162,19 @@ trait StorageIntegrationSpec
 
       maybeObjectMetadata shouldBe empty
     }
-  }
 
-  protected def createContainer(containerName: String): Future[Done] = {
-    BlobService
-      .createContainer(containerName, CreateContainer())
-      .withAttributes(getDefaultAttributes)
-      .runWith(Sink.ignore)
+    "delete container" in {
+      val maybeObjectMetadata =
+        BlobService
+          .deleteContainer(objectPath = defaultContainerName, requestBuilder = DeleteContainer())
+          .withAttributes(getDefaultAttributes)
+          .runWith(Sink.head)
+          .futureValue
+
+      maybeObjectMetadata shouldBe defined
+      val objectMetadata = maybeObjectMetadata.get
+      objectMetadata.contentLength shouldBe 0L
+    }
   }
 
   protected def calculateDigest(text: String): String = {

azure-storage/src/test/scala/akka/stream/alpakka/azure/storage/scaladsl/StorageWireMockBase.scala

@@ -62,6 +62,16 @@ abstract class StorageWireMockBase(_system: ActorSystem, val _wireMockServer: Wi
         )
     )
 
+  protected def mockDeleteContainer(): StubMapping =
+    mock.register(
+      delete(urlEqualTo(s"/$AccountName/$containerName?restype=container"))
+        .willReturn(
+          aResponse()
+            .withStatus(202)
+            .withHeader(`Content-Length`.name, "0")
+        )
+    )
+
   protected def mockPutBlockBlob(): StubMapping =
     mock.register(
       put(urlEqualTo(s"/$AccountName/$containerName/$blobName"))