PERL-791 Change stream support

lib/MongoDB/ChangeStream.pm

@@ -0,0 +1,199 @@
+#
+#  Copyright 2009-2018 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;
+package MongoDB::ChangeStream;
+
+# ABSTRACT: A stream providing update information for collections.
+
+use version;
+our $VERSION = 'v1.999.0';
+
+use Moo;
+use Try::Tiny;
+use MongoDB::Cursor;
+use MongoDB::Op::_Aggregate;
+use MongoDB::Error;
+use Safe::Isa;
+use MongoDB::_Types qw(
+    MongoDBCollection
+    ArrayOfHashRef
+);
+use Types::Standard qw(
+    InstanceOf
+    HashRef
+    Str
+);
+
+use namespace::clean -except => 'meta';
+
+has _result => (
+    is => 'rw',
+    isa => InstanceOf['MongoDB::QueryResult'],
+    lazy => 1,
+    builder => '_build_result',
+    clearer => '_clear_result',
+);
+
+has collection => (
+    is => 'ro',
+    isa => MongoDBCollection,
+    required => 1,
+);
+
+has aggregation_options => (
+    is => 'ro',
+    isa => HashRef,
+);
+
+has pipeline => (
+    is => 'ro',
+    isa => ArrayOfHashRef,
+    required => 1,
+);
+
+has full_document => (
+    is => 'ro',
+    isa => Str,
+    predicate => '_has_full_document',
+);
+
+has _resume_token => (
+    is => 'rw',
+    init_arg => 'resume_after',
+    predicate => '_has_resume_token',
+    lazy => 1,
+);
+
+sub BUILD {
+    my ($self) = @_;
+
+    # starting point is construction time instead of first next call
+    $self->_result;
+}
+
+sub _build_result {
+    my ($self) = @_;
+
+    my @pipeline = @{ $self->pipeline };
+    @pipeline = (
+        {'$changeStream' => {
+            ($self->_has_full_document
+                ? (fullDocument => $self->full_document)
+                : ()
+            ),
+            ($self->_has_resume_token
+                ? (resumeAfter => $self->_resume_token)
+                : ()
+            ),
+        }},
+        @pipeline,
+    );
+
+    return $self->collection->aggregate(
+        \@pipeline,
+        $self->aggregation_options,
+    );
+}
+
+=head1 STREAM METHODS
+
+=cut
+
+=head2 next
+
+    $change_stream = $collection->watch(...);
+    $change = $change_stream->next;
+
+Waits for the next change in the collection and returns it.
+
+B<Note>: This method will wait for the amount of milliseconds passed
+as C<maxAwaitTimeMS> to L<MongoDB::Collection/watch> or the server's
+default wait-time. It will not wait indefinitely.
+
+=cut
+
+sub next {
+    my ($self) = @_;
+
+    my $change;
+    my $retried;
+    while (1) {
+        last if try {
+            $change = $self->_result->next;
+            1 # successfully fetched result
+        }
+        catch {
+            my $error = $_;
+            if (
+                not($retried)
+                and $error->$_isa('MongoDB::Error')
+                and $error->_is_resumable
+            ) {
+                $retried = 1;
+                $self->_result($self->_build_result);
+            }
+            else {
+                die $error;
+            }
+            0 # failed, cursor was rebuilt
+        };
+    }
+
+    # this differs from drivers that block indefinitely. we have to
+    # deal with the situation where no results are available.
+    if (not defined $change) {
+        return undef;
+    }
+
+    if (exists $change->{_id}) {
+        $self->_resume_token($change->{_id});
+        return $change;
+    }
+    else {
+        MongoDB::InvalidOperationError->throw(
+            "Cannot provide resume functionality when the ".
+            "resume token is missing");
+    }
+}
+
+1;
+
+=head1 SYNOPSIS
+
+    $stream = $collection->watch( $pipeline, $options );
+    while(1) {
+
+        # This inner loop will only iterate until there are no more
+        # changes available.
+        while (my $change = $stream->next) {
+            ...
+        }
+    }
+
+=head1 DESCRIPTION
+
+This class models change stream results as returned by the
+L<MongoDB::Collection/watch> method.
+
+=head1 SEE ALSO
+
+The L<Change Streams manual section|https://docs.mongodb.com/manual/changeStreams/>.
+
+The L<Change Streams specification|https://github.com/mongodb/specifications/blob/master/source/change-streams.rst>.
+
+=cut

lib/MongoDB/Collection.pm

@@ -24,6 +24,7 @@ package MongoDB::Collection;
 use version;
 our $VERSION = 'v1.999.0';
 
+use MongoDB::ChangeStream;
 use MongoDB::Error;
 use MongoDB::IndexView;
 use MongoDB::InsertManyResult;
@@ -1000,6 +1001,85 @@ sub find_one_and_update {
     return $self->_find_one_and_update_or_replace($filter, $update, $options);
 }
 
