Meta and Evelope format factories now implement Meta and Envelope formats (default representation).

dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt

@@ -64,21 +64,13 @@ val <T : Any> DataItem<T>?.node: DataNode<T>? get() = (this as? DataItem.Node<T>
 val <T : Any> DataItem<T>?.data: Data<T>? get() = (this as? DataItem.Leaf<T>)?.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 {
+fun DataNode<*>.launchAll(scope: CoroutineScope): Job = scope.launch {
-    when (it) {
+    items.values.forEach {
-        is DataItem.Node<*> -> it.value.startAll(scope)
+        when (it) {
-        is DataItem.Leaf<*> -> it.value.start(scope)
+            is DataItem.Node<*> -> it.value.launchAll(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)
         }
     }
 }

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() }
+
     }
 }

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) }
     }
 }

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 <R> 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 <R> read(from: UInt, size: UInt, block: Input.() -> R): R {
         error("The binary is empty")
     }
-
 }
 
 @ExperimentalUnsignedTypes
@@ -79,9 +82,9 @@ fun <T : Any> Binary.readWith(format: IOFormat<T>): T = format.run {
     }
 }
 
-fun <T : Any> IOFormat<T>.writeBinary(obj: T): Binary {
+//fun <T : Any> IOFormat<T>.writeBinary(obj: T): Binary {
-    val packet = buildPacket {
+//    val packet = buildPacket {
-        writeObject(obj)
+//        writeObject(obj)
-    }
+//    }
-    return ArrayBinary(packet.readBytes())
+//    return ArrayBinary(packet.readBytes())
-}
+//}

dataforge-io/src/commonMain/kotlin/hep/dataforge/io/EnvelopeFormat.kt

@@ -31,7 +31,7 @@ interface EnvelopeFormat : IOFormat<Envelope> {
 }
 
 @Type(ENVELOPE_FORMAT_TYPE)
-interface EnvelopeFormatFactory : IOFormatFactory<Envelope> {
+interface EnvelopeFormatFactory : IOFormatFactory<Envelope>, EnvelopeFormat {
     override val name: Name get() = "envelope".asName()
     override val type: KClass<out Envelope> get() = Envelope::class
 

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()
     }
 }
 

dataforge-io/src/commonMain/kotlin/hep/dataforge/io/MetaFormat.kt

@@ -27,7 +27,7 @@ interface MetaFormat : IOFormat<Meta> {
 }
 
 @Type(META_FORMAT_TYPE)
-interface MetaFormatFactory : IOFormatFactory<Meta> {
+interface MetaFormatFactory : IOFormatFactory<Meta>, MetaFormat {
     override val name: Name get() = "meta".asName()
 
     override val type: KClass<out Meta> 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) }
 }
 

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() }
     }
 
 }

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 {

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)

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 bytes = TaggedEnvelopeFormat.writeBytes(partsEnvelope)
-        val reconstructed = TaggedEnvelopeFormat.default.readBytes(bytes)
+        val reconstructed = TaggedEnvelopeFormat.readBytes(bytes)
-        val parts = reconstructed.parts().toList()
+        val parts = reconstructed.parts()?.toList() ?: emptyList()
         assertEquals(2, parts[2].meta["value"].int)
     }
 

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 <reified T : Any, reified R : Any> 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 <reified T : Any, reified R : Any> FunctionServer.function(
+    functionName: String
+): (suspend (T) -> R) {
+    val plugin = context.plugins.get<IOPlugin>() ?: error("IO plugin not loaded")
+    val meta = plugin.generateFunctionMeta<T, R>(functionName)
+    return function(meta)
+}

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.meta.DFExperimental
-import hep.dataforge.io.functions.FunctionServer.Companion.FUNCTION_NAME_KEY
+import hep.dataforge.meta.EmptyMeta
-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.Meta
-import hep.dataforge.meta.buildMeta
+import kotlinx.io.nio.asInput
-import hep.dataforge.names.Name
 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 <reified T : Any> IOPlugin.resolveIOFormat(): IOFormat<T>? {
     return ioFormats.values.find { it.type.isSuperclassOf(T::class) } as IOFormat<T>?
 }
 
-fun IOPlugin.resolveIOFormatName(type: KClass<*>): Name {
+/**
-    return ioFormats.entries.find { it.value.type.isSuperclassOf(type) }?.key
+ * Read file containing meta using given [formatOverride] or file extension to infer meta type.
-        ?: error("Can't resolve IOFormat for type $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 <reified T : Any, reified R : Any> IOPlugin.generateFunctionMeta(functionName: String): Meta = buildMeta {
+fun IOPlugin.writeMetaFile(
-    FUNCTION_NAME_KEY put functionName
+    path: Path,
-    INPUT_FORMAT_KEY put resolveIOFormatName(T::class).toString()
+    metaFormat: MetaFormat = JsonMetaFormat,
-    OUTPUT_FORMAT_KEY put resolveIOFormatName(R::class).toString()
+    descriptor: NodeDescriptor? = null
-}
+) {
-
+    metaFormat.run {
-inline fun <reified T : Any, reified R : Any> FunctionServer.function(
+        Files.newByteChannel(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW).asOutput().use {
-    functionName: String
+            it.writeMeta(meta, descriptor)
-): (suspend (T) -> R) {
+        }
-    val plugin = context.plugins.get<IOPlugin>() ?: error("IO plugin not loaded")
+    }
-    val meta = plugin.generateFunctionMeta<T, R>(functionName)
-    return function(meta)
 }
 
 /**
- * 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) {
+@DFExperimental
-    format.run {
+fun IOPlugin.readEnvelopeFromFile(path: Path, readNonEnvelopes: Boolean = false): Envelope? {
-        Files.newByteChannel(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW)
+    if (!Files.exists(path)) return null
-            .asOutput()
+
-            .writeMeta(this@write, descriptor)
+    //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")
     }
 }

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)
     }
 
 }

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

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 <T : Any> IOPlugin.readData(
     dataFormat: IOFormat<T>,
     envelopeFormatFactory: EnvelopeFormatFactory? = null,
     metaFile: Path = path.resolveSibling("${path.fileName}.meta"),
-    metaFileFormat: MetaFormat = JsonMetaFormat.default
+    metaFileFormat: MetaFormat = JsonMetaFormat
 ): Data<T> {
     val externalMeta = if (Files.exists(metaFile)) {
-        metaFileFormat.readMetaFile(metaFile)
+        readMetaFile(metaFile)
     } else {
         null
     }