IOneStreamProvider, IOneSinkProvider and implementations

EtLast.EPPlus/Mutators/EpPlusSimpleRowWriterMutator.cs

@@ -3,7 +3,7 @@
 public sealed class EpPlusSimpleRowWriterMutator : AbstractMutator, IRowSink
 {
     [ProcessParameterMustHaveValue]
-    public required ISinkProvider SinkProvider { get; init; }
+    public required IPartitionedSinkProvider SinkProvider { get; init; }
 
     [ProcessParameterMustHaveValue]
     public required string SheetName { get; init; }

EtLast.EPPlus/Readers/EpPlusExcelReader.cs

@@ -3,7 +3,7 @@
 public sealed class EpPlusExcelReader : AbstractEpPlusExcelReader
 {
     [ProcessParameterMustHaveValue]
-    public required IStreamProvider StreamProvider { get; init; }
+    public required IManyStreamProvider StreamProvider { get; init; }
 
     /// <summary>
     /// First stream index is (integer) 0

EtLast.EPPlus/Readers/EpPlusExcelSheetListReader.cs

@@ -3,7 +3,7 @@
 public sealed class EpPlusExcelSheetListReader : AbstractRowSource
 {
     [ProcessParameterMustHaveValue]
-    public required IStreamProvider StreamProvider { get; init; }
+    public required IManyStreamProvider StreamProvider { get; init; }
 
     /// <summary>
     /// Default value is "Stream".

EtLast.LocalFiles/Processes/SerializeToJsonFileJobFluent.cs

@@ -11,7 +11,7 @@ public static IFlow SerializeToLocalJsonFile<T>(this IFlow builder, string targe
             Encoding = customEncoding ?? Encoding.UTF8,
             SinkProvider = new LocalFileSinkProvider()
             {
-                FileNameGenerator = _ => targetFileName,
+                FileName = targetFileName,
                 ActionWhenFileExists = LocalSinkFileExistsAction.DeleteAndContinue,
                 FileMode = FileMode.CreateNew,
             },

EtLast.LocalFiles/Streams/LocalFileSinkProvider.cs

@@ -1,15 +1,13 @@
 namespace FizzCode.EtLast;
 
-public enum LocalSinkFileExistsAction { Continue, Exception, DeleteAndContinue }
-
 [ContainsProcessParameterValidation]
-public class LocalFileSinkProvider : ISinkProvider
+public class LocalFileSinkProvider : IOneSinkProvider
 {
     /// <summary>
-    /// Generates file name based on a partition key.
+    /// Generates file name.
     /// </summary>
     [ProcessParameterMustHaveValue]
-    public required Func<string, string> FileNameGenerator { get; init; }
+    public required string FileName { get; init; }
 
     /// <summary>
     /// Default value is <see cref="LocalSinkFileExistsAction.Exception"/>.
@@ -33,26 +31,24 @@ public class LocalFileSinkProvider : ISinkProvider
 
     public bool AutomaticallyDispose => true;
 
-    public NamedSink GetSink(IProcess caller, string partitionKey, string sinkFormat, string[] columns)
+    public NamedSink GetSink(IProcess caller, string sinkFormat, string[] columns)
     {
-        var fileName = FileNameGenerator.Invoke(partitionKey);
-
         var ioCommand = caller.Context.RegisterIoCommand(new IoCommand()
         {
             Process = caller,
             Kind = IoCommandKind.fileWrite,
-            Location = Path.GetDirectoryName(fileName),
-            Path = Path.GetFileName(fileName),
+            Location = Path.GetDirectoryName(FileName),
+            Path = Path.GetFileName(FileName),
             Message = "writing to local file",
         });
 
-        if (ActionWhenFileExists != LocalSinkFileExistsAction.Continue && File.Exists(fileName))
+        if (ActionWhenFileExists != LocalSinkFileExistsAction.Continue && File.Exists(FileName))
         {
             if (ActionWhenFileExists == LocalSinkFileExistsAction.Exception)
             {
-                var exception = new LocalFileWriteException(caller, "local file already exist", fileName);
+                var exception = new LocalFileWriteException(caller, "local file already exist", FileName);
                 exception.AddOpsMessage(string.Format(CultureInfo.InvariantCulture, "local file already exist: {0}",
-                    fileName));
+                    FileName));
 
                 ioCommand.AffectedDataCount = 0;
                 ioCommand.Failed(exception);
@@ -62,14 +58,14 @@ public NamedSink GetSink(IProcess caller, string partitionKey, string sinkFormat
             {
                 try
                 {
-                    File.Delete(fileName);
+                    File.Delete(FileName);
                 }
                 catch (Exception ex)
                 {
-                    var exception = new LocalFileWriteException(caller, "error while writing local file / file deletion failed", fileName, ex);
+                    var exception = new LocalFileWriteException(caller, "error while writing local file / file deletion failed", FileName, ex);
                     exception.AddOpsMessage(string.Format(CultureInfo.InvariantCulture, "error while writing local file: {0}, file deletion failed, message: {1}",
-                        fileName, ex.Message));
-                    exception.Data["FileName"] = fileName;
+                        FileName, ex.Message));
+                    exception.Data["FileName"] = FileName;
 
                     ioCommand.AffectedDataCount = 0;
                     ioCommand.Failed(exception);
@@ -78,7 +74,7 @@ public NamedSink GetSink(IProcess caller, string partitionKey, string sinkFormat
             }
         }
 
-        var directory = Path.GetDirectoryName(fileName);
+        var directory = Path.GetDirectoryName(FileName);
         if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory))
         {
             try
@@ -87,10 +83,10 @@ public NamedSink GetSink(IProcess caller, string partitionKey, string sinkFormat
             }
             catch (Exception ex)
             {
-                var exception = new LocalFileWriteException(caller, "error while writing local file / directory creation failed", fileName, ex);
+                var exception = new LocalFileWriteException(caller, "error while writing local file / directory creation failed", FileName, ex);
                 exception.AddOpsMessage(string.Format(CultureInfo.InvariantCulture, "error while writing local file: {0}, directory creation failed, message: {1}",
-                    fileName, ex.Message));
-                exception.Data["FileName"] = fileName;
+                    FileName, ex.Message));
+                exception.Data["FileName"] = FileName;
                 exception.Data["Directory"] = directory;
 
                 ioCommand.AffectedDataCount = 0;
@@ -101,17 +97,17 @@ public NamedSink GetSink(IProcess caller, string partitionKey, string sinkFormat
 
         try
         {
-            var sink = caller.Context.GetSink(Path.GetDirectoryName(fileName), Path.GetFileName(fileName), sinkFormat, caller, columns);
+            var sink = caller.Context.GetSink(Path.GetDirectoryName(FileName), Path.GetFileName(FileName), sinkFormat, caller, columns);
 
-            var stream = new FileStream(fileName, FileMode, FileAccess, FileShare);
-            return new NamedSink(fileName, stream, ioCommand, sink);
+            var stream = new FileStream(FileName, FileMode, FileAccess, FileShare);
+            return new NamedSink(FileName, stream, ioCommand, sink);
         }
         catch (Exception ex)
         {
-            var exception = new LocalFileWriteException(caller, "error while writing local file", fileName, ex);
+            var exception = new LocalFileWriteException(caller, "error while writing local file", FileName, ex);
             exception.AddOpsMessage(string.Format(CultureInfo.InvariantCulture, "error while writing local file: {0}, message: {1}",
-                fileName, ex.Message));
-            exception.Data["FileName"] = fileName;
+                FileName, ex.Message));
+            exception.Data["FileName"] = FileName;
 
             ioCommand.AffectedDataCount = 0;
             ioCommand.Failed(exception);

EtLast.LocalFiles/Streams/LocalFileStreamProvider.cs

@@ -1,7 +1,7 @@
 namespace FizzCode.EtLast;
 
 [ContainsProcessParameterValidation]
-public class LocalFileStreamProvider : IStreamProvider
+public class LocalFileStreamProvider : IManyStreamProvider, IOneStreamProvider
 {
     [ProcessParameterMustHaveValue]
     public required string FileName { get; init; }
@@ -33,7 +33,7 @@ public string GetTopic()
             : null;
     }
 
-    public IEnumerable<NamedStream> GetStreams(IProcess caller)
+    public NamedStream GetStream(IProcess caller)
     {
         var ioCommand = caller.Context.RegisterIoCommand(new IoCommand()
         {
@@ -59,16 +59,13 @@ public IEnumerable<NamedStream> GetStreams(IProcess caller)
 
             ioCommand.AffectedDataCount = 0;
             ioCommand.End();
-            return Enumerable.Empty<NamedStream>();
+            return null;
         }
 
         try
         {
             var stream = new FileStream(FileName, Options);
-            return new[]
-            {
-                new NamedStream(FileName, stream, ioCommand),
-            };
+            return new NamedStream(FileName, stream, ioCommand);
         }
         catch (Exception ex)
         {
@@ -82,4 +79,10 @@ public IEnumerable<NamedStream> GetStreams(IProcess caller)
             throw exception;
         }
     }
-}
+
+    public IEnumerable<NamedStream> GetStreams(IProcess caller)
+    {
+        var stream = GetStream(caller);
+        return new[] { stream };
+    }
+}