support for create container operation akka#3253

sfali · Aug 25, 2024 · 44c2f84 · 44c2f84
1 parent 07e64a5
commit 44c2f84
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 1 deletion.
diff --git a/azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/impl/AzureStorageStream.scala b/azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/impl/AzureStorageStream.scala
@@ -231,6 +231,24 @@ object AzureStorageStream {
       .mapMaterializedValue(_ => NotUsed)
   }
 
+  private[storage] def createContainer(objectPath: String): Source[Option[ObjectMetadata], NotUsed] =
+    Source
+      .fromMaterializer { (mat, attr) =>
+        implicit val system: ActorSystem = mat.system
+        val settings = resolveSettings(attr, system)
+        val request =
+          createRequest(
+            method = HttpMethods.PUT,
+            accountName = settings.azureNameKeyCredential.accountName,
+            storageType = BlobType,
+            objectPath = objectPath,
+            queryString = createQueryString(settings, Some("restype=container")),
+            headers = populateCommonHeaders(HttpEntity.Empty)
+          )
+        handlePutRequest(request, settings)
+      }
+      .mapMaterializedValue(_ => NotUsed)
+
   private def handlePutRequest(request: HttpRequest, settings: StorageSettings)(implicit system: ActorSystem) = {
     import system.dispatcher
     signAndRequest(request, settings).flatMapConcat {
@@ -330,7 +348,7 @@ object AzureStorageStream {
                                     overrideContentLength: Option[Long] = None,
                                     range: Option[ByteRange] = None,
                                     blobType: Option[String] = None,
-                                    leaseId: Option[String],
+                                    leaseId: Option[String] = None,
                                     writeType: Option[String] = None) = {
     // Azure required to have these two headers (Content-Length & Content-Type) in the request
     // in some cases Content-Length header must be set as 0

diff --git a/azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/javadsl/BlobService.scala b/azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/javadsl/BlobService.scala
@@ -100,6 +100,7 @@ object BlobService {
       .asJava
 
   /**
+   * Put blob.
    *
    * @param objectPath path of the object, should start with "/" and separated by `/`, e.g. `/container/blob`
    * @param contentType content type of the blob
@@ -124,4 +125,14 @@ object BlobService {
                Option(leaseId.orElse(null)))
       .map(opt => Optional.ofNullable(opt.orNull))
       .asJava
+
+  /**
+   * Create container.
+   *
+   * @param objectPath path of the object, should start with "/" and separated by `/`, e.g. `/container/blob`
+   * @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 createContainer(objectPath: String): Source[Optional[ObjectMetadata], NotUsed] =
+    AzureStorageStream.createContainer(objectPath).map(opt => Optional.ofNullable(opt.orNull)).asJava
 }
diff --git a/azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/scaladsl/BlobService.scala b/azure-storage/src/main/scala/akka/stream/alpakka/azure/storage/scaladsl/BlobService.scala
@@ -66,6 +66,7 @@ object BlobService {
     AzureStorageStream.deleteObject(BlobType, objectPath, versionId, leaseId)
 
   /**
+   * Put blob.
    *
    * @param objectPath path of the object, should start with "/" and separated by `/`, e.g. `/container/blob`
    * @param contentType content type of the blob
@@ -82,4 +83,14 @@ object BlobService {
               blobType: String = "BlockBlob",
               leaseId: Option[String] = None): Source[Option[ObjectMetadata], NotUsed] =
     AzureStorageStream.putBlob(blobType, objectPath, contentType, contentLength, payload, leaseId)
+
+  /**
+   * Create container.
+   *
+   * @param objectPath path of the object, should start with "/" and separated by `/`, e.g. `/container/blob`
+   * @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 createContainer(objectPath: String): Source[Option[ObjectMetadata], NotUsed] =
+    AzureStorageStream.createContainer(objectPath)
 }