Suggestion: Support compressed files (Gzip, Brotli, zlib)

CosmosDbDataMigrationTool.sln

@@ -109,6 +109,10 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "PostgreSQL", "PostgreSQL",
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Cosmos.DataTransfer.MongoExtension", "Extensions\Mongo\Cosmos.DataTransfer.MongoExtension\Cosmos.DataTransfer.MongoExtension.csproj", "{31BC84E1-55E5-45AA-BFAC-90732F20588B}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Interfaces", "Interfaces", "{AC56D40C-6A65-42E5-8881-D126FD080774}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Cosmos.DataTransfer.Common.UnitTests", "Interfaces\Cosmos.DataTransfer.Common.UnitTests\Cosmos.DataTransfer.Common.UnitTests.csproj", "{8C755891-38F4-4155-9A60-51BC6841FA36}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -203,6 +207,10 @@ Global
 		{31BC84E1-55E5-45AA-BFAC-90732F20588B}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{31BC84E1-55E5-45AA-BFAC-90732F20588B}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{31BC84E1-55E5-45AA-BFAC-90732F20588B}.Release|Any CPU.Build.0 = Release|Any CPU
+		{8C755891-38F4-4155-9A60-51BC6841FA36}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{8C755891-38F4-4155-9A60-51BC6841FA36}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{8C755891-38F4-4155-9A60-51BC6841FA36}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{8C755891-38F4-4155-9A60-51BC6841FA36}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -237,6 +245,7 @@ Global
 		{85820167-DB94-458B-B09B-9E823996C692} = {1B927C5F-50FC-42A6-BAF6-B00E6D760543}
 		{1B927C5F-50FC-42A6-BAF6-B00E6D760543} = {A8A1CEAB-2D82-460C-9B86-74ABD17CD201}
 		{31BC84E1-55E5-45AA-BFAC-90732F20588B} = {F18E789A-D32D-48D3-B75F-1196D7215F74}
+		{8C755891-38F4-4155-9A60-51BC6841FA36} = {AC56D40C-6A65-42E5-8881-D126FD080774}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {662B3F27-70D8-45E6-A1C0-1438A9C8A542}

Directory.Packages.props

@@ -12,6 +12,7 @@
     <PackageVersion Include="Azure.Security.KeyVault.Keys" Version="4.7.0" />
     <PackageVersion Include="Azure.Storage.Blobs" Version="12.22.2" />
     <PackageVersion Include="coverlet.collector" Version="6.0.2" />
+    <PackageVersion Include="coverlet.msbuild" Version="2.8.0" />
     <PackageVersion Include="CsvHelper" Version="33.0.1" />
     <PackageVersion Include="Microsoft.Azure.Cosmos" Version="3.46.1" />
     <PackageVersion Include="Microsoft.Azure.Cosmos.Encryption" Version="2.0.4" />

Interfaces/Cosmos.DataTransfer.Common.UnitTests/Cosmos.DataTransfer.Common.UnitTests.csproj

@@ -0,0 +1,38 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+
+    <IsPackable>false</IsPackable>
+    <CollectCoverage>true</CollectCoverage>
+    <CoverletOutputFormat>cobertura</CoverletOutputFormat>
+    <CoverletOutput>../Cosmos.DataTransfer.Common/coverage.cobertura.xml</CoverletOutput>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" />
+    <PackageReference Include="MSTest.TestAdapter" />
+    <PackageReference Include="MSTest.TestFramework" />
+    <PackageReference Include="System.Linq.Async" />
+    <PackageReference Include="Moq" />
+    <PackageReference Include="coverlet.collector" />
+    <PackageReference Include="coverlet.msbuild">
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+      <PrivateAssets>all</PrivateAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Cosmos.DataTransfer.Common\Cosmos.DataTransfer.Common.csproj" />
+    <ProjectReference Include="..\..\Extensions\Json\Cosmos.DataTransfer.JsonExtension.UnitTests\Cosmos.DataTransfer.JsonExtension.UnitTests.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="Data\*">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+</Project>

Interfaces/Cosmos.DataTransfer.Common.UnitTests/Data/foobar.txt

@@ -0,0 +1 @@
+Hello world!

Interfaces/Cosmos.DataTransfer.Common.UnitTests/Data/foobar.txt.br

@@ -0,0 +1 @@
+‹€Hello world!

