Updated FFmpeg RunProcess and RunProcessAsyncFunctions

FFMpegCore/FFMPEG/FFMpeg.cs

@@ -15,6 +15,8 @@
 using System.Text.RegularExpressions;
 using System.Threading.Tasks;
 using Instances;
+using System.Runtime.CompilerServices;
+using System.Threading;
 
 namespace FFMpegCore.FFMPEG
 {
@@ -391,6 +393,9 @@ public VideoInfo Convert(ArgumentContainer arguments, bool skipExistsCheck = fal
                 throw new FFMpegException(FFMpegExceptionType.Conversion, "Could not process file without error");
 
             _totalTime = TimeSpan.MinValue;
+
+            if (output == null)
+                return null;
             return new VideoInfo(output);
         }
         public async Task<VideoInfo> ConvertAsync(ArgumentContainer arguments, bool skipExistsCheck = false)
@@ -403,12 +408,21 @@ public async Task<VideoInfo> ConvertAsync(ArgumentContainer arguments, bool skip
                 throw new FFMpegException(FFMpegExceptionType.Conversion, "Could not process file without error");
 
             _totalTime = TimeSpan.MinValue;
+            if (output == null)
+                return null;
             return new VideoInfo(output);
         }
 
         private static (VideoInfo[] Input, FileInfo Output) GetInputOutput(ArgumentContainer arguments)
         {
-            var output = ((OutputArgument)arguments[typeof(OutputArgument)]).GetAsFileInfo();
+            FileInfo output;
+            if (arguments.TryGetArgument<OutputArgument>(out var outputArg))
+                output = outputArg.GetAsFileInfo();
+            else if (arguments.TryGetArgument<OutputPipeArgument>(out var outputPipeArg))
+                output = null;
+            else
+                throw new FFMpegException(FFMpegExceptionType.Operation, "No output argument found");
+
             VideoInfo[] sources;
             if (arguments.TryGetArgument<InputArgument>(out var input))
                 sources = input.GetAsVideoInfo();
@@ -452,24 +466,53 @@ private bool RunProcess(ArgumentContainer container, FileInfo output, bool skipE
             {
                 inputPipeArgument.OpenPipe();
             }
-
+            if (container.TryGetArgument<OutputPipeArgument>(out var outputPipeArgument))
-            try
             {
+                outputPipeArgument.OpenPipe();
+            }
+
+
             _instance = new Instance(_ffmpegPath, arguments);
             _instance.DataReceived += OutputData;
 
-                if (inputPipeArgument != null)
+            if (inputPipeArgument != null || outputPipeArgument != null)
             {
                 try
                 {
-                        var task = _instance.FinishedRunning();
+                    using (var tokenSource = new CancellationTokenSource())
-                        inputPipeArgument.FlushPipe();
+                    {
+                        var concurrentTasks = new List<Task>();
+                        concurrentTasks.Add(_instance.FinishedRunning()
+                            .ContinueWith((t =>
+                            {
+                                exitCode = t.Result;
+                                if (exitCode != 0)
+                                    tokenSource.Cancel();
+                            })));
+                        if (inputPipeArgument != null)
+                            concurrentTasks.Add(inputPipeArgument.ProcessDataAsync(tokenSource.Token)
+                                .ContinueWith((t) =>
+                                {
                                     inputPipeArgument.ClosePipe();
-                        task.Wait();
+                                    if (t.Exception != null)
-                        exitCode = task.Result;
+                                        throw t.Exception;
+                                }));
+                        if (outputPipeArgument != null)
+                            concurrentTasks.Add(outputPipeArgument.ProcessDataAsync(tokenSource.Token)
+                                .ContinueWith((t) =>
+                                {
+                                    outputPipeArgument.ClosePipe();
+                                    if (t.Exception != null)
+                                        throw t.Exception;
+                                }));
+
+                        Task.WaitAll(concurrentTasks.ToArray()/*, tokenSource.Token*/);
+                    }
                 }
                 catch (Exception ex)
                 {
+                    inputPipeArgument?.ClosePipe();
+                    outputPipeArgument?.ClosePipe();
                     throw new FFMpegException(FFMpegExceptionType.Process, string.Join("\n", _instance.ErrorData), ex);
                 }
             }
@@ -478,59 +521,87 @@ private bool RunProcess(ArgumentContainer container, FileInfo output, bool skipE
                 exitCode = _instance.BlockUntilFinished();
             }
 
-                if (!skipExistsCheck && (!File.Exists(output.FullName) || new FileInfo(output.FullName).Length == 0))
+            if(exitCode != 0)
+                throw new FFMpegException(FFMpegExceptionType.Process, string.Join("\n", _instance.ErrorData));
+
+            if (outputPipeArgument == null && !skipExistsCheck && (!File.Exists(output.FullName) || new FileInfo(output.FullName).Length == 0))
                 throw new FFMpegException(FFMpegExceptionType.Process, string.Join("\n", _instance.ErrorData));
 
             return exitCode == 0;
         }
-            finally
-            {
-                if (inputPipeArgument != null)
-                    inputPipeArgument.ClosePipe();
-            }
-        }
         private async Task<bool> RunProcessAsync(ArgumentContainer container, FileInfo output, bool skipExistsCheck)
         {
             _instance?.Dispose();
             var arguments = ArgumentBuilder.BuildArguments(container);
+            var exitCode = -1;
 
-            int exitCode = -1;
             if (container.TryGetArgument<InputPipeArgument>(out var inputPipeArgument))
             {
                 inputPipeArgument.OpenPipe();
             }
-            try
+            if (container.TryGetArgument<OutputPipeArgument>(out var outputPipeArgument))
             {
+                outputPipeArgument.OpenPipe();
+            }
+
 
             _instance = new Instance(_ffmpegPath, arguments);
             _instance.DataReceived += OutputData;
 
-                if (inputPipeArgument != null)
+            if (inputPipeArgument != null || outputPipeArgument != null)
+            {
+                try
+                {
+                    using (var tokenSource = new CancellationTokenSource())
+                    {
+                        var concurrentTasks = new List<Task>();
+                        concurrentTasks.Add(_instance.FinishedRunning()
+                            .ContinueWith((t =>
+                            {
+                                exitCode = t.Result;
+                                if (exitCode != 0)
+                                    tokenSource.Cancel();
+                            })));
+                        if (inputPipeArgument != null)
+                            concurrentTasks.Add(inputPipeArgument.ProcessDataAsync(tokenSource.Token)
+                                .ContinueWith((t) =>
                                 {
-                    var task = _instance.FinishedRunning();
-                    inputPipeArgument.FlushPipe();
                                     inputPipeArgument.ClosePipe();
+                                    if (t.Exception != null)
+                                        throw t.Exception;
+                                }));
+                        if (outputPipeArgument != null)
+                            concurrentTasks.Add(outputPipeArgument.ProcessDataAsync(tokenSource.Token)
+                                .ContinueWith((t) =>
+                                {
+                                    outputPipeArgument.ClosePipe();
+                                    if (t.Exception != null)
+                                        throw t.Exception;
+                                }));
 
-                    exitCode = await task;
+                        await Task.WhenAll(concurrentTasks);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    inputPipeArgument?.ClosePipe();
+                    outputPipeArgument?.ClosePipe();
+                    throw new FFMpegException(FFMpegExceptionType.Process, string.Join("\n", _instance.ErrorData), ex);
+                }
             }
             else
             {
                 exitCode = await _instance.FinishedRunning();
             }
 
-                if (!skipExistsCheck && (!File.Exists(output.FullName) || new FileInfo(output.FullName).Length == 0))
+            if (exitCode != 0)
+                throw new FFMpegException(FFMpegExceptionType.Process, string.Join("\n", _instance.ErrorData));
+
+            if (outputPipeArgument == null && !skipExistsCheck && (!File.Exists(output.FullName) || new FileInfo(output.FullName).Length == 0))
                 throw new FFMpegException(FFMpegExceptionType.Process, string.Join("\n", _instance.ErrorData));
 
             return exitCode == 0;
         }
-            finally
-            {
-                if (inputPipeArgument != null)
-                {
-                    inputPipeArgument.ClosePipe();
-                }
-            }
-        }
 
         private void Cleanup(IEnumerable<string> pathList)
         {