This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 063773f16 File: open zip archives in preStart rather than the stage
constructor (#1929)
063773f16 is described below
commit 063773f16d17a4e105ad50db9e7b12f75e69a05b
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 8 08:52:05 2026 +0100
File: open zip archives in preStart rather than the stage constructor
(#1929)
* File: open zip archives in preStart rather than the stage constructor
ZipEntrySource and ZipSource opened the file in the GraphStageLogic body,
so a
logic that was constructed but never started never reached postStop and
never
released the handle. A throwing ZipInputStream constructor also orphaned the
FileInputStream it had been handed.
Acquire in preStart instead, via a helper that closes the underlying file if
the zip stream cannot be opened, and null-guard postStop. Failing to open
the
archive now fails the stream rather than throwing out of runWith.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
* File: open the zip archive with Files rather than FileInputStream
Files.newInputStream reports a missing archive as NoSuchFileException
instead
of FileNotFoundException; both are IOExceptions and both now arrive as a
stream
failure rather than being thrown out of runWith.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../file/impl/archive/ZipReaderSource.scala | 39 +++++++++++++++++++---
.../src/test/scala/docs/scaladsl/ArchiveSpec.scala | 12 ++++++-
2 files changed, 45 insertions(+), 6 deletions(-)
diff --git
a/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
b/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
index ece284b43..b805abfed 100644
---
a/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
+++
b/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
@@ -22,9 +22,28 @@ import pekko.stream.scaladsl.Source
import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
import pekko.util.ByteString
-import java.io.{ File, FileInputStream }
+import java.io.File
import java.nio.charset.{ Charset, StandardCharsets }
+import java.nio.file.Files
import java.util.zip.{ ZipEntry, ZipInputStream }
+import scala.util.control.NonFatal
+
+@InternalApi private[archive] object ZipReaderSource {
+
+ /**
+ * Opens `f` as a zip stream, closing the underlying file if the zip stream
itself cannot be opened.
+ */
+ def openZip(f: File, fileCharset: Charset): ZipInputStream = {
+ val in = Files.newInputStream(f.toPath)
+ try new ZipInputStream(in, fileCharset)
+ catch {
+ case NonFatal(e) =>
+ try in.close()
+ catch { case NonFatal(suppressed) => e.addSuppressed(suppressed) }
+ throw e
+ }
+ }
+}
@InternalApi class ZipEntrySource(n: ZipArchiveMetadata, f: File, chunkSize:
Int, fileCharset: Charset)
extends GraphStage[SourceShape[ByteString]] {
@@ -34,10 +53,15 @@ import java.util.zip.{ ZipEntry, ZipInputStream }
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) {
- val zis = new ZipInputStream(new FileInputStream(f), fileCharset)
+ private var zis: ZipInputStream = _
var entry: ZipEntry = null
val data = new Array[Byte](chunkSize)
+ override def preStart(): Unit = {
+ super.preStart()
+ zis = ZipReaderSource.openZip(f, fileCharset)
+ }
+
def seek() = {
while ({
entry = zis.getNextEntry()
@@ -69,7 +93,7 @@ import java.util.zip.{ ZipEntry, ZipInputStream }
override def postStop(): Unit = {
super.postStop()
- zis.close()
+ if (zis ne null) zis.close()
}
}
}
@@ -82,7 +106,12 @@ import java.util.zip.{ ZipEntry, ZipInputStream }
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) {
- val zis = new ZipInputStream(new FileInputStream(f), fileCharset)
+ private var zis: ZipInputStream = _
+
+ override def preStart(): Unit = {
+ super.preStart()
+ zis = ZipReaderSource.openZip(f, fileCharset)
+ }
setHandler(
out,
@@ -102,7 +131,7 @@ import java.util.zip.{ ZipEntry, ZipInputStream }
override def postStop(): Unit = {
super.postStop()
- zis.close()
+ if (zis ne null) zis.close()
}
}
}
diff --git a/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
b/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
index 859d09d9d..0383c76c8 100644
--- a/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
@@ -14,7 +14,7 @@
package docs.scaladsl
import java.io._
-import java.nio.file.{ Files, Path, Paths }
+import java.nio.file.{ Files, NoSuchFileException, Path, Paths }
import java.util.zip.Deflater
import org.apache.pekko
import pekko.actor.ActorSystem
@@ -179,6 +179,16 @@ class ArchiveSpec
Files.delete(p))
}
}
+
+ "fail the stream when the archive cannot be opened" in {
+ val missing = new
File(Files.createTempDirectory("pekko-connectors-zip-").toFile,
"no-such-file.zip")
+
+ Archive
+ .zipReader(missing)
+ .runWith(Sink.ignore)
+ .failed
+ .futureValue shouldBe a[NoSuchFileException]
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
