Add support for more data source types

core/src/main/java/io/xpipe/core/source/CollectionDataSourceDescriptor.java

@@ -0,0 +1,30 @@
+package io.xpipe.core.source;
+
+import io.xpipe.core.store.DataStore;
+
+public interface CollectionDataSourceDescriptor<DS extends DataStore> extends DataSourceDescriptor<DS> {
+
+    @Override
+    default DataSourceInfo determineInfo(DS store) throws Exception {
+        try (var con = openReadConnection(store)) {
+            var c = (int) con.listEntries().count();
+            return new DataSourceInfo.Structure(c);
+        }
+    }
+
+    default CollectionReadConnection openReadConnection(DS store) throws Exception {
+        var con = newReadConnection(store);
+        con.init();
+        return con;
+    }
+
+    default CollectionWriteConnection openWriteConnection(DS store) throws Exception {
+        var con = newWriteConnection(store);
+        con.init();
+        return con;
+    }
+
+    CollectionWriteConnection newWriteConnection(DS store);
+
+    CollectionReadConnection newReadConnection(DS store);
+}

core/src/main/java/io/xpipe/core/source/CollectionReadConnection.java

@@ -0,0 +1,24 @@
+package io.xpipe.core.source;
+
+import lombok.SneakyThrows;
+
+import java.util.stream.Stream;
+
+public interface CollectionReadConnection extends DataSourceReadConnection {
+
+    <T extends DataSourceReadConnection> T open(String entry) throws Exception;
+
+    Stream<String> listEntries() throws Exception;
+
+    @SneakyThrows
+    default void forward(DataSourceConnection con) throws Exception {
+        try (var tCon = (CollectionWriteConnection) con) {
+            tCon.init();
+            listEntries().forEach(s -> {
+                try (var subCon = open(s)) {
+                    ((CollectionWriteConnection) con).write(s, subCon);
+                }
+            });
+        }
+    }
+}

core/src/main/java/io/xpipe/core/source/CollectionWriteConnection.java

@@ -0,0 +1,6 @@
+package io.xpipe.core.source;
+
+public interface CollectionWriteConnection extends DataSourceConnection {
+
+    void write(String entry, DataSourceReadConnection con) throws Exception;
+}

core/src/main/java/io/xpipe/core/source/DataSourceDescriptor.java

