OP_COMPRESSION

devel/config/mongod-3.6-compression-zlib.yml

@@ -0,0 +1,7 @@
+---
+type: single
+default_args: -v --bind_ip 0.0.0.0 --enableMajorityReadConcern --networkMessageCompressors zlib
+default_version: 3.6
+mongod:
+    - name: host1
+

devel/t-dynamic/OP_COMPRESSION.t

@@ -0,0 +1,123 @@
+#
+#  Copyright 2015 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.
+#
+
+use strict;
+use warnings;
+use Test::More 0.96;
+use Test::Fatal;
+use Test::Deep qw/!blessed/;
+use UUID::URandom qw/create_uuid/;
+
+use utf8;
+use Tie::IxHash;
+
+use MongoDB;
+use MongoDB::Error;
+
+use lib "t/lib";
+use lib "devel/lib";
+
+use if $ENV{MONGOVERBOSE}, qw/Log::Any::Adapter Stderr/;
+
+use MongoDBTest::Orchestrator;
+
+use MongoDBTest qw/
+    build_client
+    get_test_db
+    server_version
+    server_type
+    clear_testdbs
+    get_unique_collection
+    uuid_to_string
+/;
+
+my $orc =
+MongoDBTest::Orchestrator->new(
+  config_file => "devel/config/mongod-3.6-compression-zlib.yml" );
+$orc->start;
+
+$ENV{MONGOD} = $orc->as_uri;
+
+print $ENV{MONGOD}, "\n";
+
+my $conn = build_client(
+    compressors => ['zlib'],
+    zlib_compression_level => 9,
+);
+my $testdb         = get_test_db($conn);
+my $server_version = server_version($conn);
+my $server_type    = server_type($conn);
+my $coll           = $testdb->get_collection('test_collection');
+
+plan skip_all => "Requires MongoDB 3.6"
+    if $server_version < v3.6.0;
+
+my $server = $orc->get_server('host1');
+my $logfile = $server->logfile;
+
+open my $logfile_fh, '<', $server->logfile
+    or die "Unable to read $logfile";
+
+my @init_messages = collect_log_messages();
+ok scalar(grep { /zlib is supported/ } @init_messages),
+    'zlib is supported';
+
+$coll->insert_one({ value => 23 });
+subtest 'compression for insert one' => \&subtest_roundtrip;
+
+$coll->insert_many([{ value => 24 }, { value => 25 }]);
+subtest 'compression for insert many' => \&subtest_roundtrip;
+
+$testdb->run_command([getnonce => 1]);
+subtest 'no compression on getnonce' => \&subtest_no_compression;
+
+subtest 'connection string' => sub {
+    my $client = MongoDB->connect(
+        $orc->as_uri.'/?compressors=zlib&zlibCompressionLevel=9',
+    );
+    is_deeply $client->compressors, ['zlib'], 'compressors';
+    is $client->zlib_compression_level, 9, 'zlib compression level';
+};
+
+clear_testdbs;
+
+done_testing;
+
+sub subtest_no_compression {
+    my @messages = collect_log_messages();
+    is scalar(grep { /\bdecompressing message with zlib/i } @messages), 0,
+        'no decompressed message';
+    is scalar(grep { /\bcompressing message with zlib/i } @messages), 0,
+        'no compressed message';
+}
+
+sub subtest_roundtrip {
+    my @messages = collect_log_messages();
+    is scalar(grep { /\bdecompressing message with zlib/i } @messages), 1,
+        'decompressed message';
+    is scalar(grep { /\bcompressing message with zlib/i } @messages), 1,
+        'compressed message';
+}
+
+sub collect_log_messages {
+    my @messages;
+    while (defined(my $line = <$logfile_fh>)) {
+        chomp $line;
+        push @messages, $line
+            if $line =~ m{zlib};
+    }
+    return @messages;
+}

lib/MongoDB/MongoClient.pm