Interfaces/Cosmos.DataTransfer.Common.UnitTests/Data/rich.json

@@ -0,0 +1 @@
+[{"id":1,"name":"john"}]

Interfaces/Cosmos.DataTransfer.Common.UnitTests/FileDataSinkTests.cs

@@ -0,0 +1,100 @@
+using System.IO.Compression;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Configuration;
+using System.Text;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Cosmos.DataTransfer.JsonExtension;
+using Cosmos.DataTransfer.JsonExtension.UnitTests;
+using Cosmos.DataTransfer.Interfaces;
+using Moq;
+
+namespace Cosmos.DataTransfer.Common.UnitTests;
+
+[TestClass]
+public class FileSinkDataSinkTests {
+
+    public static IEnumerable<object[]> Test_WriteToTargetAsyncData { get {
+        yield return new object[] { CompressionEnum.None, "", "" };
+        yield return new object[] { CompressionEnum.None, ".txt", ".txt" };
+        yield return new object[] { CompressionEnum.Brotli, "", ".br" };
+        yield return new object[] { CompressionEnum.Brotli, ".br", ".br" };
+        yield return new object[] { CompressionEnum.Gzip, "", ".gz" };
+        yield return new object[] { CompressionEnum.Gzip, ".gz", ".gz" };
+        yield return new object[] { CompressionEnum.Gzip, ".gzip", ".gzip" };
+        yield return new object[] { CompressionEnum.Deflate, "", ".zz" };
+        yield return new object[] { CompressionEnum.Deflate, ".zz", ".zz" };
+    } }
+
+    [TestMethod]
+    [DynamicData(nameof(Test_WriteToTargetAsyncData))]
+    public async Task Test_WriteToTargetAsync(CompressionEnum compression, string suffix, string expected_ext) {
+        var source = new Mock<IDataSourceExtension>();
+        FileDataSink sink = new();
+        var filePath = Path.GetTempFileName();
+        var config = TestHelpers.CreateConfig(new Dictionary<string,string>() {
+            { "FilePath", filePath + suffix },
+            { "Compression", compression.ToString() }, 
+            { "Append", "false" }
+        });
+
+        var str = Encoding.UTF8.GetBytes("Hello world!");
+        await sink.WriteToTargetAsync(writer => writer.WriteAsync(str).AsTask(),
+            config, source.Object, NullLogger.Instance);
+
+        var dataSource = new FileDataSource();
+        var stream = dataSource.ReadFile(filePath + expected_ext, CompressionEnum.None, NullLogger.Instance)!;
+        var buffer = new byte[100];
+        await stream.ReadAsync(buffer.AsMemory(0, buffer.Length));
+        var result = Encoding.UTF8.GetString(buffer);
+        Assert.AreEqual("Hello world!", result.TrimEnd('\0'), $"compression: {compression}, suffix: {suffix}, expected extension: {expected_ext}.");
+    }
+
+    [TestMethod]
+    public void Test_GetSettings() {
+        var sink = new FileDataSink();
+        var response = sink.GetSettings().ToArray();
+        Assert.AreEqual(1, response.Length);
+        Assert.IsInstanceOfType(response[0], typeof(FileSinkSettings));
+    }
+
+    public static IEnumerable<object[]> Test_WriteToTargetAsyncAppendData { get {
+        yield return new object[] { "foobar.txt", CompressionEnum.None };
+        yield return new object[] { "foobar.txt.gz", CompressionEnum.Gzip };
+        yield return new object[] { "foobar.txt.br", CompressionEnum.Brotli };
+        yield return new object[] { "foobar.txt.zz", CompressionEnum.Deflate };
+    } }
+
+    [TestMethod]
+    [DynamicData(nameof(Test_WriteToTargetAsyncAppendData))]
+    public async Task Test_WriteToTargetAsyncAppend(string filePath, CompressionEnum compression) {
+        var source = new Mock<IDataSourceExtension>();
+        FileDataSink sink = new();
+        var tmpdir = FileSinkDataSourceTests.GetTemporaryDirectory();
+        var destfile = Path.Combine(tmpdir, filePath);
+        File.Copy(Path.Combine("Data", filePath), destfile);
+        var config = TestHelpers.CreateConfig(new Dictionary<string,string>() {
+            { "FilePath", destfile },
+            { "Compression", compression.ToString() }, 
+            { "Append", "true" }
+        });
+
+        if (compression != CompressionEnum.None) {
+            var e = Assert.ThrowsException<AggregateException>(() => {
+                var settings = config.Get<FileSinkSettings>();
+                settings.Validate();
+            });
+            return;
+        }
+        var str = Encoding.UTF8.GetBytes("\nIt's Valentines day! Lovely!");
+        await sink.WriteToTargetAsync(writer => writer.WriteAsync(str).AsTask(),
+            config, source.Object, NullLogger.Instance);
+
+        var dataSource = new FileDataSource();
+        var stream = dataSource.ReadFile(destfile, CompressionEnum.None, NullLogger.Instance)!;
+        var buffer = new byte[100];
+        await stream.ReadAsync(buffer.AsMemory(0, buffer.Length));
+        var result = Encoding.UTF8.GetString(buffer);
+        Assert.AreEqual("Hello world!\nIt's Valentines day! Lovely!", result.TrimEnd('\0'));
+    }
+
+}

