-
Notifications
You must be signed in to change notification settings - Fork 645
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
File: TAR archive generation (#2241)
- Loading branch information
1 parent
6264cd2
commit 6020e28
Showing
10 changed files
with
478 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
48 changes: 48 additions & 0 deletions
48
file/src/main/scala/akka/stream/alpakka/file/impl/archive/EnsureByteStreamSize.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /* | ||
| * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.stream.alpakka.file.impl.archive | ||
|
|
||
| import akka.annotation.InternalApi | ||
| import akka.stream.{Attributes, FlowShape, Inlet, Outlet} | ||
| import akka.stream.stage.{GraphStage, GraphStageLogic, InHandler, OutHandler} | ||
| import akka.util.ByteString | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[file] class EnsureByteStreamSize(expectedSize: Long) | ||
| extends GraphStage[FlowShape[ByteString, ByteString]] { | ||
|
|
||
| val in = Inlet[ByteString]("EnsureByteStreamSize.in") | ||
| val out = Outlet[ByteString]("EnsureByteStreamSize.out") | ||
|
|
||
| override val shape = FlowShape.of(in, out) | ||
|
|
||
| override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) { | ||
| private var currentSize = 0L | ||
|
|
||
| setHandler( | ||
| in, | ||
| new InHandler { | ||
| override def onPush(): Unit = { | ||
| val elem = grab(in) | ||
| currentSize = currentSize + elem.size | ||
| push(out, elem) | ||
| } | ||
|
|
||
| override def onUpstreamFinish(): Unit = { | ||
| if (currentSize == expectedSize) super.onUpstreamFinish() | ||
| else failStage(new IllegalStateException(s"Expected ${expectedSize} bytes but got ${currentSize} bytes")) | ||
| } | ||
| } | ||
| ) | ||
| setHandler(out, new OutHandler { | ||
| override def onPull(): Unit = { | ||
| pull(in) | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| } |
79 changes: 79 additions & 0 deletions
79
file/src/main/scala/akka/stream/alpakka/file/impl/archive/TarArchiveEntry.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.stream.alpakka.file.impl.archive | ||
|
|
||
| import java.lang.Long.toOctalString | ||
|
|
||
| import akka.annotation.InternalApi | ||
| import akka.stream.alpakka.file.TarArchiveMetadata | ||
| import akka.util.ByteString | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[file] final class TarArchiveEntry(metadata: TarArchiveMetadata) { | ||
|
|
||
| def headerBytes: ByteString = { | ||
| val withoutChecksum = headerBytesWithoutChecksum | ||
| val checksumLong = withoutChecksum.foldLeft(0L)((sum, byte) => sum + byte) | ||
| val checksumBytes = ByteString(toOctalString(checksumLong).reverse.padTo(6, '0').take(6).reverse) ++ ByteString( | ||
| new Array[Byte](1) ++ ByteString(" ") | ||
| ) | ||
| val withChecksum = withoutChecksum.take(148) ++ checksumBytes ++ withoutChecksum.drop(148 + 8) | ||
| withChecksum.compact | ||
| } | ||
|
|
||
| def trailingBytes: ByteString = { | ||
| val paddingSize = if (metadata.size % 512 > 0) (512 - metadata.size % 512).toInt else 0 | ||
| padded(ByteString.empty, paddingSize) | ||
| } | ||
|
|
||
| private def headerBytesWithoutChecksum: ByteString = { | ||
| // [0, 100) | ||
| val fileNameBytes = padded(ByteString(metadata.filePathName), 100) | ||
| // [100, 108) | ||
| val fileModeBytes = padded(ByteString("0755"), 8) | ||
| // [108, 116) | ||
| val ownerIdBytes = padded(ByteString.empty, 8) | ||
| // [116, 124) | ||
| val groupIdBytes = padded(ByteString.empty, 8) | ||
| // [124, 136) | ||
| val fileSizeBytes = padded(ByteString("0" + toOctalString(metadata.size)), 12) | ||
| // [136, 148) | ||
| val lastModificationBytes = padded(ByteString(toOctalString(metadata.lastModification.getEpochSecond)), 12) | ||
| // [148, 156) | ||
| val checksumPlaceholderBytes = ByteString(" ") | ||
| // [156, 157) | ||
| val linkIndicatorBytes = padded(ByteString.empty, 1) | ||
| // [157, 257) | ||
| val linkFileNameBytes = padded(ByteString.empty, 100) | ||
| // [257, 263) | ||
| val ustarIndicatorBytes = ByteString("ustar") ++ ByteString(new Array[Byte](1)) | ||
| // [263, 265) | ||
| val ustarVersionBytes = ByteString(new Array[Byte](2)) | ||
| // [265, 297) | ||
| val ownerNameBytes = padded(ByteString.empty, 32) | ||
| // [297, 329) | ||
| val groupNameBytes = padded(ByteString.empty, 32) | ||
| // [329, 337) | ||
| val deviceMajorNumberBytes = padded(ByteString.empty, 8) | ||
| // [337, 345) | ||
| val deviceMinorNumberBytes = padded(ByteString.empty, 8) | ||
| // [345, 500) | ||
| val fileNamePrefixBytes = padded(metadata.filePathPrefix.map(ByteString.apply).getOrElse(ByteString.empty), 155) | ||
|
|
||
| padded( | ||
| fileNameBytes ++ fileModeBytes ++ ownerIdBytes ++ groupIdBytes ++ fileSizeBytes ++ lastModificationBytes ++ checksumPlaceholderBytes ++ linkIndicatorBytes ++ linkFileNameBytes ++ ustarIndicatorBytes ++ ustarVersionBytes ++ ownerNameBytes ++ groupNameBytes ++ deviceMajorNumberBytes ++ deviceMinorNumberBytes ++ fileNamePrefixBytes, | ||
| 512 | ||
| ) | ||
| } | ||
|
|
||
| private def padded(bytes: ByteString, targetSize: Int): ByteString = { | ||
| require(bytes.size <= targetSize) | ||
| if (bytes.size < targetSize) bytes ++ ByteString(new Array[Byte](targetSize - bytes.size)) | ||
| else bytes | ||
| } | ||
|
|
||
| } |
30 changes: 30 additions & 0 deletions
30
file/src/main/scala/akka/stream/alpakka/file/impl/archive/TarArchiveManager.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| /* | ||
| * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.stream.alpakka.file.impl.archive | ||
|
|
||
| import akka.NotUsed | ||
| import akka.annotation.InternalApi | ||
| import akka.stream.alpakka.file.TarArchiveMetadata | ||
| import akka.stream.scaladsl.{Flow, Source} | ||
| import akka.util.ByteString | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[file] object TarArchiveManager { | ||
|
|
||
| def tarFlow(): Flow[(TarArchiveMetadata, Source[ByteString, _]), ByteString, NotUsed] = { | ||
| Flow[(TarArchiveMetadata, Source[ByteString, Any])] | ||
| .flatMapConcat { | ||
| case (metadata, stream) => | ||
| val entry = new TarArchiveEntry(metadata) | ||
| Source | ||
| .single(entry.headerBytes) | ||
| .concat(stream.via(new EnsureByteStreamSize(metadata.size))) | ||
| .concat(Source.single(entry.trailingBytes)) | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.