CSHARP-1991: Add zlib and snappy compressions.

Author: DmitryLukyanov

README.md

@@ -103,6 +103,7 @@ Please see our [guidelines](CONTRIBUTING.md) for contributing to the driver.
 * Teun Duynstee             teun@duynstee.com
 * Einar Egilsson            https://github.com/einaregilsson
 * Ken Egozi                 mail@kenegozi.com
+* Alexander Endris          https://github.com/AlexEndris
 * Daniel Goldman            daniel@stackwave.com
 * Simon Green               simon@captaincodeman.com
 * James Hadwen              james.hadwen@sociustec.com

build.cake

@@ -96,6 +96,12 @@ Task("BuildArtifacts")
                     fileNames.Add("DnsClient.dll");
                 }
 
+                // SharpCompress.dll is needed by Sandcastle
+                if (targetFramework == "net452" && project == "MongoDB.Driver.Core")
+                {
+                    fileNames.Add("SharpCompress.dll");
+                }
+
                 foreach (var fileName in fileNames)
                 {
                     var fromFile = fromDirectory.CombineWithFilePath(fileName);

evergreen/evergreen.yml

@@ -434,6 +434,18 @@ axes:
         variables:
            SSL: "nossl"
 
+  - id: compressor
+    display_name: Compressor
+    values:
+      - id: "zlib"
+        display_name: Zlib
+        variables:
+           COMPRESSOR: "zlib"
+      - id: "snappy"
+        display_name: Snappy
+        variables:
+           COMPRESSOR: "snappy"
+
 buildvariants:
 
 - name: windows-64-compile
@@ -457,6 +469,20 @@ buildvariants:
   tasks:
      - name: test
 
+- matrix_name: "tests-zlib-compression"
+  matrix_spec: { compressor : "zlib", auth: "noauth", ssl: "nossl", version: ["3.4", "3.6", "4.0", "latest"], topology: "standalone", os: "*" }
+  display_name: "${version} ${compressor} ${topology} ${auth} ${ssl} ${os} "
+  tags: ["tests-variant"]
+  tasks:
+     - name: "test"
+
+#- matrix_name: "tests-snappy-compression"
+#  matrix_spec: { compressor : "snappy", auth: "noauth", ssl: "nossl", version: ["3.4", "3.6", "4.0", "latest"], topology: "standalone", os: "*" }
+#  display_name: "${version} ${compressor} ${topology} ${auth} ${ssl} ${os} "
+#  tags: ["tests-variant"]
+#  tasks:
+#     - name: "test"
+
 - name: atlas-connectivity-tests
   display_name: "Atlas Connectivity Tests"
   run_on:

src/MongoDB.Bson/IO/ByteBufferStream.cs

@@ -23,7 +23,7 @@ namespace MongoDB.Bson.IO
     /// Represents a Stream backed by an IByteBuffer. Similar to MemoryStream but backed by an IByteBuffer
     /// instead of a byte array and also implements the BsonStream interface for higher performance BSON I/O.
     /// </summary>
-    public class ByteBufferStream : BsonStream
+    public class ByteBufferStream : BsonStream, IStreamEfficientCopyTo
     {
         // private fields
         private IByteBuffer _buffer;
@@ -122,6 +122,19 @@ public override long Position
         }
 
         // public methods
+        /// <inheritdoc/>
+        public void EfficientCopyTo(Stream destination)
+        {
+            long remainingCount;
+            while ((remainingCount = Length - Position) > 0)
+            {
+                var segment = _buffer.AccessBackingBytes((int)Position);
+                var count = (int)Math.Min(segment.Count, remainingCount);
+                destination.Write(segment.Array, segment.Offset, count);
+                Position += count;
+            }
+        }
+
         /// <inheritdoc/>
         public override void Flush()
         {

src/MongoDB.Bson/IO/IStreamEfficientCopyTo.cs

@@ -0,0 +1,31 @@
+/* Copyright 2019-present MongoDB Inc.
+*
+* 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.
+*/
+
+using System.IO;
+
+namespace MongoDB.Bson.IO
+{
+    /// <summary>
+    /// Represents the effective CopyTo method.
+    /// </summary>
+    public interface IStreamEfficientCopyTo
+    {
+        /// <summary>
+        /// Copy the current stream to the destination without making unnecessary copies of the bytes.
+        /// </summary>
+        /// <param name="destination">The destination stream.</param>
+        void EfficientCopyTo(Stream destination);
+    }
+}

src/MongoDB.Bson/MongoDB.Bson.csproj

@@ -42,7 +42,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.CodeAnalysis.FxCopAnalyzers" Version="2.6.1" PrivateAssets="All" />
+    <PackageReference Include="Microsoft.CodeAnalysis.FxCopAnalyzers" Version="2.6.2" PrivateAssets="All" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.5'">

src/MongoDB.Driver.Core/Core/Compression/CompressorSource.cs

@@ -0,0 +1,107 @@
+/* Copyright 2019–present MongoDB Inc.
+*
+* 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.
+*/
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using MongoDB.Driver.Core.Configuration;
+using MongoDB.Driver.Core.Misc;
+
+namespace MongoDB.Driver.Core.Compression
+{
+    /// <summary>
+    /// Represents a compressor source.
+    /// </summary>
+    public interface ICompressorSource
+    {
+        /// <summary>
+        /// Gets or creates a compressor based on the compressor type.
+        /// </summary>
+        /// <param name="compressorType">The compressor type.</param>
+        /// <returns>The compressor.</returns>
+        ICompressor Get(CompressorType compressorType);
+    }
+
+    internal class CompressorSource : ICompressorSource
+    {
+        #region static
+        public static bool IsCompressorSupported(CompressorType compressorType)
+        {
+            switch (compressorType)
+            {
+#if NET452 || NETSTANDARD2_0
+                case CompressorType.Snappy:
+                    return true;
+#endif
+                case CompressorType.Zlib:
+                    return true;
+                case CompressorType.Noop: // This is realistically only used for testing
+                    return true;
+                default:
+                    return false;
+            }
+        }
+        #endregion
+
+        private readonly IReadOnlyList<CompressorConfiguration> _allowedCompressors;
+        private readonly ConcurrentDictionary<CompressorType, ICompressor> _cache;
+
+        public CompressorSource(IReadOnlyList<CompressorConfiguration> allowedCompressors)
+        {
+            _allowedCompressors = Ensure.IsNotNull(allowedCompressors, nameof(allowedCompressors));
+            _cache = new ConcurrentDictionary<CompressorType, ICompressor>();
+        }
+
+        public ICompressor Get(CompressorType compressorType)
+        {
+            return _cache.GetOrAdd(compressorType, CreateCompressor);
+        }
+
+        private ICompressor CreateCompressor(CompressorConfiguration compressorConfiguration)
+        {
+            switch (compressorConfiguration.Type)
+            {
+                case CompressorType.Noop:
+                    return new NoopCompressor();
+                case CompressorType.Snappy:
+                    return new SnappyCompressor();
+                case CompressorType.Zlib:
+                    {
+                        int? zlibCompressionLevel = null;
+                        if (compressorConfiguration.Properties.ContainsKey("Level"))
+                        {
+                            zlibCompressionLevel = (int)compressorConfiguration.Properties["Level"];
+                        }
+
+                        return new ZlibCompressor(zlibCompressionLevel);
+                    }
+            }
+
+            throw new NotSupportedException($"The compressor {compressorConfiguration.Type} is not supported.");
+        }
+
+        private ICompressor CreateCompressor(CompressorType compressorType)
+        {
+            var compressorConfiguration = _allowedCompressors.FirstOrDefault(c => c.Type == compressorType);
+            if (compressorConfiguration == null)
+            {
+                throw new NotSupportedException($"The compressor {compressorType} is not one of the allowed compressors.");
+            }
+
+            return CreateCompressor(compressorConfiguration);
+        }
+    }
+}

src/MongoDB.Driver.Core/Core/Compression/ICompressor.cs

@@ -1,4 +1,4 @@
-/* Copyright 2013-present MongoDB Inc.
+/* Copyright 2019–present MongoDB Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -13,34 +13,49 @@
 * limitations under the License.
 */
 
+using System.IO;
+
 namespace MongoDB.Driver.Core.Compression
 {
-	/// <summary>
-	/// Represents a compressor.
-	/// </summary>
-	public interface ICompressor
-	{
-		/// <summary>
-		/// Gets the name of the compressor.
-		/// </summary>
-		string Name { get; }
-		
-		/// <summary>
-		/// Gets the id of the compressor
-		/// </summary>
-		CompressorId Id { get; }
+    /// <summary>
+    /// Represents the compressor type.
+    /// </summary>
+    public enum CompressorType
+    {
+        // NOTE: the numeric values of the enum members MUST be kept in sync with the binary wire protocol
+
+        /// <summary>
+        /// The content of the message is uncompressed. This is realistically only used for testing.
+        /// </summary>
+        Noop = 0,
+        /// <summary>
+        /// The content of the message is compressed using snappy.
+        /// </summary>
+        Snappy = 1,
+        /// <summary>
+        /// The content of the message is compressed using zlib. 
+        /// </summary>
+        Zlib = 2
+    }
+
+    /// <summary>
+    /// Represents a compressor.
+    /// </summary>
+    public interface ICompressor
+    {
+        /// <summary>
+        /// Gets the compressor type.
+        /// </summary>
+        CompressorType Type { get; }
 
-		/// <summary>
-		/// Compresses the specified byte array with a given offset.
-		/// </summary>
-		/// <param name="bytesToCompress">Bytes to compress.</param>
-		/// <param name="offset">Offset of the bytes.</param>
-		byte[] Compress(byte[] bytesToCompress, int offset);
+        /// <summary>
+        /// Compresses the specified stream.
+        /// </summary>
+        void Compress(Stream input, Stream output);
 
-		/// <summary>
-		/// Decompresses the specified byte array.
-		/// </summary>
-		/// <param name="bytesToDecompress">Bytes to decompress.</param>
-		byte[] Decompress(byte[] bytesToDecompress);
-	}
-}
+        /// <summary>
+        /// Decompresses the specified stream.
+        /// </summary>
+        void Decompress(Stream input, Stream output);
+    }
+}