Interfaces/Cosmos.DataTransfer.Common.UnitTests/FileDataSourceTests.cs

@@ -0,0 +1,97 @@
+using System.Text;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Microsoft.Extensions.Logging.Abstractions;
+using Cosmos.DataTransfer.JsonExtension.UnitTests;
+
+namespace Cosmos.DataTransfer.Common.UnitTests;
+
+[TestClass]
+public class FileSinkDataSourceTests {
+
+    public static IEnumerable<object[]> Test_ReadFileData { get {
+        yield return new object[] { "Data/foobar.txt", CompressionEnum.None };
+        yield return new object[] { "Data/foobar.txt.br", CompressionEnum.None };
+        yield return new object[] { "Data/foobar.txt.gz", CompressionEnum.None };
+        yield return new object[] { "Data/foobar.txt.gzip", CompressionEnum.None };
+        yield return new object[] { "Data/foobar.txt.zz", CompressionEnum.None };
+        yield return new object[] { "Data/foobar.txt.br", CompressionEnum.Brotli};
+        yield return new object[] { "Data/foobar.txt.gz", CompressionEnum.Gzip };
+        yield return new object[] { "Data/foobar.txt.gzip", CompressionEnum.Gzip };
+        yield return new object[] { "Data/foobar.txt.zz", CompressionEnum.Deflate };
+    } }
+
+    [TestMethod]
+    [DynamicData(nameof(Test_ReadFileData))]
+    public async Task Test_ReadFile(string filePath, CompressionEnum compression, string expected = "Hello world!") {
+        var fileSource = new FileDataSource();
+        var reader = fileSource.ReadFile(filePath, compression, NullLogger.Instance)!;
+        var buffer = new byte[100];
+        await reader.ReadAsync(buffer.AsMemory(0, buffer.Length));
+        var result = Encoding.UTF8.GetString(buffer);
+        Assert.AreEqual(expected, result.TrimEnd('\0'), $"file: {filePath}, compression: {compression}.");
+    }
+
+    [TestMethod]
+    [DynamicData(nameof(Test_ReadFileData))]
+    public async Task Test_ReadSourceAsync_SingleFile(string filePath, CompressionEnum compression, string expected = "Hello world!") {
+        var fileSource = new FileDataSource();
+        var config = TestHelpers.CreateConfig(new Dictionary<string,string>() {
+            { "FilePath", filePath },
+            { "Compression", compression.ToString() }
+        });
+
+        var streams = await fileSource.ReadSourceAsync(config, NullLogger.Instance).ToListAsync();
+        Assert.AreEqual(1, streams.Count);
+        var stream = streams!.First();
+        var buffer = new byte[100];
+        await stream!.ReadAsync(buffer.AsMemory(0, buffer.Length));
+        var result = Encoding.UTF8.GetString(buffer);
+        Assert.AreEqual(expected, result.TrimEnd('\0'), $"file: {filePath}, compression: {compression}.");
+    }
+
+    internal static string GetTemporaryDirectory() {
+        string tempDirectory = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
+        if (File.Exists(tempDirectory)) 
+        {
+            return GetTemporaryDirectory();
+        } else 
+        {
+            Directory.CreateDirectory(tempDirectory);
+            return tempDirectory;
+        }
+    }
+
+    [TestMethod]
+    public async Task Test_ReadSourceAsync_Directory() {
+        var files = new string[] {
+            "rich.json", "rich.json.gz", "rich.json.gzip"
+        };
+        var tmpdir = GetTemporaryDirectory();
+        foreach (var f in files) {
+            File.Copy(Path.Combine("Data", f), Path.Combine(tmpdir, f));
+        }
+        var fileSource = new FileDataSource();
+        var logs = new List<string>();
+        var config = TestHelpers.CreateConfig(new Dictionary<string,string>() {
+            { "FilePath", tmpdir }
+        });
+
+        var result = await fileSource.ReadSourceAsync(config, NullLogger.Instance)
+            .SelectAwait(async stream => {
+                var buffer = new byte[100];
+                await stream!.ReadAsync(buffer.AsMemory(0, buffer.Length));
+                return Encoding.UTF8.GetString(buffer).TrimEnd('\0');
+            }).ToArrayAsync();
+        var expected = new string[files.Length];
+        Array.Fill(expected, "[{\"id\":1,\"name\":\"john\"}]");
+        CollectionAssert.AreEqual(expected, result);
+    }
+
+    [TestMethod]
+    public void Test_GetSettings() {
+        var sink = new FileDataSource();
+        var response = sink.GetSettings().ToArray();
+        Assert.AreEqual(1, response.Length);
+        Assert.IsInstanceOfType(response[0], typeof(FileSourceSettings));
+    }
+}

