Add textual flag to value node, prepare for text data sources, and various other small fixes

core/src/main/java/io/xpipe/core/config/ConfigOption.java

@@ -0,0 +1,14 @@
+package io.xpipe.core.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Value;
+
+@Value
+@Builder
+@AllArgsConstructor(onConstructor_={@JsonCreator})
+public class ConfigOption {
+    String name;
+    String key;
+}

core/src/main/java/io/xpipe/core/config/ConfigOptionSet.java

@@ -0,0 +1,23 @@
+package io.xpipe.core.config;
+
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Singular;
+import lombok.Value;
+
+import java.util.List;
+
+@Value
+@Builder
+@AllArgsConstructor(onConstructor_={@JsonCreator})
+public class ConfigOptionSet {
+
+    public static ConfigOptionSet empty() {
+        return new ConfigOptionSet(List.of());
+    }
+
+    @Singular
+    List<ConfigOption> options;
+}

core/src/main/java/io/xpipe/core/config/ConfigOptionSetInstance.java

@@ -0,0 +1,26 @@
+package io.xpipe.core.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import lombok.AllArgsConstructor;
+import lombok.Value;
+
+import java.util.Map;
+
+@Value
+@AllArgsConstructor(onConstructor_={@JsonCreator})
+public class ConfigOptionSetInstance {
+
+    /**
+     * The available configuration options.
+     */
+    ConfigOptionSet configOptions;
+
+    /**
+     * The current configuration options that are set.
+     */
+    Map<String, String> currentValues;
+
+    public boolean isComplete() {
+        return currentValues.size() == configOptions.getOptions().size();
+    }
+}

core/src/main/java/io/xpipe/core/data/generic/GenericArrayReader.java

@@ -127,9 +127,9 @@ public class GenericArrayReader implements GenericAbstractReader {
     }
 
     @Override