+=method watch
+
+Watches for changes on this collection-
+
+Perform an aggregation with an implicit initial C<$changeStream> stage
+and returns a L<MongoDB::ChangeStream> result which can be used to
+iterate over the changes in the collection. This functionality is
+available since MongoDB 3.6.
+
+    my $stream = $collection->watch();
+    my $stream = $collection->watch( \@pipeline );
+    my $stream = $collection->watch( \@pipeline, \%options );
+
+    while (1) {
+
+        # This inner loop will only run until no more changes are
+        # available.
+        while (my $change = $stream->next) {
+            # process $change
+        }
+    }
+
+The returned stream will not block forever waiting for changes. If you
+want to respond to changes over a longer time use C<maxAwaitTimeMS> and
+regularly call C<next> in a loop.
+
+B<Note>: Using this helper method is preferred to manually aggregating
+with a C<$changeStream> stage, since it will automatically resume when
+the connection was terminated.
+
+The optional first argument must be an array-ref of
+L<aggregation pipeline|http://docs.mongodb.org/manual/core/aggregation-pipeline/>
+documents. Each pipeline document must be a hash reference. Not all
+pipeline stages are supported after C<$changeStream>.
+
+The optional second argument is a hash reference with options:
+
+=for :list
+* C<fullDocument> - The fullDocument to pass as an option to the
+  C<$changeStream> stage. Allowed values: C<default>, C<updateLookup>.
+  Defaults to C<default>.  When set to C<updateLookup>, the change
+  notification for partial updates will include both a delta describing the
+  changes to the document, as well as a copy of the entire document that
+  was changed from some time after the change occurred.
+* C<resumeAfter> - The logical starting point for this change stream.
+  This value can be obtained from the C<_id> field of a document returned
+  by L<MongoDB::ChangeStream/next>.
+* C<maxAwaitTimeMS> - The maximum number of milliseconds for the server
+  to wait before responding.
+
+See L</aggregate> for more available options.
+
+See the L<manual section on Change Streams|https://docs.mongodb.com/manual/changeStreams/>
+for general usage information on change streams.
+
+See the L<Change Streams specification|https://github.com/mongodb/specifications/blob/master/source/change-streams.rst>
+for details on change streams.
+
+=cut
+
+sub watch {
+    my ( $self, $pipeline, $options ) = @_;
+
+    $pipeline ||= [];
+    $options ||= {};
+
+    return MongoDB::ChangeStream->new(
+        exists($options->{fullDocument})
+            ? (full_document => delete $options->{fullDocument})
+            : (full_document => 'default'),
+        exists($options->{resumeAfter})
+            ? (resume_after => delete $options->{resumeAfter})
+            : (),
+        collection => $self,
+        pipeline => $pipeline,
+        aggregation_options => $options,
+    );
+}
+
 =method aggregate
 
     @pipeline = (
@@ -1082,6 +1162,12 @@ sub aggregate {
         options      => $options,
         read_concern => $self->read_concern,
         has_out      => $last_op eq '$out',
+        exists($options->{cursorType})
+            ? (cursorType => delete $options->{cursorType})
+            : (cursorType => 'non_tailable'),
+        exists($options->{maxAwaitTimeMS})
+            ? (maxAwaitTimeMS => delete $options->{maxAwaitTimeMS})
+            : (),
         %{ $self->_op_args },
     );
 

lib/MongoDB/Error.pm

@@ -45,6 +45,7 @@ BEGIN {
         UNKNOWN_ERROR             => 8,
         NAMESPACE_NOT_FOUND       => 26,
         INDEX_NOT_FOUND           => 27,
+        CURSOR_NOT_FOUND          => 43,
         EXCEEDED_TIME_LIMIT       => 50,
         COMMAND_NOT_FOUND         => 59,
         WRITE_CONCERN_ERROR       => 64,
@@ -109,6 +110,10 @@ sub throw {
   die $throwable;
 }
 
+# internal flag indicating if an operation should be retried when
+# an error occurs.
+sub _is_resumable { 1 }
+
 #--------------------------------------------------------------------------#
 # Subclasses with attributes included inline below
 #--------------------------------------------------------------------------#
@@ -134,6 +139,8 @@ has code => (
 
 sub _build_code { return MongoDB::Error::UNKNOWN_ERROR() }
 
+sub _is_resumable { 0 }
+
 package MongoDB::DocumentError;
 
 use Moo;
@@ -191,6 +198,8 @@ use Moo;
 use namespace::clean;
 extends 'MongoDB::Error';
 
+sub _is_resumable { 1 }
+
 package MongoDB::HandshakeError;
 use Moo;
 use namespace::clean;
@@ -229,6 +238,7 @@ use Moo;
 use namespace::clean;
 extends 'MongoDB::DatabaseError';
 sub _build_code { return MongoDB::Error::NOT_MASTER() }
+sub _is_resumable { 1 }
 
 package MongoDB::WriteError;
 use Moo;
@@ -250,7 +260,9 @@ extends 'MongoDB::Error';
 package MongoDB::CursorNotFoundError;
 use Moo;
 use namespace::clean;
-extends 'MongoDB::Error';
+extends 'MongoDB::DatabaseError';
+sub _build_code { return MongoDB::Error::CURSOR_NOT_FOUND() }
+sub _is_resumable { 1 }
 
 package MongoDB::DecodingError;
 use Moo;
@@ -277,6 +289,11 @@ use Moo;
 use namespace::clean;
 extends 'MongoDB::Error';
 
+package MongoDB::InvalidOperationError;
+use Moo;
+use namespace::clean;
+extends 'MongoDB::Error';
+
 #--------------------------------------------------------------------------#
 # Private error classes
 #--------------------------------------------------------------------------#
@@ -347,10 +364,10 @@ To retry failures automatically, consider using L<Try::Tiny::Retry>.
         |   |
         |   |->MongoDB::NetworkError
         |
-        |->MongoDB::CursorNotFoundError
-        |
         |->MongoDB::DatabaseError
         |   |
+        |   |->MongoDB::CursorNotFoundError
+        |   |
         |   |->MongoDB::DuplicateKeyError
         |   |
         |   |->MongoDB::NotMasterError
@@ -367,6 +384,8 @@ To retry failures automatically, consider using L<Try::Tiny::Retry>.
         |
         |->MongoDB::InternalError
         |
+        |->MongoDB::InvalidOperationError
+        |
         |->MongoDB::ProtocolError
         |
         |->MongoDB::SelectionError