Interfaces/Cosmos.DataTransfer.Common.UnitTests/FileSinkSettingsTests.cs

@@ -0,0 +1,49 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Cosmos.DataTransfer.Interfaces;
+
+namespace Cosmos.DataTransfer.Common.UnitTests;
+
+[TestClass]
+public class FileSinkSettingsTests {
+
+    [TestMethod]
+    public void TestValidate_PassMinimum() {
+        var settings = new FileSinkSettings() {
+            FilePath = "A file",
+        };
+        settings.Validate();
+    }
+
+    [TestMethod]
+    public void TestValidate_Fails() {
+        var settings = new FileSinkSettings() {
+            FilePath = " ",
+        };
+        var e = Assert.ThrowsException<AggregateException>(() => settings.Validate());
+        Assert.AreEqual(e.InnerException!.Message, "The FilePath field is required.");
+    }
+
+    [TestMethod]
+    [DataRow(false, "None", true)]
+    [DataRow(false, "Gzip", true)]
+    [DataRow(false, "Brotli", true)]
+    [DataRow(false, "Deflate", true)]
+    [DataRow(true, "None", true)]
+    [DataRow(true, "Gzip", false)]
+    [DataRow(true, "Brotli", false)]
+    [DataRow(true, "Deflate", false)]
+    public void TestValidate_TestCombinations(bool append, string compression, bool pass) {
+        var settings = new FileSinkSettings() {
+            FilePath = "A file",
+            Append = append,
+            Compression = Enum.Parse<CompressionEnum>(compression),
+        };
+        if (!pass) {
+            Assert.ThrowsException<AggregateException>(() => settings.Validate());
+        } else {
+            settings.Validate();
+        }
+    }
+
+
+}

Interfaces/Cosmos.DataTransfer.Common/CompressionEnum.cs

@@ -0,0 +1,9 @@
+namespace Cosmos.DataTransfer.Common;
+
+public enum CompressionEnum {
+    None = 0,
+    Gzip = 1,
+    Brotli = 2,
+    Deflate = 3,
+    //Bzip = 4
+}

Interfaces/Cosmos.DataTransfer.Common/Cosmos.DataTransfer.Common.csproj

@@ -14,4 +14,15 @@
     <ProjectReference Include="..\Cosmos.DataTransfer.Interfaces\Cosmos.DataTransfer.Interfaces.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <!-- Allow tests to see internal scoped types, properties/fields and methods. -->
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+        <_Parameter1>Cosmos.DataTransfer.Common.UnitTests</_Parameter1>
+    </AssemblyAttribute>
+    <!-- Allow Moq to see internal scoped types, properties/fields and methods. -->
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+        <_Parameter1>DynamicProxyGenAssembly2</_Parameter1>
+    </AssemblyAttribute>
+  </ItemGroup>
+
 </Project>