@@ -27,11 +27,6 @@ public interface DataSourceDescriptor<DS extends DataStore> {
      */
     DataSourceInfo determineInfo(DS store) throws Exception;
 
-    /**
-     * Returns the general data source type.
-     */
-    DataSourceType getType();
-
     DataSourceReadConnection openReadConnection(DS store) throws Exception;
 
     DataSourceConnection openWriteConnection(DS store) throws Exception;

core/src/main/java/io/xpipe/core/source/DataSourceInfo.java

@@ -7,8 +7,6 @@ import io.xpipe.core.data.type.TupleType;
 import lombok.EqualsAndHashCode;
 import lombok.Value;
 
-import java.nio.ByteOrder;
-import java.nio.charset.Charset;
 import java.util.OptionalInt;
 
 /**
@@ -53,8 +51,11 @@ public abstract class DataSourceInfo {
     @JsonTypeName("structure")
     public static class Structure extends DataSourceInfo {
 
+        int entries;
+
         @JsonCreator
-        public Structure() {
+        public Structure(int entries) {
+            this.entries = entries;
         }
 
         @Override
@@ -63,17 +64,32 @@ public abstract class DataSourceInfo {
         }
     }
 
+    @EqualsAndHashCode(callSuper = false)
+    @Value
+    @JsonTypeName("collection")
+    public static class Collection extends DataSourceInfo {
+
+        int entries;
+
+        @JsonCreator
+        public Collection(int entries) {
+            this.entries = entries;
+        }
+
+        @Override
+        public DataSourceType getType() {
+            return DataSourceType.COLLECTION;
+        }
+    }
+
     @EqualsAndHashCode(callSuper = false)
     @Value
     @JsonTypeName("text")
     public static class Text extends DataSourceInfo {
-        Charset charset;
-
         int lineCount;
 
         @JsonCreator
-        public Text(Charset charset, int lineCount) {
-            this.charset = charset;
+        public Text(int lineCount) {
             this.lineCount = lineCount;
         }
 
@@ -89,12 +105,10 @@ public abstract class DataSourceInfo {
     @JsonTypeName("raw")
     public static class Raw extends DataSourceInfo {
         int byteCount;
-        ByteOrder byteOrder;
 
         @JsonCreator
-        public Raw(int byteCount, ByteOrder byteOrder) {
+        public Raw(int byteCount) {
             this.byteCount = byteCount;
-            this.byteOrder = byteOrder;
         }
 
         @Override

core/src/main/java/io/xpipe/core/source/DataSourceType.java

@@ -18,5 +18,8 @@ public enum DataSourceType {
     TEXT,
 
     @JsonProperty("raw")
-    RAW
+    RAW,
+
+    @JsonProperty("collection")
+    COLLECTION
 }

core/src/main/java/io/xpipe/core/source/RawDataSourceDescriptor.java

@@ -0,0 +1,35 @@
+package io.xpipe.core.source;
+
+import io.xpipe.core.store.DataStore;
+
+public abstract class RawDataSourceDescriptor <DS extends DataStore> implements DataSourceDescriptor<DS>  {
+
+    private static final int MAX_BYTES_READ = 100000;
+
+    @Override
+    public DataSourceInfo determineInfo(DS store) throws Exception {
+        try (var con = openReadConnection(store)) {
+            var b = con.readBytes(MAX_BYTES_READ);
+            int usedCount = b.length == MAX_BYTES_READ ? -1 : b.length;
+            return new DataSourceInfo.Raw(usedCount);
+        }
+    }
+
+    @Override
+    public RawReadConnection openReadConnection(DS store) throws Exception {
+        var con = newReadConnection(store);
+        con.init();
+        return con;
+    }
+
+    @Override
+    public RawWriteConnection openWriteConnection(DS store) throws Exception {
+        var con = newWriteConnection(store);
+        con.init();
+        return con;
+    }
+
+    protected abstract RawWriteConnection newWriteConnection(DS store);
+
+    protected abstract RawReadConnection newReadConnection(DS store);
+}

core/src/main/java/io/xpipe/core/source/RawReadConnection.java

@@ -0,0 +1,18 @@
+package io.xpipe.core.source;
+
+public interface RawReadConnection extends DataSourceReadConnection {
+
+    byte[] readBytes(int max) throws Exception;
+
+    int BUFFER_SIZE = 8192;
+
+    default void forward(DataSourceConnection con) throws Exception {
+        try (var tCon = (RawWriteConnection) con) {
+            tCon.init();
+            byte[] b;
+            while ((b = readBytes(BUFFER_SIZE)).length > 0) {
+                tCon.write(b);
+            }
+        }
+    }
+}

core/src/main/java/io/xpipe/core/source/RawWriteConnection.java

@@ -0,0 +1,6 @@
+package io.xpipe.core.source;
+
+public interface RawWriteConnection extends DataSourceConnection {
+
+    void write(byte[] bytes) throws Exception;
+}

core/src/main/java/io/xpipe/core/source/StructureDataSourceDescriptor.java

@@ -1,27 +1,43 @@
 package io.xpipe.core.source;
 
+import io.xpipe.core.data.node.DataStructureNode;
 import io.xpipe.core.store.DataStore;
 
-public abstract class StructureDataSourceDescriptor<DS extends DataStore> implements DataSourceDescriptor<DS> {
+public interface StructureDataSourceDescriptor<DS extends DataStore> extends DataSourceDescriptor<DS> {
 
-    public final StructureReadConnection openReadConnection(DS store) throws Exception {
+    private int countEntries(DataStructureNode n) {
+        if (n.isValue()) {
+            return 1;
+        }
+
+        int c = 0;
+        for (int i = 0; i < n.size(); i++) {
+            c += countEntries(n.at(i));
+        }
+        return c;
+    }
+
+    @Override
+    default DataSourceInfo determineInfo(DS store) throws Exception {
+        try (var con = openReadConnection(store)) {
+            var n = con.read();
+            var c = countEntries(n);
+            return new DataSourceInfo.Structure(c);
+        }
+    }
+
+    default StructureReadConnection openReadConnection(DS store) throws Exception {
         var con = newReadConnection(store);
         con.init();
         return con;
     }
 
-    public final StructureWriteConnection openWriteConnection(DS store) throws Exception {
+    default StructureWriteConnection openWriteConnection(DS store) throws Exception {
         var con = newWriteConnection(store);
         con.init();
         return con;
     }
+    StructureWriteConnection newWriteConnection(DS store);
 
-    protected abstract StructureWriteConnection newWriteConnection(DS store);
-
-    protected abstract StructureReadConnection newReadConnection(DS store);
-
-    @Override
-    public DataSourceType getType() {
-        return DataSourceType.STRUCTURE;
-    }
+    StructureReadConnection newReadConnection(DS store);
 }

core/src/main/java/io/xpipe/core/source/TableDataSourceDescriptor.java

@@ -19,9 +19,4 @@ public abstract class TableDataSourceDescriptor<DS extends DataStore> implements
     protected abstract TableWriteConnection newWriteConnection(DS store);
 
     protected abstract TableReadConnection newReadConnection(DS store);
-
-    @Override
-    public DataSourceType getType() {
-        return DataSourceType.TABLE;
-    }
 }

core/src/main/java/io/xpipe/core/source/TextDataSourceDescriptor.java

@@ -4,9 +4,15 @@ import io.xpipe.core.store.DataStore;
 
 public abstract class TextDataSourceDescriptor<DS extends DataStore> implements DataSourceDescriptor<DS>  {
 
+    private static final int MAX_LINE_READ = 1000;
+
     @Override
-    public DataSourceType getType() {
-        return DataSourceType.TEXT;
+    public DataSourceInfo determineInfo(DS store) throws Exception {
+        try (var con = openReadConnection(store)) {
+            int count = (int) con.lines().limit(MAX_LINE_READ).count();
+            int usedCount = count == MAX_LINE_READ ? -1 : count;
+            return new DataSourceInfo.Text(usedCount);
+        }
     }
 
     @Override

core/src/main/java/io/xpipe/core/source/TextReadConnection.java

@@ -1,23 +1,31 @@
 package io.xpipe.core.source;
 
-import java.util.List;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.stream.Stream;
 
 public interface TextReadConnection extends DataSourceReadConnection {
 
-    /**
-     * Reads the complete contents.
-     */
-    String readAll() throws Exception;
-
-    List<String> readAllLines() throws Exception;
-
-    String readLine() throws Exception;
-
     Stream<String> lines() throws Exception;
 
     boolean isFinished() throws Exception;
 
+    default void forwardLines(OutputStream out, int maxLines) throws Exception {
+        if (maxLines == 0) {
+            return;
+        }
+
+        int counter = 0;
+        for (var it = lines().iterator(); it.hasNext(); counter++) {
+            if (counter == maxLines) {
+                break;
+            }
+
+            out.write(it.next().getBytes(StandardCharsets.UTF_8));
+            out.write("\n".getBytes(StandardCharsets.UTF_8));
+        }
+    }
+
     default void forward(DataSourceConnection con) throws Exception {
         try (var tCon = (TextWriteConnection) con) {
             tCon.init();

core/src/main/java/io/xpipe/core/util/JacksonHelper.java

@@ -1,6 +1,7 @@
 package io.xpipe.core.util;
 
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.Module;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
@@ -22,6 +23,7 @@ public class JacksonHelper {
         ObjectMapper objectMapper = INSTANCE;
         objectMapper.enable(SerializationFeature.INDENT_OUTPUT);
         objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
+        objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
 
         objectMapper.registerModules(findModules(layer));
         objectMapper.setVisibility(objectMapper.getSerializationConfig().getDefaultVisibilityChecker()