Refactor

core/src/main/java/io/xpipe/core/impl/FileStore.java

@@ -59,7 +59,7 @@ public class FileStore extends JacksonizedValue implements FilenameStore, Stream
     @Override
     public void checkComplete() throws Exception {
         if (fileSystem == null) {
-            throw new IllegalStateException("Machine is missing");
+            throw new IllegalStateException("File system is missing");
         }
         if (file == null) {
             throw new IllegalStateException("File is missing");

core/src/main/java/io/xpipe/core/impl/PeekTableStreamReadConnection.java

@@ -0,0 +1,66 @@
+package io.xpipe.core.impl;
+
+import io.xpipe.core.charsetter.StreamCharset;
+import io.xpipe.core.data.node.DataStructureNodeAcceptor;
+import io.xpipe.core.data.node.TupleNode;
+import io.xpipe.core.data.type.TupleType;
+import io.xpipe.core.source.TableReadConnection;
+import io.xpipe.core.store.StreamDataStore;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+public abstract class PeekTableStreamReadConnection extends StreamReadConnection implements TableReadConnection {
+
+    private TupleNode first;
+    private TupleType type;
+
+    public PeekTableStreamReadConnection(StreamDataStore store, StreamCharset charset) {
+        super(store, charset);
+    }
+
+    @Override
+    public void init() throws Exception {
+        super.init();
+        AtomicReference<TupleNode> read = new AtomicReference<>();
+        withRowsInternal(node -> {
+            read.set(node);
+            return false;
+        });
+        if (read.get() == null) {
+            return;
+        }
+
+        first = read.get().asTuple();
+        type = convertType(first);
+    }
+
+    protected TupleType convertType(TupleNode n) {
+        return n.determineDataType().asTuple();
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (inputStream == null) {
+            throw new IllegalStateException("Not initialized");
+        }
+
+        inputStream.close();
+    }
+
+    @Override
+    public TupleType getDataType() {
+        return type;
+    }
+
+    @Override
+    public final void withRows(DataStructureNodeAcceptor<TupleNode> lineAcceptor) throws Exception {
+        if (first != null) {
+            lineAcceptor.accept(first);
+            first = null;
+        }
+
+        withRowsInternal(lineAcceptor);
+    }
+
+    protected abstract void withRowsInternal(DataStructureNodeAcceptor<TupleNode> lineAcceptor) throws Exception;
+}

core/src/main/java/io/xpipe/core/impl/StreamReadConnection.java

@@ -1,7 +1,8 @@
-package io.xpipe.core.source;
+package io.xpipe.core.impl;
 
 import io.xpipe.core.charsetter.Charsetter;
 import io.xpipe.core.charsetter.StreamCharset;
+import io.xpipe.core.source.DataSourceReadConnection;
 import io.xpipe.core.store.StreamDataStore;
 
 import java.io.InputStream;

core/src/main/java/io/xpipe/core/impl/StreamWriteConnection.java

@@ -1,6 +1,7 @@
-package io.xpipe.core.source;
+package io.xpipe.core.impl;
 
 import io.xpipe.core.charsetter.StreamCharset;
+import io.xpipe.core.source.DataSourceConnection;
 import io.xpipe.core.store.StreamDataStore;
 
 import java.io.OutputStream;

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

@@ -1,7 +1,5 @@
 package io.xpipe.core.impl;
 
-import io.xpipe.core.source.StreamReadConnection;
-
 import java.io.BufferedReader;
 import java.util.stream.Stream;
 

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

@@ -1,7 +1,5 @@
 package io.xpipe.core.impl;
 
-import io.xpipe.core.source.StreamWriteConnection;
-
 public class TextWriteConnection extends StreamWriteConnection implements io.xpipe.core.source.TextWriteConnection {
 
     private final TextSource source;

core/src/main/java/io/xpipe/core/impl/XpbsReadConnection.java

@@ -2,7 +2,6 @@ package io.xpipe.core.impl;
 
 import io.xpipe.core.data.generic.GenericDataStreamParser;
 import io.xpipe.core.data.node.DataStructureNode;
-import io.xpipe.core.source.StreamReadConnection;
 import io.xpipe.core.source.StructureReadConnection;
 
 public class XpbsReadConnection extends StreamReadConnection implements StructureReadConnection {

core/src/main/java/io/xpipe/core/impl/XpbsWriteConnection.java

@@ -2,7 +2,6 @@ package io.xpipe.core.impl;
 
 import io.xpipe.core.data.generic.GenericDataStreamWriter;
 import io.xpipe.core.data.node.DataStructureNode;
-import io.xpipe.core.source.StreamWriteConnection;
 import io.xpipe.core.source.StructureWriteConnection;
 
 public class XpbsWriteConnection extends StreamWriteConnection implements StructureWriteConnection {

core/src/main/java/io/xpipe/core/impl/XpbtReadConnection.java

@@ -7,7 +7,6 @@ import io.xpipe.core.data.node.TupleNode;
 import io.xpipe.core.data.type.TupleType;
 import io.xpipe.core.data.typed.TypedDataStreamParser;
 import io.xpipe.core.data.typed.TypedDataStructureNodeReader;
-import io.xpipe.core.source.StreamReadConnection;
 import io.xpipe.core.source.TableReadConnection;
 import io.xpipe.core.store.StreamDataStore;
 import io.xpipe.core.util.JacksonMapper;

core/src/main/java/io/xpipe/core/impl/XpbtWriteConnection.java

@@ -7,7 +7,6 @@ import io.xpipe.core.data.node.DataStructureNodeAcceptor;
 import io.xpipe.core.data.node.TupleNode;
 import io.xpipe.core.data.type.TupleType;
 import io.xpipe.core.data.typed.TypedDataStreamWriter;
-import io.xpipe.core.source.StreamWriteConnection;
 import io.xpipe.core.source.TableMapping;
 import io.xpipe.core.util.JacksonMapper;
 import lombok.Getter;