From ff59c14c173d575b4e193e42d06a73b82fff1ccb Mon Sep 17 00:00:00 2001 From: Alexander Nozik Date: Tue, 5 Nov 2019 21:42:29 +0300 Subject: [PATCH 1/3] Minor fix to multipart envelope --- .../kotlin/hep/dataforge/io/EnvelopeParts.kt | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt index ca4efea4..0a8a207d 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt @@ -17,6 +17,9 @@ object EnvelopeParts { const val PARTS_DATA_TYPE = "envelope.parts" } +/** + * Append multiple serialized envelopes to the data block. Previous data is erased if it was present + */ fun EnvelopeBuilder.parts(formatFactory: EnvelopeFormatFactory, envelopes: Collection) { dataType = PARTS_DATA_TYPE meta { @@ -36,7 +39,10 @@ fun EnvelopeBuilder.parts(formatFactory: EnvelopeFormatFactory, envelopes: Colle fun EnvelopeBuilder.parts(formatFactory: EnvelopeFormatFactory, builder: suspend SequenceScope.() -> Unit) = parts(formatFactory, sequence(builder).toList()) -fun Envelope.parts(io: IOPlugin = Global.plugins.fetch(IOPlugin)): Sequence { +/** + * If given envelope supports multipart data, return a sequence of those parts (could be empty). Otherwise return null. + */ +fun Envelope.parts(io: IOPlugin = Global.plugins.fetch(IOPlugin)): Sequence? { return when (dataType) { PARTS_DATA_TYPE -> { val size = meta[SIZE_KEY].int ?: error("Unsized parts not supported yet") @@ -55,6 +61,6 @@ fun Envelope.parts(io: IOPlugin = Global.plugins.fetch(IOPlugin)): Sequence emptySequence() + else -> null } } From 3e9cb3915cdd1e81bad42483fdb48d5af699f441 Mon Sep 17 00:00:00 2001 From: Alexander Nozik Date: Sun, 10 Nov 2019 19:08:32 +0300 Subject: [PATCH 2/3] Meta and Evelope format factories now implement Meta and Envelope formats (default representation). --- .../kotlin/hep/dataforge/data/DataNode.kt | 20 ++-- .../io/yaml/FrontMatterEnvelopeFormat.kt | 17 ++- .../hep/dataforge/io/yaml/YamlMetaFormat.kt | 10 +- .../kotlin/hep/dataforge/io/Binary.kt | 21 ++-- .../kotlin/hep/dataforge/io/EnvelopeFormat.kt | 2 +- .../kotlin/hep/dataforge/io/JsonMetaFormat.kt | 12 ++- .../kotlin/hep/dataforge/io/MetaFormat.kt | 4 +- .../hep/dataforge/io/TaggedEnvelopeFormat.kt | 18 +++- .../hep/dataforge/io/TaglessEnvelopeFormat.kt | 23 +++- .../hep/dataforge/io/EnvelopeFormatTest.kt | 4 +- .../hep/dataforge/io/EnvelopePartsTest.kt | 6 +- .../kotlin/hep/dataforge/io/functionsJVM.kt | 29 +++++ .../kotlin/hep/dataforge/io/ioFormatsJVM.kt | 100 +++++++++++++----- .../kotlin/hep/dataforge/io/FileBinaryTest.kt | 4 +- .../dataforge/workspace/WorkspaceBuilder.kt | 2 +- .../hep/dataforge/workspace/fileData.kt | 24 +---- 16 files changed, 197 insertions(+), 99 deletions(-) create mode 100644 dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/functionsJVM.kt diff --git a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt index 12bb06ab..589e875b 100644 --- a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt +++ b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt @@ -64,21 +64,13 @@ val DataItem?.node: DataNode? get() = (this as? DataItem.Node val DataItem?.data: Data? get() = (this as? DataItem.Leaf)?.value /** - * Start computation for all goals in data node + * Start computation for all goals in data node and return a job for the whole node */ -fun DataNode<*>.startAll(scope: CoroutineScope): Unit = items.values.forEach { - when (it) { - is DataItem.Node<*> -> it.value.startAll(scope) - is DataItem.Leaf<*> -> it.value.start(scope) - } -} - -fun DataNode<*>.joinAll(scope: CoroutineScope): Job = scope.launch { - startAll(scope) - items.forEach { - when (val value = it.value) { - is DataItem.Node -> value.value.joinAll(this).join() - is DataItem.Leaf -> value.value.await(scope) +fun DataNode<*>.launchAll(scope: CoroutineScope): Job = scope.launch { + items.values.forEach { + when (it) { + is DataItem.Node<*> -> it.value.launchAll(scope) + is DataItem.Leaf<*> -> it.value.start(scope) } } } diff --git a/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/FrontMatterEnvelopeFormat.kt b/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/FrontMatterEnvelopeFormat.kt index db701625..023635e2 100644 --- a/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/FrontMatterEnvelopeFormat.kt +++ b/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/FrontMatterEnvelopeFormat.kt @@ -24,7 +24,7 @@ class FrontMatterEnvelopeFormat( val readMetaFormat = metaTypeRegex.matchEntire(line)?.groupValues?.first() - ?.let { io.metaFormat(it) } ?: YamlMetaFormat.default + ?.let { io.metaFormat(it) } ?: YamlMetaFormat val metaBlock = buildPacket { do { @@ -45,7 +45,7 @@ class FrontMatterEnvelopeFormat( val readMetaFormat = metaTypeRegex.matchEntire(line)?.groupValues?.first() - ?.let { io.metaFormat(it) } ?: YamlMetaFormat.default + ?.let { io.metaFormat(it) } ?: YamlMetaFormat val metaBlock = buildPacket { do { @@ -72,7 +72,7 @@ class FrontMatterEnvelopeFormat( private val metaTypeRegex = "---(\\w*)\\s*".toRegex() override fun invoke(meta: Meta, context: Context): EnvelopeFormat { - return FrontMatterEnvelopeFormat(context.io, meta) + return FrontMatterEnvelopeFormat(context.io, meta) } override fun peekFormat(io: IOPlugin, input: Input): EnvelopeFormat? { @@ -84,5 +84,16 @@ class FrontMatterEnvelopeFormat( } } + private val default by lazy { invoke() } + + override fun Input.readPartial(): PartialEnvelope = + default.run { readPartial() } + + override fun Output.writeEnvelope(envelope: Envelope, metaFormatFactory: MetaFormatFactory, formatMeta: Meta) = + default.run { writeEnvelope(envelope, metaFormatFactory, formatMeta) } + + override fun Input.readObject(): Envelope = + default.run { readObject() } + } } \ No newline at end of file diff --git a/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/YamlMetaFormat.kt b/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/YamlMetaFormat.kt index 24ea44ec..7130518d 100644 --- a/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/YamlMetaFormat.kt +++ b/dataforge-io/dataforge-io-yaml/src/main/kotlin/hep/dataforge/io/yaml/YamlMetaFormat.kt @@ -45,12 +45,18 @@ class YamlMetaFormat(val meta: Meta) : MetaFormat { } companion object : MetaFormatFactory { - val default = YamlMetaFormat() - override fun invoke(meta: Meta, context: Context): MetaFormat = YamlMetaFormat(meta) override val name: Name = super.name + "yaml" override val key: Short = 0x594d //YM + + private val default = YamlMetaFormat() + + override fun Output.writeMeta(meta: Meta, descriptor: NodeDescriptor?) = + default.run { writeMeta(meta, descriptor) } + + override fun Input.readMeta(descriptor: NodeDescriptor?): Meta = + default.run { readMeta(descriptor) } } } \ No newline at end of file diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Binary.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Binary.kt index ca05de4d..bd1f2249 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Binary.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Binary.kt @@ -8,7 +8,7 @@ import kotlin.math.min */ interface Binary { /** - * The size of binary in bytes + * The size of binary in bytes. [ULong.MAX_VALUE] if size is not defined and input should be read until its end is reached */ val size: ULong @@ -18,6 +18,10 @@ interface Binary { * Some implementation may forbid this to be called twice. In this case second call will throw an exception. */ fun read(block: Input.() -> R): R + + companion object { + val EMPTY = EmptyBinary + } } /** @@ -48,12 +52,11 @@ fun RandomAccessBinary.readPacket(from: UInt, size: UInt): ByteReadPacket = read @ExperimentalUnsignedTypes object EmptyBinary : RandomAccessBinary { - override val size: ULong = 0.toULong() + override val size: ULong = 0u override fun read(from: UInt, size: UInt, block: Input.() -> R): R { error("The binary is empty") } - } @ExperimentalUnsignedTypes @@ -79,9 +82,9 @@ fun Binary.readWith(format: IOFormat): T = format.run { } } -fun IOFormat.writeBinary(obj: T): Binary { - val packet = buildPacket { - writeObject(obj) - } - return ArrayBinary(packet.readBytes()) -} \ No newline at end of file +//fun IOFormat.writeBinary(obj: T): Binary { +// val packet = buildPacket { +// writeObject(obj) +// } +// return ArrayBinary(packet.readBytes()) +//} \ No newline at end of file diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeFormat.kt index c52b9e1d..4f747ea2 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeFormat.kt @@ -31,7 +31,7 @@ interface EnvelopeFormat : IOFormat { } @Type(ENVELOPE_FORMAT_TYPE) -interface EnvelopeFormatFactory : IOFormatFactory { +interface EnvelopeFormatFactory : IOFormatFactory, EnvelopeFormat { override val name: Name get() = "envelope".asName() override val type: KClass get() = Envelope::class diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/JsonMetaFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/JsonMetaFormat.kt index a95cdec4..5c10505d 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/JsonMetaFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/JsonMetaFormat.kt @@ -38,12 +38,18 @@ class JsonMetaFormat(private val json: Json = Json.indented) : MetaFormat { } companion object : MetaFormatFactory { - val default = JsonMetaFormat() - override fun invoke(meta: Meta, context: Context): MetaFormat = default override val name: Name = super.name + "json" override val key: Short = 0x4a53//"JS" + + private val default = JsonMetaFormat() + + override fun Output.writeMeta(meta: Meta, descriptor: NodeDescriptor?) = + default.run { writeMeta(meta,descriptor) } + + override fun Input.readMeta(descriptor: NodeDescriptor?): Meta = + default.run { readMeta(descriptor) } } } @@ -90,7 +96,7 @@ fun Meta.toJson(descriptor: NodeDescriptor? = null): JsonObject { fun JsonElement.toMeta(descriptor: NodeDescriptor? = null): Meta { return when (val item = toMetaItem(descriptor)) { is MetaItem.NodeItem<*> -> item.node - is MetaItem.ValueItem ->item.value.toMeta() + is MetaItem.ValueItem -> item.value.toMeta() } } diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/MetaFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/MetaFormat.kt index ca9a53a2..9d1af81a 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/MetaFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/MetaFormat.kt @@ -27,7 +27,7 @@ interface MetaFormat : IOFormat { } @Type(META_FORMAT_TYPE) -interface MetaFormatFactory : IOFormatFactory { +interface MetaFormatFactory : IOFormatFactory, MetaFormat { override val name: Name get() = "meta".asName() override val type: KClass get() = Meta::class @@ -47,7 +47,7 @@ fun Meta.toString(format: MetaFormat): String = buildPacket { fun Meta.toString(formatFactory: MetaFormatFactory): String = toString(formatFactory()) -fun Meta.toBytes(format: MetaFormat = JsonMetaFormat.default): ByteReadPacket = buildPacket { +fun Meta.toBytes(format: MetaFormat = JsonMetaFormat): ByteReadPacket = buildPacket { format.run { writeObject(this@toBytes) } } diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt index cce3eade..5f9164dc 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt @@ -39,7 +39,12 @@ class TaggedEnvelopeFormat( override fun Output.writeEnvelope(envelope: Envelope, metaFormatFactory: MetaFormatFactory, formatMeta: Meta) { val metaFormat = metaFormatFactory.invoke(formatMeta, io.context) val metaBytes = metaFormat.writeBytes(envelope.meta) - val tag = Tag(metaFormatFactory.key, metaBytes.size.toUInt() + 2u, envelope.data?.size ?: 0.toULong()) + val actualSize: ULong = if (envelope.data == null) { + 0u + } else { + envelope.data?.size ?: ULong.MAX_VALUE + } + val tag = Tag(metaFormatFactory.key, metaBytes.size.toUInt() + 2u, actualSize) writePacket(tag.toBytes()) writeFully(metaBytes) writeText("\r\n") @@ -134,7 +139,16 @@ class TaggedEnvelopeFormat( } } - val default by lazy { invoke() } + private val default by lazy { invoke() } + + override fun Input.readPartial(): PartialEnvelope = + default.run { readPartial() } + + override fun Output.writeEnvelope(envelope: Envelope, metaFormatFactory: MetaFormatFactory, formatMeta: Meta) = + default.run { writeEnvelope(envelope, metaFormatFactory, formatMeta) } + + override fun Input.readObject(): Envelope = + default.run { readObject() } } } \ No newline at end of file diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaglessEnvelopeFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaglessEnvelopeFormat.kt index 14d871db..d3953a0c 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaglessEnvelopeFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaglessEnvelopeFormat.kt @@ -27,7 +27,13 @@ class TaglessEnvelopeFormat( //printing all properties writeProperty(META_TYPE_PROPERTY, metaFormatFactory.type) //TODO add optional metaFormat properties - writeProperty(DATA_LENGTH_PROPERTY, envelope.data?.size ?: 0) + val actualSize: ULong = if (envelope.data == null) { + 0u + } else { + envelope.data?.size ?: ULong.MAX_VALUE + } + + writeProperty(DATA_LENGTH_PROPERTY, actualSize) //Printing meta if (!envelope.meta.isEmpty()) { @@ -66,7 +72,7 @@ class TaglessEnvelopeFormat( var meta: Meta = EmptyMeta if (line.startsWith(metaStart)) { - val metaFormat = properties[META_TYPE_PROPERTY]?.let { io.metaFormat(it) } ?: JsonMetaFormat.default + val metaFormat = properties[META_TYPE_PROPERTY]?.let { io.metaFormat(it) } ?: JsonMetaFormat val metaSize = properties.get(META_LENGTH_PROPERTY)?.toInt() meta = if (metaSize != null) { val metaPacket = buildPacket { @@ -121,7 +127,7 @@ class TaglessEnvelopeFormat( var meta: Meta = EmptyMeta if (line.startsWith(metaStart)) { - val metaFormat = properties[META_TYPE_PROPERTY]?.let { io.metaFormat(it) } ?: JsonMetaFormat.default + val metaFormat = properties[META_TYPE_PROPERTY]?.let { io.metaFormat(it) } ?: JsonMetaFormat val metaSize = properties.get(META_LENGTH_PROPERTY)?.toInt() meta = if (metaSize != null) { @@ -170,7 +176,16 @@ class TaglessEnvelopeFormat( return TaglessEnvelopeFormat(context.io, meta) } - val default by lazy { invoke() } + private val default by lazy { invoke() } + + override fun Input.readPartial(): PartialEnvelope = + default.run { readPartial() } + + override fun Output.writeEnvelope(envelope: Envelope, metaFormatFactory: MetaFormatFactory, formatMeta: Meta) = + default.run { writeEnvelope(envelope, metaFormatFactory, formatMeta) } + + override fun Input.readObject(): Envelope = + default.run { readObject() } override fun peekFormat(io: IOPlugin, input: Input): EnvelopeFormat? { return try { diff --git a/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopeFormatTest.kt b/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopeFormatTest.kt index 29e60f2f..37ee827d 100644 --- a/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopeFormatTest.kt +++ b/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopeFormatTest.kt @@ -18,7 +18,7 @@ class EnvelopeFormatTest { @ExperimentalStdlibApi @Test fun testTaggedFormat(){ - TaggedEnvelopeFormat.default.run { + TaggedEnvelopeFormat.run { val bytes = writeBytes(envelope) println(bytes.decodeToString()) val res = readBytes(bytes) @@ -32,7 +32,7 @@ class EnvelopeFormatTest { @Test fun testTaglessFormat(){ - TaglessEnvelopeFormat.default.run { + TaglessEnvelopeFormat.run { val bytes = writeBytes(envelope) println(bytes.decodeToString()) val res = readBytes(bytes) diff --git a/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt b/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt index 5122680a..c33b3179 100644 --- a/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt +++ b/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt @@ -23,9 +23,9 @@ class EnvelopePartsTest { @Test fun testParts() { - val bytes = TaggedEnvelopeFormat.default.writeBytes(partsEnvelope) - val reconstructed = TaggedEnvelopeFormat.default.readBytes(bytes) - val parts = reconstructed.parts().toList() + val bytes = TaggedEnvelopeFormat.writeBytes(partsEnvelope) + val reconstructed = TaggedEnvelopeFormat.readBytes(bytes) + val parts = reconstructed.parts()?.toList() ?: emptyList() assertEquals(2, parts[2].meta["value"].int) } diff --git a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/functionsJVM.kt b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/functionsJVM.kt new file mode 100644 index 00000000..fae986d7 --- /dev/null +++ b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/functionsJVM.kt @@ -0,0 +1,29 @@ +package hep.dataforge.io + +import hep.dataforge.io.functions.FunctionServer +import hep.dataforge.io.functions.function +import hep.dataforge.meta.Meta +import hep.dataforge.meta.buildMeta +import hep.dataforge.names.Name +import kotlin.reflect.KClass +import kotlin.reflect.full.isSuperclassOf + + +fun IOPlugin.resolveIOFormatName(type: KClass<*>): Name { + return ioFormats.entries.find { it.value.type.isSuperclassOf(type) }?.key + ?: error("Can't resolve IOFormat for type $type") +} + +inline fun IOPlugin.generateFunctionMeta(functionName: String): Meta = buildMeta { + FunctionServer.FUNCTION_NAME_KEY put functionName + FunctionServer.INPUT_FORMAT_KEY put resolveIOFormatName(T::class).toString() + FunctionServer.OUTPUT_FORMAT_KEY put resolveIOFormatName(R::class).toString() +} + +inline fun FunctionServer.function( + functionName: String +): (suspend (T) -> R) { + val plugin = context.plugins.get() ?: error("IO plugin not loaded") + val meta = plugin.generateFunctionMeta(functionName) + return function(meta) +} \ No newline at end of file diff --git a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt index c926d07a..d0ce18b0 100644 --- a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt +++ b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt @@ -1,51 +1,93 @@ package hep.dataforge.io import hep.dataforge.descriptors.NodeDescriptor -import hep.dataforge.io.functions.FunctionServer -import hep.dataforge.io.functions.FunctionServer.Companion.FUNCTION_NAME_KEY -import hep.dataforge.io.functions.FunctionServer.Companion.INPUT_FORMAT_KEY -import hep.dataforge.io.functions.FunctionServer.Companion.OUTPUT_FORMAT_KEY -import hep.dataforge.io.functions.function +import hep.dataforge.meta.DFExperimental +import hep.dataforge.meta.EmptyMeta import hep.dataforge.meta.Meta -import hep.dataforge.meta.buildMeta -import hep.dataforge.names.Name +import kotlinx.io.nio.asInput import kotlinx.io.nio.asOutput import java.nio.file.Files import java.nio.file.Path import java.nio.file.StandardOpenOption -import kotlin.reflect.KClass import kotlin.reflect.full.isSuperclassOf +import kotlin.streams.asSequence inline fun IOPlugin.resolveIOFormat(): IOFormat? { return ioFormats.values.find { it.type.isSuperclassOf(T::class) } as IOFormat? } -fun IOPlugin.resolveIOFormatName(type: KClass<*>): Name { - return ioFormats.entries.find { it.value.type.isSuperclassOf(type) }?.key - ?: error("Can't resolve IOFormat for type $type") +/** + * Read file containing meta using given [formatOverride] or file extension to infer meta type. + */ +fun IOPlugin.readMetaFile(path: Path, formatOverride: MetaFormat? = null, descriptor: NodeDescriptor? = null): Meta { + if (!Files.exists(path)) error("Meta file $path does not exist") + val extension = path.fileName.toString().substringAfterLast('.') + + val metaFormat = formatOverride ?: metaFormat(extension) ?: error("Can't resolve meta format $extension") + return metaFormat.run { + Files.newByteChannel(path, StandardOpenOption.READ).asInput().use { it.readMeta(descriptor) } + } } -inline fun IOPlugin.generateFunctionMeta(functionName: String): Meta = buildMeta { - FUNCTION_NAME_KEY put functionName - INPUT_FORMAT_KEY put resolveIOFormatName(T::class).toString() - OUTPUT_FORMAT_KEY put resolveIOFormatName(R::class).toString() -} - -inline fun FunctionServer.function( - functionName: String -): (suspend (T) -> R) { - val plugin = context.plugins.get() ?: error("IO plugin not loaded") - val meta = plugin.generateFunctionMeta(functionName) - return function(meta) +fun IOPlugin.writeMetaFile( + path: Path, + metaFormat: MetaFormat = JsonMetaFormat, + descriptor: NodeDescriptor? = null +) { + metaFormat.run { + Files.newByteChannel(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW).asOutput().use { + it.writeMeta(meta, descriptor) + } + } } /** - * Write meta to file in a given [format] + * Read and envelope from file if the file exists, return null if file does not exist. */ -fun Meta.write(path: Path, format: MetaFormat, descriptor: NodeDescriptor? = null) { - format.run { - Files.newByteChannel(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW) - .asOutput() - .writeMeta(this@write, descriptor) +@DFExperimental +fun IOPlugin.readEnvelopeFromFile(path: Path, readNonEnvelopes: Boolean = false): Envelope? { + if (!Files.exists(path)) return null + + //read two-files directory + if (Files.isDirectory(path)) { + val metaFile = Files.list(path).asSequence() + .singleOrNull { it.fileName.toString().startsWith("meta") } + + val meta = if (metaFile == null) { + EmptyMeta + } else { + readMetaFile(metaFile) + } + + val dataFile = path.resolve("data") + + val data: Binary? = if (Files.exists(dataFile)) { + dataFile.asBinary() + } else { + null + } + + return SimpleEnvelope(meta, data) + } + + val binary = path.asBinary() + + val formats = envelopeFormatFactories.mapNotNull { factory -> + binary.read { + factory.peekFormat(this@readEnvelopeFromFile, this@read) + } + } + return when (formats.size) { + 0 -> if (readNonEnvelopes) { + SimpleEnvelope(Meta.empty, binary) + } else { + null + } + 1 -> formats.first().run { + binary.read { + readObject() + } + } + else -> error("Envelope format file recognition clash") } } \ No newline at end of file diff --git a/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt b/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt index 94403dcd..545601cd 100644 --- a/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt +++ b/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt @@ -39,7 +39,7 @@ class FileBinaryTest { } val binary = envelopeFromFile.data!! println(binary.toBytes().size) - assertEquals(binary.size.toInt(), binary.toBytes().size) + assertEquals(binary.size?.toInt(), binary.toBytes().size) } @@ -50,7 +50,7 @@ class FileBinaryTest { Global.io.writeEnvelopeFile(tmpPath, envelope) val binary = Global.io.readEnvelopeFile(tmpPath).data!! - assertEquals(binary.size.toInt(), binary.toBytes().size) + assertEquals(binary.size?.toInt(), binary.toBytes().size) } } \ No newline at end of file diff --git a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt index 2f717f78..b8f3ffa0 100644 --- a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt +++ b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt @@ -12,7 +12,7 @@ import hep.dataforge.names.toName import kotlin.jvm.JvmName import kotlin.reflect.KClass -@TaskBuildScope +@DFBuilder interface WorkspaceBuilder { val parentContext: Context var context: Context diff --git a/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt b/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt index a483c78b..4fbfa72a 100644 --- a/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt +++ b/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt @@ -4,36 +4,16 @@ import hep.dataforge.data.Data import hep.dataforge.data.DataNode import hep.dataforge.data.DataTreeBuilder import hep.dataforge.data.datum -import hep.dataforge.descriptors.NodeDescriptor import hep.dataforge.io.* import hep.dataforge.meta.EmptyMeta -import hep.dataforge.meta.Meta import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext import kotlinx.io.nio.asInput -import kotlinx.io.nio.asOutput import java.nio.file.Files import java.nio.file.Path import java.nio.file.StandardOpenOption import kotlin.reflect.KClass -/** - * Read meta from file in a given [MetaFormat] - */ -fun MetaFormat.readMetaFile(path: Path, descriptor: NodeDescriptor? = null): Meta { - return Files.newByteChannel(path, StandardOpenOption.READ) - .asInput() - .readMeta(descriptor) -} - -/** - * Write meta to file using given [MetaFormat] - */ -fun MetaFormat.writeMetaFile(path: Path, meta: Meta, descriptor: NodeDescriptor? = null) { - return Files.newByteChannel(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW) - .asOutput() - .writeMeta(meta, descriptor) -} /** * Read data with supported envelope format and binary format. If envelope format is null, then read binary directly from file. @@ -50,10 +30,10 @@ fun IOPlugin.readData( dataFormat: IOFormat, envelopeFormatFactory: EnvelopeFormatFactory? = null, metaFile: Path = path.resolveSibling("${path.fileName}.meta"), - metaFileFormat: MetaFormat = JsonMetaFormat.default + metaFileFormat: MetaFormat = JsonMetaFormat ): Data { val externalMeta = if (Files.exists(metaFile)) { - metaFileFormat.readMetaFile(metaFile) + readMetaFile(metaFile) } else { null } From 41d0cdb2b157b68ae9bfbdf7242e3ad22f0474f3 Mon Sep 17 00:00:00 2001 From: Alexander Nozik Date: Wed, 13 Nov 2019 18:08:48 +0300 Subject: [PATCH 3/3] Fix large buffers IO. A lot of refactoring. --- build.gradle.kts | 2 +- .../kotlin/hep/dataforge/data/DataNode.kt | 27 ++++-- .../kotlin/hep/dataforge/data/dataCast.kt | 1 + .../dataforge/data/TypeFilteredDataNode.kt | 2 + .../kotlin/hep/dataforge/data/dataJVM.kt | 10 +-- .../kotlin/hep/dataforge/io/Envelope.kt | 1 + .../hep/dataforge/io/EnvelopeBuilder.kt | 1 + .../kotlin/hep/dataforge/io/EnvelopeParts.kt | 25 +++--- .../kotlin/hep/dataforge/io/IOFormat.kt | 29 +----- .../hep/dataforge/io/TaggedEnvelopeFormat.kt | 5 +- .../hep/dataforge/io/EnvelopePartsTest.kt | 6 +- .../kotlin/hep/dataforge/io/FileEnvelope.kt | 31 +------ .../io/{ioFormatsJVM.kt => fileIO.kt} | 56 ++++++++++-- .../dataforge/io/tcp/InputStreamAsInput.kt | 3 +- .../kotlin/hep/dataforge/io/FileBinaryTest.kt | 4 +- .../hep/dataforge/io/FileEnvelopeTest.kt | 2 +- .../hep/dataforge/workspace/fileData.kt | 90 ++++++++----------- 17 files changed, 147 insertions(+), 148 deletions(-) rename dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/{ioFormatsJVM.kt => fileIO.kt} (56%) diff --git a/build.gradle.kts b/build.gradle.kts index cd3e5d61..7f190853 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -6,7 +6,7 @@ plugins { id("scientifik.publish") version "0.2.2" apply false } -val dataforgeVersion by extra("0.1.5-dev-1") +val dataforgeVersion by extra("0.1.5-dev-2") val bintrayRepo by extra("dataforge") val githubProject by extra("dataforge-core") diff --git a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt index 589e875b..65c07676 100644 --- a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt +++ b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt @@ -13,16 +13,22 @@ import kotlin.reflect.KClass sealed class DataItem : MetaRepr { abstract val type: KClass + abstract val meta: Meta + class Node(val value: DataNode) : DataItem() { override val type: KClass get() = value.type override fun toMeta(): Meta = value.toMeta() + + override val meta: Meta get() = value.meta } class Leaf(val value: Data) : DataItem() { override val type: KClass get() = value.type override fun toMeta(): Meta = value.toMeta() + + override val meta: Meta get() = value.meta } } @@ -38,6 +44,8 @@ interface DataNode : MetaRepr { val items: Map> + val meta: Meta + override fun toMeta(): Meta = buildMeta { "type" put (type.simpleName ?: "undefined") "items" put { @@ -117,12 +125,9 @@ operator fun DataNode.iterator(): Iterator>> class DataTree internal constructor( override val type: KClass, - override val items: Map> -) : DataNode { - override fun toString(): String { - return super.toString() - } -} + override val items: Map>, + override val meta: Meta +) : DataNode private sealed class DataTreeBuilderItem { class Node(val tree: DataTreeBuilder) : DataTreeBuilderItem() @@ -136,6 +141,8 @@ private sealed class DataTreeBuilderItem { class DataTreeBuilder(val type: KClass) { private val map = HashMap>() + private var meta = MetaBuilder() + operator fun set(token: NameToken, node: DataTreeBuilder) { if (map.containsKey(token)) error("Tree entry with name $token is not empty") map[token] = DataTreeBuilderItem.Node(node) @@ -203,13 +210,19 @@ class DataTreeBuilder(val type: KClass) { infix fun String.put(block: DataTreeBuilder.() -> Unit) = set(toName(), DataTreeBuilder(type).apply(block)) + /** + * Update data with given node data and meta with node meta. + */ fun update(node: DataNode) { node.dataSequence().forEach { //TODO check if the place is occupied this[it.first] = it.second } + meta.update(node.meta) } + fun meta(block: MetaBuilder.() -> Unit) = meta.apply(block) + fun build(): DataTree { val resMap = map.mapValues { (_, value) -> when (value) { @@ -217,7 +230,7 @@ class DataTreeBuilder(val type: KClass) { is DataTreeBuilderItem.Node -> DataItem.Node(value.tree.build()) } } - return DataTree(type, resMap) + return DataTree(type, resMap, meta.seal()) } } diff --git a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/dataCast.kt b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/dataCast.kt index 556b77fc..0b9a4910 100644 --- a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/dataCast.kt +++ b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/dataCast.kt @@ -52,6 +52,7 @@ inline fun Data<*>.cast(): Data = cast(R::class) @Suppress("UNCHECKED_CAST") fun DataNode<*>.cast(type: KClass): DataNode { return object : DataNode { + override val meta: Meta get() = this@cast.meta override val type: KClass = type override val items: Map> get() = this@cast.items as Map> } diff --git a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/TypeFilteredDataNode.kt b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/TypeFilteredDataNode.kt index d24de964..331f3b0e 100644 --- a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/TypeFilteredDataNode.kt +++ b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/TypeFilteredDataNode.kt @@ -1,5 +1,6 @@ package hep.dataforge.data +import hep.dataforge.meta.Meta import hep.dataforge.names.NameToken import kotlin.reflect.KClass @@ -8,6 +9,7 @@ import kotlin.reflect.KClass * A zero-copy data node wrapper that returns only children with appropriate type. */ class TypeFilteredDataNode(val origin: DataNode<*>, override val type: KClass) : DataNode { + override val meta: Meta get() = origin.meta override val items: Map> by lazy { origin.items.mapNotNull { (key, item) -> when (item) { diff --git a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/dataJVM.kt b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/dataJVM.kt index 5b5507b2..29d048ed 100644 --- a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/dataJVM.kt +++ b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/dataJVM.kt @@ -30,12 +30,10 @@ fun Data<*>.filterIsInstance(type: KClass): Data? = * but could contain empty nodes */ fun DataNode<*>.filterIsInstance(type: KClass): DataNode { - return if (canCast(type)) { - cast(type) - } else if (this is TypeFilteredDataNode) { - origin.filterIsInstance(type) - } else { - TypeFilteredDataNode(this, type) + return when { + canCast(type) -> cast(type) + this is TypeFilteredDataNode -> origin.filterIsInstance(type) + else -> TypeFilteredDataNode(this, type) } } diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Envelope.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Envelope.kt index 61aeb4d2..7cb918df 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Envelope.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/Envelope.kt @@ -21,6 +21,7 @@ interface Envelope { val ENVELOPE_DATA_TYPE_KEY = ENVELOPE_NODE_KEY + "dataType" val ENVELOPE_DATA_ID_KEY = ENVELOPE_NODE_KEY + "dataID" val ENVELOPE_DESCRIPTION_KEY = ENVELOPE_NODE_KEY + "description" + val ENVELOPE_NAME_KEY = ENVELOPE_NODE_KEY + "name" //const val ENVELOPE_TIME_KEY = "@envelope.time" /** diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeBuilder.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeBuilder.kt index b8d0b660..354b4586 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeBuilder.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeBuilder.kt @@ -21,6 +21,7 @@ class EnvelopeBuilder { var dataType by metaBuilder.string(key = Envelope.ENVELOPE_DATA_TYPE_KEY) var dataID by metaBuilder.string(key = Envelope.ENVELOPE_DATA_ID_KEY) var description by metaBuilder.string(key = Envelope.ENVELOPE_DESCRIPTION_KEY) + var name by metaBuilder.string(key = Envelope.ENVELOPE_NAME_KEY) /** * Construct a binary and transform it into byte-array based buffer diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt index 0a8a207d..eb1dd696 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeParts.kt @@ -3,30 +3,31 @@ package hep.dataforge.io import hep.dataforge.context.Global import hep.dataforge.io.EnvelopeParts.FORMAT_META_KEY import hep.dataforge.io.EnvelopeParts.FORMAT_NAME_KEY -import hep.dataforge.io.EnvelopeParts.PARTS_DATA_TYPE +import hep.dataforge.io.EnvelopeParts.MULTIPART_DATA_TYPE import hep.dataforge.io.EnvelopeParts.SIZE_KEY import hep.dataforge.meta.* +import hep.dataforge.names.asName import hep.dataforge.names.plus import hep.dataforge.names.toName object EnvelopeParts { - val SIZE_KEY = Envelope.ENVELOPE_NODE_KEY + "parts" + "size" - val FORMAT_NAME_KEY = Envelope.ENVELOPE_NODE_KEY + "parts" + "format" - val FORMAT_META_KEY = Envelope.ENVELOPE_NODE_KEY + "parts" + "meta" + val MULTIPART_KEY = "multipart".asName() + val SIZE_KEY = Envelope.ENVELOPE_NODE_KEY + MULTIPART_KEY + "size" + val FORMAT_NAME_KEY = Envelope.ENVELOPE_NODE_KEY + MULTIPART_KEY + "format" + val FORMAT_META_KEY = Envelope.ENVELOPE_NODE_KEY + MULTIPART_KEY + "meta" - const val PARTS_DATA_TYPE = "envelope.parts" + const val MULTIPART_DATA_TYPE = "envelope.multipart" } /** * Append multiple serialized envelopes to the data block. Previous data is erased if it was present */ -fun EnvelopeBuilder.parts(formatFactory: EnvelopeFormatFactory, envelopes: Collection) { - dataType = PARTS_DATA_TYPE +fun EnvelopeBuilder.multipart(format: EnvelopeFormatFactory, envelopes: Collection) { + dataType = MULTIPART_DATA_TYPE meta { SIZE_KEY put envelopes.size - FORMAT_NAME_KEY put formatFactory.name.toString() + FORMAT_NAME_KEY put format.name.toString() } - val format = formatFactory() data { format.run { envelopes.forEach { @@ -36,15 +37,15 @@ fun EnvelopeBuilder.parts(formatFactory: EnvelopeFormatFactory, envelopes: Colle } } -fun EnvelopeBuilder.parts(formatFactory: EnvelopeFormatFactory, builder: suspend SequenceScope.() -> Unit) = - parts(formatFactory, sequence(builder).toList()) +fun EnvelopeBuilder.multipart(formatFactory: EnvelopeFormatFactory, builder: suspend SequenceScope.() -> Unit) = + multipart(formatFactory, sequence(builder).toList()) /** * If given envelope supports multipart data, return a sequence of those parts (could be empty). Otherwise return null. */ fun Envelope.parts(io: IOPlugin = Global.plugins.fetch(IOPlugin)): Sequence? { return when (dataType) { - PARTS_DATA_TYPE -> { + MULTIPART_DATA_TYPE -> { val size = meta[SIZE_KEY].int ?: error("Unsized parts not supported yet") val formatName = meta[FORMAT_NAME_KEY].string?.toName() ?: error("Inferring parts format is not supported at the moment") diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/IOFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/IOFormat.kt index 093ffbc8..bd1e54f4 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/IOFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/IOFormat.kt @@ -16,7 +16,6 @@ import kotlinx.serialization.ImplicitReflectionSerializer import kotlinx.serialization.KSerializer import kotlinx.serialization.cbor.Cbor import kotlinx.serialization.serializer -import kotlin.math.min import kotlin.reflect.KClass /** @@ -80,33 +79,9 @@ inline fun buildPacketWithoutPool(headerSizeHint: Int = 0, block: BytePacketBuil } fun IOFormat.writePacket(obj: T): ByteReadPacket = buildPacket { writeObject(obj) } +//TODO Double buffer copy. fix all that with IO-2 fun IOFormat.writeBytes(obj: T): ByteArray = buildPacket { writeObject(obj) }.readBytes() -fun IOFormat.readBytes(array: ByteArray): T { - //= ByteReadPacket(array).readThis() - val byteArrayInput: Input = object : AbstractInput( - IoBuffer.Pool.borrow(), - remaining = array.size.toLong(), - pool = IoBuffer.Pool - ) { - var written = 0 - override fun closeSource() { - // do nothing - } - - override fun fill(): IoBuffer? { - if (array.size - written <= 0) return null - - return IoBuffer.Pool.fill { - reserveEndGap(IoBuffer.ReservedSize) - val toWrite = min(capacity, array.size - written) - writeFully(array, written, toWrite) - written += toWrite - } - } - - } - return byteArrayInput.readObject() -} +fun IOFormat.readBytes(array: ByteArray): T = buildPacket { writeFully(array) }.readObject() object DoubleIOFormat : IOFormat, IOFormatFactory { override fun invoke(meta: Meta, context: Context): IOFormat = this diff --git a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt index 5f9164dc..a95b7bfb 100644 --- a/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt +++ b/dataforge-io/src/commonMain/kotlin/hep/dataforge/io/TaggedEnvelopeFormat.kt @@ -64,7 +64,10 @@ class TaggedEnvelopeFormat( val metaFormat = io.metaFormat(tag.metaFormatKey) ?: error("Meta format with key ${tag.metaFormatKey} not found") - val metaPacket = ByteReadPacket(readBytes(tag.metaSize.toInt())) + val metaBytes = readBytes(tag.metaSize.toInt()) + val metaPacket = buildPacket { + writeFully(metaBytes) + } val dataBytes = readBytes(tag.dataSize.toInt()) val meta = metaFormat.run { metaPacket.readObject() } diff --git a/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt b/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt index c33b3179..d123d632 100644 --- a/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt +++ b/dataforge-io/src/commonTest/kotlin/hep/dataforge/io/EnvelopePartsTest.kt @@ -14,11 +14,14 @@ class EnvelopePartsTest { } data { writeText("Hello World $it") + repeat(200){ + writeInt(it) + } } } } val partsEnvelope = Envelope { - parts(TaggedEnvelopeFormat, envelopes) + multipart(TaggedEnvelopeFormat, envelopes) } @Test @@ -27,6 +30,7 @@ class EnvelopePartsTest { val reconstructed = TaggedEnvelopeFormat.readBytes(bytes) val parts = reconstructed.parts()?.toList() ?: emptyList() assertEquals(2, parts[2].meta["value"].int) + println(reconstructed.data!!.size) } } \ No newline at end of file diff --git a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/FileEnvelope.kt b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/FileEnvelope.kt index 3187cd54..5f21e6ae 100644 --- a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/FileEnvelope.kt +++ b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/FileEnvelope.kt @@ -1,9 +1,7 @@ package hep.dataforge.io -import hep.dataforge.meta.EmptyMeta import hep.dataforge.meta.Meta import kotlinx.io.nio.asInput -import kotlinx.io.nio.asOutput import java.nio.file.Files import java.nio.file.Path import java.nio.file.StandardOpenOption @@ -15,7 +13,7 @@ class FileEnvelope internal constructor(val path: Path, val format: EnvelopeForm init { val input = Files.newByteChannel(path, StandardOpenOption.READ).asInput() - partialEnvelope = format.run { input.use { it.readPartial()} } + partialEnvelope = format.run { input.use { it.readPartial() } } } override val meta: Meta get() = partialEnvelope.meta @@ -23,30 +21,3 @@ class FileEnvelope internal constructor(val path: Path, val format: EnvelopeForm override val data: Binary? = FileBinary(path, partialEnvelope.dataOffset, partialEnvelope.dataSize) } -fun IOPlugin.readEnvelopeFile( - path: Path, - formatFactory: EnvelopeFormatFactory = TaggedEnvelopeFormat, - formatMeta: Meta = EmptyMeta -): FileEnvelope { - val format = formatFactory(formatMeta, context) - return FileEnvelope(path, format) -} - -fun IOPlugin.writeEnvelopeFile( - path: Path, - envelope: Envelope, - formatFactory: EnvelopeFormatFactory = TaggedEnvelopeFormat, - formatMeta: Meta = EmptyMeta -) { - val output = Files.newByteChannel( - path, - StandardOpenOption.WRITE, - StandardOpenOption.CREATE, - StandardOpenOption.TRUNCATE_EXISTING - ).asOutput() - - with(formatFactory(formatMeta, context)) { - output.writeObject(envelope) - } -} - diff --git a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/fileIO.kt similarity index 56% rename from dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt rename to dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/fileIO.kt index d0ce18b0..14e4c077 100644 --- a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/ioFormatsJVM.kt +++ b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/fileIO.kt @@ -18,24 +18,41 @@ inline fun IOPlugin.resolveIOFormat(): IOFormat? { /** * Read file containing meta using given [formatOverride] or file extension to infer meta type. + * If [path] is a directory search for file starting with `meta` in it */ fun IOPlugin.readMetaFile(path: Path, formatOverride: MetaFormat? = null, descriptor: NodeDescriptor? = null): Meta { if (!Files.exists(path)) error("Meta file $path does not exist") - val extension = path.fileName.toString().substringAfterLast('.') + + val actualPath: Path = if (Files.isDirectory(path)) { + Files.list(path).asSequence().find { it.fileName.startsWith("meta") } + ?: error("The directory $path does not contain meta file") + } else { + path + } + val extension = actualPath.fileName.toString().substringAfterLast('.') val metaFormat = formatOverride ?: metaFormat(extension) ?: error("Can't resolve meta format $extension") return metaFormat.run { - Files.newByteChannel(path, StandardOpenOption.READ).asInput().use { it.readMeta(descriptor) } + Files.newByteChannel(actualPath, StandardOpenOption.READ).asInput().use { it.readMeta(descriptor) } } } +/** + * Write meta to file using [metaFormat]. If [path] is a directory, write a file with name equals name of [metaFormat]. + * Like "meta.json" + */ fun IOPlugin.writeMetaFile( path: Path, - metaFormat: MetaFormat = JsonMetaFormat, + metaFormat: MetaFormatFactory = JsonMetaFormat, descriptor: NodeDescriptor? = null ) { + val actualPath = if (Files.isDirectory(path)) { + path.resolve(metaFormat.name.toString()) + } else { + path + } metaFormat.run { - Files.newByteChannel(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW).asOutput().use { + Files.newByteChannel(actualPath, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW).asOutput().use { it.writeMeta(meta, descriptor) } } @@ -43,9 +60,19 @@ fun IOPlugin.writeMetaFile( /** * Read and envelope from file if the file exists, return null if file does not exist. + * + * If file is directory, then expect two files inside: + * * **meta.** for meta + * * **data** for data + * + * If the file is envelope read it using [EnvelopeFormatFactory.peekFormat] functionality to infer format. + * + * If the file is not an envelope and [readNonEnvelopes] is true, return an Envelope without meta, using file as binary. + * + * Return null otherwise. */ @DFExperimental -fun IOPlugin.readEnvelopeFromFile(path: Path, readNonEnvelopes: Boolean = false): Envelope? { +fun IOPlugin.readEnvelopeFile(path: Path, readNonEnvelopes: Boolean = false): Envelope? { if (!Files.exists(path)) return null //read two-files directory @@ -74,7 +101,7 @@ fun IOPlugin.readEnvelopeFromFile(path: Path, readNonEnvelopes: Boolean = false) val formats = envelopeFormatFactories.mapNotNull { factory -> binary.read { - factory.peekFormat(this@readEnvelopeFromFile, this@read) + factory.peekFormat(this@readEnvelopeFile, this@read) } } return when (formats.size) { @@ -90,4 +117,21 @@ fun IOPlugin.readEnvelopeFromFile(path: Path, readNonEnvelopes: Boolean = false) } else -> error("Envelope format file recognition clash") } +} + +fun IOPlugin.writeEnvelopeFile( + path: Path, + envelope: Envelope, + format: EnvelopeFormat = TaggedEnvelopeFormat +) { + val output = Files.newByteChannel( + path, + StandardOpenOption.WRITE, + StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING + ).asOutput() + + with(format) { + output.writeObject(envelope) + } } \ No newline at end of file diff --git a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/tcp/InputStreamAsInput.kt b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/tcp/InputStreamAsInput.kt index 1c711be0..eb743625 100644 --- a/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/tcp/InputStreamAsInput.kt +++ b/dataforge-io/src/jvmMain/kotlin/hep/dataforge/io/tcp/InputStreamAsInput.kt @@ -3,7 +3,6 @@ package hep.dataforge.io.tcp import kotlinx.io.core.AbstractInput import kotlinx.io.core.Input import kotlinx.io.core.IoBuffer -import kotlinx.io.core.IoBuffer.Companion.NoPool import kotlinx.io.core.writePacket import kotlinx.io.streams.readPacketAtMost import java.io.InputStream @@ -13,7 +12,7 @@ import java.io.InputStream */ internal class InputStreamAsInput( private val stream: InputStream -) : AbstractInput(pool = NoPool) { +) : AbstractInput(pool = IoBuffer.Pool) { override fun fill(): IoBuffer? { diff --git a/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt b/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt index 545601cd..d8c7c67a 100644 --- a/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt +++ b/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileBinaryTest.kt @@ -49,8 +49,8 @@ class FileBinaryTest { val tmpPath = Files.createTempFile("dataforge_test", ".df") Global.io.writeEnvelopeFile(tmpPath, envelope) - val binary = Global.io.readEnvelopeFile(tmpPath).data!! - assertEquals(binary.size?.toInt(), binary.toBytes().size) + val binary = Global.io.readEnvelopeFile(tmpPath)?.data!! + assertEquals(binary.size.toInt(), binary.toBytes().size) } } \ No newline at end of file diff --git a/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileEnvelopeTest.kt b/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileEnvelopeTest.kt index ba7f7cc5..f4847cfd 100644 --- a/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileEnvelopeTest.kt +++ b/dataforge-io/src/jvmTest/kotlin/hep/dataforge/io/FileEnvelopeTest.kt @@ -25,7 +25,7 @@ class FileEnvelopeTest { val tmpPath = Files.createTempFile("dataforge_test", ".df") Global.io.writeEnvelopeFile(tmpPath,envelope) println(tmpPath.toUri()) - val restored: Envelope = Global.io.readEnvelopeFile(tmpPath) + val restored: Envelope = Global.io.readEnvelopeFile(tmpPath)!! assertTrue { envelope.contentEquals(restored) } } } \ No newline at end of file diff --git a/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt b/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt index 4fbfa72a..b5ecb519 100644 --- a/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt +++ b/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/fileData.kt @@ -1,17 +1,15 @@ package hep.dataforge.workspace -import hep.dataforge.data.Data -import hep.dataforge.data.DataNode -import hep.dataforge.data.DataTreeBuilder -import hep.dataforge.data.datum -import hep.dataforge.io.* -import hep.dataforge.meta.EmptyMeta -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.withContext -import kotlinx.io.nio.asInput +import hep.dataforge.data.* +import hep.dataforge.io.Envelope +import hep.dataforge.io.IOFormat +import hep.dataforge.io.IOPlugin +import hep.dataforge.io.readEnvelopeFile +import hep.dataforge.meta.Meta +import hep.dataforge.meta.get +import hep.dataforge.meta.string import java.nio.file.Files import java.nio.file.Path -import java.nio.file.StandardOpenOption import kotlin.reflect.KClass @@ -20,54 +18,30 @@ import kotlin.reflect.KClass * The operation is blocking since it must read meta header. The reading of envelope body is lazy * @param type explicit type of data read * @param dataFormat binary format - * @param envelopeFormatFactory the format of envelope. If null, file is read directly + * @param envelopeFormat the format of envelope. If null, file is read directly * @param metaFile the relative file for optional meta override * @param metaFileFormat the meta format for override */ -fun IOPlugin.readData( +fun IOPlugin.readDataFile( path: Path, type: KClass, - dataFormat: IOFormat, - envelopeFormatFactory: EnvelopeFormatFactory? = null, - metaFile: Path = path.resolveSibling("${path.fileName}.meta"), - metaFileFormat: MetaFormat = JsonMetaFormat + formatResolver: (Meta) -> IOFormat ): Data { - val externalMeta = if (Files.exists(metaFile)) { - readMetaFile(metaFile) - } else { - null - } - return if (envelopeFormatFactory == null) { - Data(type, externalMeta ?: EmptyMeta) { - withContext(Dispatchers.IO) { - dataFormat.run { - Files.newByteChannel(path, StandardOpenOption.READ) - .asInput() - .readObject() - } - } - } - } else { - readEnvelopeFile(path, envelopeFormatFactory).let { - if (externalMeta == null) { - it - } else { - it.withMetaLayers(externalMeta) - } - }.toData(type, dataFormat) - } + val envelope = readEnvelopeFile(path, true) ?: error("Can't read data from $path") + val format = formatResolver(envelope.meta) + return envelope.toData(type, format) } //TODO wants multi-receiver fun DataTreeBuilder.file( plugin: IOPlugin, path: Path, - dataFormat: IOFormat, - envelopeFormatFactory: EnvelopeFormatFactory? = null + formatResolver: (Meta) -> IOFormat ) { plugin.run { - val data = readData(path, type, dataFormat, envelopeFormatFactory) - val name = path.fileName.toString().replace(".df", "") + val data = readDataFile(path, type, formatResolver) + val name = data.meta[Envelope.ENVELOPE_NAME_KEY].string + ?: path.fileName.toString().replace(".df", "") datum(name, data) } } @@ -75,23 +49,35 @@ fun DataTreeBuilder.file( /** * Read the directory as a data node */ -fun IOPlugin.readDataNode( +fun IOPlugin.readDataDirectory( path: Path, type: KClass, - dataFormat: IOFormat, - envelopeFormatFactory: EnvelopeFormatFactory? = null + formatResolver: (Meta) -> IOFormat ): DataNode { if (!Files.isDirectory(path)) error("Provided path $this is not a directory") return DataNode(type) { Files.list(path).forEach { path -> if (!path.fileName.toString().endsWith(".meta")) { - file(this@readDataNode,path, dataFormat, envelopeFormatFactory) + file(this@readDataDirectory, path, formatResolver) } } } } -//suspend fun Path.writeData( -// data: Data, -// format: IOFormat, -// ) \ No newline at end of file +fun DataTreeBuilder.directory( + plugin: IOPlugin, + path: Path, + formatResolver: (Meta) -> IOFormat +) { + plugin.run { + val data = readDataDirectory(path, type, formatResolver) + val name = data.meta[Envelope.ENVELOPE_NAME_KEY].string + ?: path.fileName.toString().replace(".df", "") + node(name, data) + } +} + + + + +