@@ -53,12 +53,14 @@ use MongoDB::_Types qw(
     AuthMechanism
     Boolish
     BSONCodec
+    CompressionType
     Document
     HeartbeatFreq
     MaxStalenessNum
     NonNegNum
     ReadPrefMode
     ReadPreference
+    ZlibCompressionLevel
 );
 use Types::Standard qw(
     CodeRef
@@ -230,6 +232,50 @@ sub _build_bson_codec {
     return BSON->new();
 }
 
+=attr compressors
+
+An array reference of compression type names. Currently only C<zlib>
+is supported.
+
+=cut
+
+has compressors => (
+    is      => 'lazy',
+    isa     => ArrayRef[CompressionType],
+    builder => '_build_compressors',
+);
+
+sub _build_compressors {
+    my ($self) = @_;
+    return $self->__uri_or_else(
+        u => 'compressors',
+        e => 'compressors',
+        d => [],
+    );
+}
+
+=attr zlib_compression_level
+
+An integer from C<-1> to C<9> specifying the compression level to use
+when L</compression> is set to C<zlib>.
+
+=cut
+
+has zlib_compression_level => (
+    is      => 'lazy',
+    isa     => ZlibCompressionLevel,
+    builder => '_build_zlib_compression_level',
+);
+
+sub _build_zlib_compression_level {
+    my ($self) = @_;
+    return $self->__uri_or_else(
+        u => 'zlibcompressionlevel',
+        e => 'zlib_compression_level',
+        d => -1,
+    );
+}
+
 =attr connect_timeout_ms
 
 This attribute specifies the amount of time in milliseconds to wait for a
@@ -1222,6 +1268,8 @@ sub _build__topology {
             ( ref( $self->ssl ) eq 'HASH' ? ( SSL_options => $self->ssl ) : () ),
         },
         monitoring_callback => $self->monitoring_callback,
+        compression => $self->compressors,
+        zlib_compression_level => $self->zlib_compression_level,
     );
 }
 
@@ -2040,6 +2088,7 @@ The currently supported connection string options are:
 *appName
 *authMechanism
 *authMechanism.SERVICE_NAME
+*compressors
 *connectTimeoutMS
 *journal
 *readPreference
@@ -2048,6 +2097,7 @@ The currently supported connection string options are:
 *ssl
 *w
 *wtimeoutMS
+*zlibCompressionLevel
 
 See the official MongoDB documentation on connection strings for more on the URI
 format and connection string options:

lib/MongoDB/Op/_Command.pm

@@ -89,9 +89,16 @@ sub execute {
     $self->publish_command_started( $link, $self->{query}, $request_id )
       if $self->monitoring_callback;
 
+    my %write_opt;
+    my $command_name = do {
+        my $command = _to_tied_ixhash($self->{query});
+        lc tied(%$command)->Keys(0);
+    };
+    $write_opt{disable_compression} = !is_compressible($command_name);
+
     my $result;
     eval {
-        $link->write( $op_bson ),
+        $link->write( $op_bson, %write_opt ),
         ( $result = MongoDB::_Protocol::parse_reply( $link->read, $request_id ) );
     };
     if ( my $err = $@ ) {
@@ -117,4 +124,20 @@ sub execute {
     return $res;
 }
 
+sub is_compressible {
+    my $command_name = lc shift;
+    return not grep { $_ eq $command_name } qw(
+        ismaster
+        saslstart
+        saslcontinue
+        getnonce
+        authenticate
+        createuser
+        updateuser
+        copydbsaslstart
+        copydbgetnonce
+        copydb
+    );
+}
+
 1;

lib/MongoDB/_Link.pm

@@ -35,6 +35,7 @@ use Socket qw/SOL_SOCKET SO_KEEPALIVE SO_RCVBUF IPPROTO_TCP TCP_NODELAY AF_INET/
 use Time::HiRes qw/time/;
 use MongoDB::Error;
 use MongoDB::_Constants;
+use MongoDB::_Protocol;
 use MongoDB::_Types qw(
     Boolish
     HostAddress
@@ -422,7 +423,20 @@ sub is_connected {
 }
 
 sub write {
-    my ( $self, $buf ) = @_;
+    my ( $self, $buf, %write_opt ) = @_;
+
+    if (
+        !$write_opt{disable_compression}
+        && $self->server
+        && $self->server->compressor
+        && MongoDB::_Protocol::is_compressible($buf)
+    ) {
+        $buf = MongoDB::_Protocol::compress(
+            $buf,
+            $self->server->compressor,
+            zlib_compression_level => $self->server->zlib_compression_level,
+        );
+    }
 
     my ( $len, $off, $pending, $nfound, $r ) = ( length($buf), 0 );