You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bl...@apache.org on 2024/11/29 10:52:13 UTC
(pulsar-dotpulsar) branch master updated: Add ZLib compression using System.IO.Compression (#238)
This is an automated email from the ASF dual-hosted git repository.
blankensteiner pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-dotpulsar.git
The following commit(s) were added to refs/heads/master by this push:
new a39573d Add ZLib compression using System.IO.Compression (#238)
a39573d is described below
commit a39573d11bda5e6d34df7b36c2deae066016dd27
Author: Bjørn Madsen <bm...@aeons.dk>
AuthorDate: Fri Nov 29 11:52:09 2024 +0100
Add ZLib compression using System.IO.Compression (#238)
* Add ZLib compression using System.IO.Compression
* Update benchmark with builtin zlib compression
* Add test for BuiltinZlibCompression
* Add Apache header
---
benchmarks/Compression/Compress.cs | 7 ++
benchmarks/Compression/Decompress.cs | 10 +++
benchmarks/Compression/Factories.cs | 8 +++
benchmarks/Compression/Program.cs | 2 +
.../Internal/Compression/BuiltinZlibCompression.cs | 77 ++++++++++++++++++++++
.../Internal/Compression/CompressionFactories.cs | 4 +-
.../Compression/BuiltinZlibCompressionTests.cs | 37 +++++++++++
7 files changed, 144 insertions(+), 1 deletion(-)
diff --git a/benchmarks/Compression/Compress.cs b/benchmarks/Compression/Compress.cs
index 66b7b56..994de36 100644
--- a/benchmarks/Compression/Compress.cs
+++ b/benchmarks/Compression/Compress.cs
@@ -46,6 +46,13 @@ public class Compress
_ = compressor.Compress(Data);
}
+ [Benchmark]
+ public void BuiltinZlib()
+ {
+ using var compressor = Factories.BuiltinZlibCompressionCompressorFactory.Create();
+ _ = compressor.Compress(Data);
+ }
+
[Benchmark]
public void DotNetZip()
{
diff --git a/benchmarks/Compression/Decompress.cs b/benchmarks/Compression/Decompress.cs
index 902511a..2980705 100644
--- a/benchmarks/Compression/Decompress.cs
+++ b/benchmarks/Compression/Decompress.cs
@@ -37,6 +37,8 @@ public class Decompress
K4aosCompressionLz4Data = compressor.Compress(data);
using (var compressor = Factories.IronSnappyCompressorFactory.Create())
IronSnappyData = compressor.Compress(data);
+ using (var compressor = Factories.BuiltinZlibCompressionCompressorFactory.Create())
+ BuiltinZlibData = compressor.Compress(data);
using (var compressor = Factories.DotNetZipCompressorFactory.Create())
DotNetZipData = compressor.Compress(data);
using (var compressor = Factories.ZstdNetCompressorFactory.Create())
@@ -48,6 +50,7 @@ public class Decompress
public int DecompressedSize { get; private set; }
public ReadOnlySequence<byte> K4aosCompressionLz4Data { get; private set; }
public ReadOnlySequence<byte> IronSnappyData { get; private set; }
+ public ReadOnlySequence<byte> BuiltinZlibData { get; private set; }
public ReadOnlySequence<byte> DotNetZipData { get; private set; }
public ReadOnlySequence<byte> ZstdNetData { get; private set; }
public ReadOnlySequence<byte> ZstdSharpData { get; private set; }
@@ -66,6 +69,13 @@ public class Decompress
_ = decompressor.Decompress(IronSnappyData, DecompressedSize);
}
+ [Benchmark]
+ public void BuiltinZlib()
+ {
+ using var decompressor = Factories.BuiltinZlibCompressionDecompressorFactory.Create();
+ _ = decompressor.Decompress(BuiltinZlibData, DecompressedSize);
+ }
+
[Benchmark]
public void DotNetZip()
{
diff --git a/benchmarks/Compression/Factories.cs b/benchmarks/Compression/Factories.cs
index 045e377..7de2a2f 100644
--- a/benchmarks/Compression/Factories.cs
+++ b/benchmarks/Compression/Factories.cs
@@ -31,6 +31,11 @@ public static class Factories
IronSnappyCompressorFactory = compressor!;
IronSnappyDecompressorFactory = decompressor!;
+ if (!BuiltinZlibCompression.TryLoading(out compressor, out decompressor))
+ throw new Exception("Could not load BuiltinZlibCompression");
+ BuiltinZlibCompressionCompressorFactory = compressor!;
+ BuiltinZlibCompressionDecompressorFactory = decompressor!;
+
if (!ZlibCompression.TryLoading(out compressor, out decompressor))
throw new Exception("Could not load DotNetZip");
DotNetZipCompressorFactory = compressor!;
@@ -53,6 +58,9 @@ public static class Factories
public static ICompressorFactory IronSnappyCompressorFactory { get; }
public static IDecompressorFactory IronSnappyDecompressorFactory { get; }
+ public static ICompressorFactory BuiltinZlibCompressionCompressorFactory { get; }
+ public static IDecompressorFactory BuiltinZlibCompressionDecompressorFactory { get; }
+
public static ICompressorFactory DotNetZipCompressorFactory { get; }
public static IDecompressorFactory DotNetZipDecompressorFactory { get; }
diff --git a/benchmarks/Compression/Program.cs b/benchmarks/Compression/Program.cs
index 1b91426..6f0aa4e 100644
--- a/benchmarks/Compression/Program.cs
+++ b/benchmarks/Compression/Program.cs
@@ -33,6 +33,8 @@ static void OutputCompressionInfo(MessageSize size, MessageType type)
Console.WriteLine($"\tCompressed with IronSnappy: {compressor.Compress(data).Length}");
using (var compressor = Factories.DotNetZipCompressorFactory.Create())
Console.WriteLine($"\tCompressed with DotNetZip: {compressor.Compress(data).Length}");
+ using (var compressor = Factories.BuiltinZlibCompressionCompressorFactory.Create())
+ Console.WriteLine($"\tCompressed with System.IO.Compression: {compressor.Compress(data).Length}");
using (var compressor = Factories.ZstdNetCompressorFactory.Create())
Console.WriteLine($"\tCompressed with ZstdNet: {compressor.Compress(data).Length}");
using (var compressor = Factories.ZstdSharpCompressorFactory.Create())
diff --git a/src/DotPulsar/Internal/Compression/BuiltinZlibCompression.cs b/src/DotPulsar/Internal/Compression/BuiltinZlibCompression.cs
new file mode 100644
index 0000000..d1b3cc6
--- /dev/null
+++ b/src/DotPulsar/Internal/Compression/BuiltinZlibCompression.cs
@@ -0,0 +1,77 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Internal.Compression;
+
+using DotPulsar.Internal.Abstractions;
+using System.Buffers;
+using System.Reflection;
+
+public static class BuiltinZlibCompression
+{
+ public static bool TryLoading(out ICompressorFactory? compressorFactory, out IDecompressorFactory? decompressorFactory)
+ {
+ try
+ {
+ var assembly = Assembly.Load("System.IO.Compression");
+ var types = assembly.GetTypes();
+
+ // Find the ZLibStream
+ var zlibStreamType = types.FirstOrDefault(t => t.FullName == "System.IO.Compression.ZLibStream");
+
+ // Find the enum values for the ZLibStream constructors
+ var compressionLevelType = types.FirstOrDefault(t => t.FullName == "System.IO.Compression.CompressionLevel");
+ var compressionLevelOptimal = compressionLevelType?.GetEnumValues().GetValue(0);
+ var compressionModeType = types.FirstOrDefault(t => t.FullName == "System.IO.Compression.CompressionMode");
+ var compressionModeDecompress = compressionModeType?.GetEnumValues().GetValue(0);
+
+ compressorFactory = new CompressorFactory(PulsarApi.CompressionType.Zlib, () => new Compressor(CreateCompressor(zlibStreamType, compressionLevelOptimal)));
+ decompressorFactory = new DecompressorFactory(PulsarApi.CompressionType.Zlib, () => new Decompressor(CreateDecompressor(zlibStreamType, compressionModeDecompress)));
+
+ return CompressionTester.TestCompression(compressorFactory, decompressorFactory);
+ }
+ catch
+ {
+ // Ignore
+ }
+
+ compressorFactory = null;
+ decompressorFactory = null;
+
+ return false;
+ }
+
+ private static Func<ReadOnlySequence<byte>, ReadOnlySequence<byte>> CreateCompressor(Type? zlibStreamType, object? compressionLevelOptimal)
+ => data =>
+ {
+ using var dataStream = new MemoryStream(data.ToArray());
+ using var compressedStream = new MemoryStream();
+ using var compressor = (Stream) Activator.CreateInstance(zlibStreamType!, new object[] { compressedStream, compressionLevelOptimal! })!;
+ dataStream.CopyTo(compressor);
+ compressor.Close();
+ return new ReadOnlySequence<byte>(compressedStream.ToArray());
+ };
+
+ private static Func<ReadOnlySequence<byte>, int, ReadOnlySequence<byte>> CreateDecompressor(Type? zlibStreamType, object? compressionModeDecompress)
+ => (data, _) =>
+ {
+ using var dataStream = new MemoryStream(data.ToArray());
+ using var decompressedStream = new MemoryStream();
+ using var decompressor = (Stream) Activator.CreateInstance(zlibStreamType!, new object[] { dataStream, compressionModeDecompress! })!;
+ decompressor.CopyTo(decompressedStream);
+ decompressor.Close();
+ decompressedStream.Position = 0;
+ return new ReadOnlySequence<byte>(decompressedStream.ToArray());
+ };
+}
diff --git a/src/DotPulsar/Internal/Compression/CompressionFactories.cs b/src/DotPulsar/Internal/Compression/CompressionFactories.cs
index 533fb1b..288d168 100644
--- a/src/DotPulsar/Internal/Compression/CompressionFactories.cs
+++ b/src/DotPulsar/Internal/Compression/CompressionFactories.cs
@@ -46,7 +46,9 @@ public static class CompressionFactories
private static void LoadSupportForZlib()
{
- if (ZlibCompression.TryLoading(out var compressorFactory, out var decompressorFactory))
+ if (BuiltinZlibCompression.TryLoading(out var compressorFactory, out var decompressorFactory))
+ Add(compressorFactory, decompressorFactory);
+ else if (ZlibCompression.TryLoading(out compressorFactory, out decompressorFactory))
Add(compressorFactory, decompressorFactory);
}
diff --git a/tests/DotPulsar.Tests/Internal/Compression/BuiltinZlibCompressionTests.cs b/tests/DotPulsar.Tests/Internal/Compression/BuiltinZlibCompressionTests.cs
new file mode 100644
index 0000000..360250d
--- /dev/null
+++ b/tests/DotPulsar.Tests/Internal/Compression/BuiltinZlibCompressionTests.cs
@@ -0,0 +1,37 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Tests.Internal.Compression;
+
+using DotPulsar.Internal.Compression;
+
+[Trait("Category", "Unit")]
+public class BuiltinZlibCompressionTests
+{
+ [Fact]
+ public void Compression_GivenDataToCompressAndDecompress_ShouldReturnOriginalData()
+ {
+ // Arrange
+ var couldLoad = BuiltinZlibCompression.TryLoading(out var compressorFactory, out var decompressorFactory);
+ couldLoad.Should().BeTrue();
+ using var compressor = compressorFactory!.Create();
+ using var decompressor = decompressorFactory!.Create();
+
+ // Act
+ var compressionWorks = CompressionTester.TestCompression(compressorFactory, decompressorFactory);
+
+ // Assert
+ compressionWorks.Should().BeTrue();
+ }
+}