-    public void onValue(byte[] value) {
+    public void onValue(byte[] value, boolean textual) {
         if (currentReader != null) {
-            currentReader.onValue(value);
+            currentReader.onValue(value, textual);
             return;
         }
 
@@ -141,7 +141,7 @@ public class GenericArrayReader implements GenericAbstractReader {
             throw new IllegalStateException("Array is full but got another value");
         }
 
-        put(ValueNode.mutable(value));
+        put(ValueNode.mutable(value, textual));
     }
 
     @Override

core/src/main/java/io/xpipe/core/data/generic/GenericDataStreamCallback.java

@@ -17,6 +17,6 @@ public interface GenericDataStreamCallback {
     default void onTupleEnd() {
     }
 
-    default void onValue(byte[] value) {
+    default void onValue(byte[] value, boolean textual) {
     }
 }

core/src/main/java/io/xpipe/core/data/generic/GenericDataStreamParser.java

@@ -79,8 +79,9 @@ public class GenericDataStreamParser {
     }
 
     private static void parseValue(InputStream in, GenericDataStreamCallback cb) throws IOException {
+        var textual = in.read() != 0;
         var size = in.read();
         var data = in.readNBytes(size);
-        cb.onValue(data);
+        cb.onValue(data, textual);
     }
 }

core/src/main/java/io/xpipe/core/data/generic/GenericDataStreamWriter.java

@@ -53,6 +53,7 @@ public class GenericDataStreamWriter {
 
     private static void writeValue(OutputStream out, ValueNode value) throws IOException {
         out.write(DataStructureNodeIO.GENERIC_VALUE_ID);
+        out.write(value.isTextual() ? 1 : 0);
         out.write(value.getRawData().length);
         out.write(value.getRawData());
     }

core/src/main/java/io/xpipe/core/data/generic/GenericDataStructureNodeReader.java

@@ -78,12 +78,12 @@ public class GenericDataStructureNodeReader implements GenericDataStreamCallback
     }
 
     @Override
-    public void onValue(byte[] value) {
+    public void onValue(byte[] value, boolean textual) {
         if (hasReader()) {
-            reader.onValue(value);
+            reader.onValue(value, textual);
             return;
         }
 
-        node = ValueNode.mutable(value);
+        node = ValueNode.mutable(value, textual);
     }
 }

core/src/main/java/io/xpipe/core/data/generic/GenericTupleReader.java

@@ -139,9 +139,9 @@ public class GenericTupleReader implements GenericAbstractReader {
     }
 
     @Override
-    public void onValue(byte[] value) {
+    public void onValue(byte[] value, boolean textual) {
         if (currentReader != null) {
-            currentReader.onValue(value);
+            currentReader.onValue(value, textual);
             return;
         }
 
@@ -153,7 +153,7 @@ public class GenericTupleReader implements GenericAbstractReader {
             throw new IllegalStateException("Tuple is full but got another value");
         }
 
-        putNode(ValueNode.mutable(value));
+        putNode(ValueNode.mutable(value, textual));
     }
 
     @Override

core/src/main/java/io/xpipe/core/data/node/DataStructureNode.java

@@ -51,12 +51,20 @@ public abstract class DataStructureNode implements Iterable<DataStructureNode> {
         throw unsupported("clear");
     }
 
-    public DataStructureNode setRawData(byte[] data) {
+    public boolean isTextual() {
+        throw unsupported("textual check");
+    }
+
+    public DataStructureNode setRaw(byte[] data) {
         throw unsupported("set raw data");
     }
 
-    public DataStructureNode setNull() {
-        throw unsupported("set null");
+    public DataStructureNode set(Object newValue) {
+        throw unsupported("set");
+    }
+
+    public DataStructureNode set(Object newValue, boolean textual) {
+        throw unsupported("set");
     }
 
     public DataStructureNode set(int index, DataStructureNode node) {

core/src/main/java/io/xpipe/core/data/node/ImmutableValueNode.java

@@ -3,14 +3,16 @@ package io.xpipe.core.data.node;
 public class ImmutableValueNode extends ValueNode {
 
     private final byte[] data;
+    private final boolean textual;
 
-    ImmutableValueNode(byte[] data) {
+    ImmutableValueNode(byte[] data, boolean textual) {
         this.data = data;
+        this.textual = textual;
     }
 
     @Override
     public String toString(int indent) {
-        return new String(data) + "(I)";
+        return (textual ? "\"" : "") + new String(data) + (textual ? "\"" : "") + " (I)";
     }
 
     @Override
@@ -25,11 +27,26 @@ public class ImmutableValueNode extends ValueNode {
 
     @Override
     public ValueNode mutableCopy() {
-        return ValueNode.mutable(data);
+        return ValueNode.mutable(data, textual);
     }
 
     @Override
-    public DataStructureNode setRawData(byte[] data) {
+    public boolean isTextual() {
+        return textual;
+    }
+
+    @Override
+    public DataStructureNode setRaw(byte[] data) {
+        throw new UnsupportedOperationException("Value node is immutable");
+    }
+
+    @Override
+    public DataStructureNode set(Object newValue) {
+        throw new UnsupportedOperationException("Value node is immutable");
+    }
+
+    @Override
+    public DataStructureNode set(Object newValue, boolean textual) {
         throw new UnsupportedOperationException("Value node is immutable");
     }
 

core/src/main/java/io/xpipe/core/data/node/MutableValueNode.java

@@ -1,16 +1,27 @@
 package io.xpipe.core.data.node;
 
+import java.nio.charset.StandardCharsets;
+
 public class MutableValueNode extends ValueNode {
 
-    private byte[] data;
+    static final MutableValueNode NULL = new MutableValueNode(null, false);
 
-    MutableValueNode(byte[] data) {
+    private byte[] data;
+    private boolean textual;
+
+    MutableValueNode(byte[] data, boolean textual) {
         this.data = data;
+        this.textual = textual;
     }
 
     @Override
     public String toString(int indent) {
-        return new String(data) + "(M)";
+        return (textual ? "\"" : "") + new String(data) + (textual ? "\"" : "") +  " (M)";
+    }
+
+    @Override
+    public boolean isTextual() {
+        return textual;
     }
 
     @Override
@@ -20,20 +31,49 @@ public class MutableValueNode extends ValueNode {
 
     @Override
     public ValueNode immutableView() {
-        return new ImmutableValueNode(data);
+        return new ImmutableValueNode(data, textual);
     }
 
     @Override
     public ValueNode mutableCopy() {
-        return new MutableValueNode(data);
+        return new MutableValueNode(data, textual);
     }
 
     @Override
-    public DataStructureNode setRawData(byte[] data) {
+    public DataStructureNode setRaw(byte[] data) {
         this.data = data;
         return this;
     }
 
+    @Override
+    public DataStructureNode set(Object newValue) {
+        if (newValue == null) {
+            this.data = null;
+            this.textual = false;
+        } else {
+            setRaw(newValue.toString().getBytes(StandardCharsets.UTF_8));
+        }
+
+        return this;
+    }
+
+    @Override
+    public DataStructureNode set(Object newValue, boolean textual) {
+        if (newValue == null && textual) {
+            throw new IllegalArgumentException("Can't set a textual null");
+        }
+
+        if (newValue == null) {
+            this.data = null;
+            this.textual = false;
+        } else {
+            setRaw(newValue.toString().getBytes(StandardCharsets.UTF_8));
+            this.textual = textual;
+        }
+
+        return this;
+    }
+
     public byte[] getRawData() {
         return data;
     }

core/src/main/java/io/xpipe/core/data/node/ValueNode.java

@@ -8,8 +8,6 @@ import java.util.Arrays;
 
 public abstract class ValueNode extends DataStructureNode {
 
-    private static final byte[] NULL = new byte[]{0};
-
     protected ValueNode() {
     }
 
@@ -31,40 +29,57 @@ public abstract class ValueNode extends DataStructureNode {
     @Override
     public abstract ValueNode mutableCopy();
 
-    public static ValueNode immutable(byte[] data) {
-        return new ImmutableValueNode(data);
+    public static ValueNode immutable(byte[] data, boolean textual) {
+        return new ImmutableValueNode(data, textual);
     }
 
-    public static ValueNode immutable(Object o) {
-        return immutable(o.toString().getBytes(StandardCharsets.UTF_8));
+    public static ValueNode immutable(Object o, boolean textual) {
+        return immutable(o.toString().getBytes(StandardCharsets.UTF_8), textual);
+    }
+
+    public static ValueNode immutableNull() {
+        return MutableValueNode.NULL.immutableView();
     }
 
     public static ValueNode mutableNull() {
-        return mutable(NULL);
+        return MutableValueNode.NULL.mutableCopy();
     }
 
-    public static ValueNode nullValue() {
-        return mutable(NULL);
+    public static ValueNode mutable(byte[] data, boolean textual) {
+        return new MutableValueNode(data, textual);
     }
 
-    public static ValueNode mutable(byte[] data) {
-        return new MutableValueNode(data);
-    }
-
-    public static ValueNode mutable(Object o) {
-        return mutable(o.toString().getBytes(StandardCharsets.UTF_8));
+    public static ValueNode mutable(Object o, boolean textual) {
+        return mutable(o.toString().getBytes(StandardCharsets.UTF_8), textual);
     }
 
     public static ValueNode of(byte[] data) {
-        return mutable(data);
+        return mutable(data, false);
     }
 
     public static ValueNode of(Object o) {
-        return mutable(o);
+        return mutable(o, false);
+    }
+
+    public static ValueNode ofText(byte[] data) {
+        return mutable(data, true);
+    }
+
+    public static ValueNode ofText(Object o) {
+        return mutable(o, true);
     }
 
     @Override
-    public abstract DataStructureNode setRawData(byte[] data);
+    public abstract boolean isTextual();
+
+    @Override
+    public abstract DataStructureNode setRaw(byte[] data);
+
+    @Override
+    public abstract DataStructureNode set(Object newValue);
+
+    @Override
+    public abstract DataStructureNode set(Object newValue, boolean textual);
 
     @Override
     public final int asInt() {

core/src/main/java/io/xpipe/core/data/typed/TypedDataStreamCallback.java

@@ -5,7 +5,7 @@ import io.xpipe.core.data.type.TupleType;
 
 public interface TypedDataStreamCallback {
 
-    default void onValue(byte[] data) {
+    default void onValue(byte[] data, boolean textual) {
     }
 
     default void onGenericNode(DataStructureNode node) {

core/src/main/java/io/xpipe/core/data/typed/TypedDataStreamParser.java

@@ -129,8 +129,9 @@ public class TypedDataStreamParser {
     }
 
     private void parseValue(InputStream in, TypedDataStreamCallback cb) throws IOException {
+        var textual = in.read() != 0;
         var size = in.read();
         var data = in.readNBytes(size);
-        cb.onValue(data);
+        cb.onValue(data, textual);
     }
 }

core/src/main/java/io/xpipe/core/data/typed/TypedDataStreamWriter.java

@@ -32,6 +32,7 @@ public class TypedDataStreamWriter {
 
     private static void writeValue(OutputStream out, ValueNode n) throws IOException {
         out.write(DataStructureNodeIO.TYPED_VALUE_ID);
+        out.write(n.isTextual() ? 1 : 0);
         out.write(n.getRawData().length);
         out.write(n.getRawData());
     }

core/src/main/java/io/xpipe/core/data/typed/TypedDataStructureNodeReader.java

@@ -81,12 +81,12 @@ public class TypedDataStructureNodeReader implements TypedAbstractReader {
     }
 
     @Override
-    public void onValue(byte[] data) {
+    public void onValue(byte[] data, boolean textual) {
         if (!expectedType.isValue()) {
             throw new IllegalStateException("Expected " + expectedType.getName() + " but got value");
         }
 
-        var val = makeImmutable ? ValueNode.immutable(data) : ValueNode.mutable(data);
+        var val = makeImmutable ? ValueNode.immutable(data, textual) : ValueNode.mutable(data, textual);
         finishNode(val);
         moveExpectedType(false);
     }

core/src/main/java/io/xpipe/core/data/typed/TypedReusableDataStructureNodeReader.java

@@ -47,16 +47,16 @@ public class TypedReusableDataStructureNodeReader implements TypedAbstractReader
     }
 
     @Override
-    public void onValue(byte[] data) {
+    public void onValue(byte[] data, boolean textual) {
         if (!initialized()) {
-            initialReader.onValue(data);
+            initialReader.onValue(data, textual);
             return;
         }
 
         if (isInArray()) {
-            getCurrentParent().set(indices.peek(), ValueNode.mutable(data));
+            getCurrentParent().set(indices.peek(), ValueNode.mutable(data, textual));
         } else {
-            getCurrent().setRawData(data);
+            getCurrent().setRaw(data);
         }
 
         if (!indices.isEmpty()) {

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

@@ -1,9 +1,10 @@
 package io.xpipe.core.source;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import io.xpipe.core.config.ConfigOptionSet;
+import io.xpipe.core.config.ConfigOptionSetInstance;
 import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Value;
-import lombok.extern.jackson.Jacksonized;
 
 import java.util.Map;
 
@@ -12,13 +13,11 @@ import java.util.Map;
  * This configuration can either be in progress or complete.
  */
 @Value
-@Builder
-@Jacksonized
-@AllArgsConstructor
+@AllArgsConstructor(onConstructor_={@JsonCreator})
 public class DataSourceConfigInstance {
 
     public static DataSourceConfigInstance xpbt() {
-        return new DataSourceConfigInstance("xpbt", DataSourceConfigOptions.empty(), Map.of());
+        return new DataSourceConfigInstance("xpbt", ConfigOptionSet.empty(), Map.of());
     }
 
     /**
@@ -29,7 +28,7 @@ public class DataSourceConfigInstance {
     /**
      * The available configuration options.
      */
-    DataSourceConfigOptions configOptions;
+    ConfigOptionSet configOptions;
 
     /**
      * The current configuration options that are set.
@@ -39,4 +38,10 @@ public class DataSourceConfigInstance {
     public boolean isComplete() {
         return currentValues.size() == configOptions.getOptions().size();
     }
+
+    public DataSourceConfigInstance(String provider, ConfigOptionSetInstance cInstance) {
+        this.provider = provider;
+        this.configOptions = cInstance.getConfigOptions();
+        this.currentValues = cInstance.getCurrentValues();
+    }
 }

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

@@ -1,32 +0,0 @@
-package io.xpipe.core.source;
-
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Singular;
-import lombok.Value;
-import lombok.extern.jackson.Jacksonized;
-
-import java.util.List;
-
-@Value
-@Builder
-@Jacksonized
-public class DataSourceConfigOptions {
-
-    public static DataSourceConfigOptions empty() {
-        return new DataSourceConfigOptions(List.of());
-    }
-
-    @Singular
-    List<Option> options;
-
-    @Value
-    @Builder
-    @Jacksonized
-    @AllArgsConstructor
-    public static class Option {
-        String name;
-        String key;
-    }
-}

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

@@ -31,4 +31,8 @@ public interface DataSourceDescriptor<DS extends DataStore> {
      * 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

@@ -67,11 +67,14 @@ public abstract class DataSourceInfo {
     @Value
     @JsonTypeName("text")
     public static class Text extends DataSourceInfo {
-        Charset encoding;
+        Charset charset;
+
+        int lineCount;
 
         @JsonCreator
-        public Text(Charset encoding) {
-            this.encoding = encoding;
+        public Text(Charset charset, int lineCount) {
+            this.charset = charset;
+            this.lineCount = lineCount;
         }
 
         @Override

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

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

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

@@ -2,10 +2,17 @@ package io.xpipe.core.source;
 
 import io.xpipe.core.data.node.DataStructureNode;
 
-public interface StructureReadConnection extends DataSourceConnection {
+public interface StructureReadConnection extends DataSourceReadConnection {
 
     /**
      * Reads the complete contents.
      */
     DataStructureNode read() throws Exception;
+
+    default void forward(DataSourceConnection con) throws Exception {
+        try (var tCon = (StructureWriteConnection) con) {
+            tCon.init();
+            tCon.write(read());
+        }
+    }
 }

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

@@ -13,7 +13,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 /**
  * A connection for sequentially reading the data of a table data source.
  */
-public interface TableReadConnection extends DataSourceConnection {
+public interface TableReadConnection extends DataSourceReadConnection {
 
     /**
      * Returns the data type of the table data.
@@ -39,6 +39,10 @@ public interface TableReadConnection extends DataSourceConnection {
      * Writes the rows to an OutputStream in the X-Pipe binary format.
      */
     default void forwardRows(OutputStream out, int maxLines) throws Exception {
+        if (maxLines == 0) {
+            return;
+        }
+
         var dataType = getDataType();
         AtomicInteger rowCounter = new AtomicInteger();
         withRows(l -> {
@@ -47,4 +51,11 @@ public interface TableReadConnection extends DataSourceConnection {
             return rowCounter.get() != maxLines;
         });
     }
+
+    default void forward(DataSourceConnection con) throws Exception {
+        try (var tCon = (TableWriteConnection) con) {
+            tCon.init();
+            withRows(tCon.writeLinesAcceptor());
+        }
+    }
 }

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

@@ -2,15 +2,28 @@ package io.xpipe.core.source;
 
 import io.xpipe.core.store.DataStore;
 
-public class TextDataSourceDescriptor<DS extends DataStore> implements DataSourceDescriptor<DS>  {
-
-    @Override
-    public DataSourceInfo determineInfo(DS store) throws Exception {
-        return null;
-    }
+public abstract class TextDataSourceDescriptor<DS extends DataStore> implements DataSourceDescriptor<DS>  {
 
     @Override
     public DataSourceType getType() {
         return DataSourceType.TEXT;
     }
+
+    @Override
+    public TextReadConnection openReadConnection(DS store) throws Exception {
+        var con = newReadConnection(store);
+        con.init();
+        return con;
+    }
+
+    @Override
+    public TextWriteConnection openWriteConnection(DS store) throws Exception {
+        var con = newWriteConnection(store);
+        con.init();
+        return con;
+    }
+
+    protected abstract TextWriteConnection newWriteConnection(DS store);
+
+    protected abstract TextReadConnection newReadConnection(DS store);
 }

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

@@ -1,8 +1,29 @@
 package io.xpipe.core.source;
 
-import java.nio.charset.Charset;
+import java.util.List;
+import java.util.stream.Stream;
 
-public interface TextReadConnection extends DataSourceConnection {
+public interface TextReadConnection extends DataSourceReadConnection {
 
-    Charset getEncoding();
+    /**
+     * 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 forward(DataSourceConnection con) throws Exception {
+        try (var tCon = (TextWriteConnection) con) {
+            tCon.init();
+            for (var it = lines().iterator(); it.hasNext(); ) {
+                tCon.writeLine(it.next());
+            }
+        }
+    }
 }

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

@@ -1,8 +1,6 @@
 package io.xpipe.core.source;
 
-import java.io.OutputStream;
-
 public interface TextWriteConnection extends DataSourceConnection {
 
-    OutputStream getOutputStream();
+    void writeLine(String line) throws Exception;
 }

core/src/main/java/io/xpipe/core/store/FileDataStore.java

@@ -0,0 +1,6 @@
+package io.xpipe.core.store;
+
+public interface FileDataStore extends StreamDataStore {
+
+    String getFileName();
+}

core/src/main/java/io/xpipe/core/store/LocalFileDataStore.java

@@ -15,7 +15,7 @@ import java.util.Optional;
 
 @JsonTypeName("local")
 @EqualsAndHashCode
-public class LocalFileDataStore implements StreamDataStore {
+public class LocalFileDataStore implements FileDataStore {
 
     private final Path file;
 
@@ -57,4 +57,9 @@ public class LocalFileDataStore implements StreamDataStore {
     public boolean exists() {
         return Files.exists(file);
     }
+
+    @Override
+    public String getFileName() {
+        return file.getFileName().toString();
+    }
 }

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

@@ -33,7 +33,10 @@ public class CoreJacksonModule extends SimpleModule {
                 new NamedType(TupleType.class),
                 new NamedType(ArrayType.class),
                 new NamedType(WildcardType.class),
-                new NamedType(DataSourceInfo.Table.class)
+                new NamedType(DataSourceInfo.Table.class),
+                new NamedType(DataSourceInfo.Structure.class),
+                new NamedType(DataSourceInfo.Text.class),
+                new NamedType(DataSourceInfo.Raw.class)
         );
 
         addSerializer(Charset.class, new CharsetSerializer());

core/src/main/java/module-info.java

@@ -16,6 +16,8 @@ module io.xpipe.core {
     opens io.xpipe.core.util;
     opens io.xpipe.core.data.node;
     opens io.xpipe.core.data.typed;
+    exports io.xpipe.core.config;
+    opens io.xpipe.core.config;
 
     requires com.fasterxml.jackson.core;
     requires com.fasterxml.jackson.databind;