File Coverage

blib/lib/MongoDB/ChangeStream.pm

Criterion	Covered	Total	%
statement	36	73	49.3
branch	0	26	0.0
condition	0	11	0.0
subroutine	12	16	75.0
pod	2	3	66.6
total	50	129	38.7

#  Copyright 2018 - 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.

use strict;
use warnings;
package MongoDB::ChangeStream;

# ABSTRACT: A stream providing update information for collections.

use version;
our $VERSION = 'v2.2.0';

use Moo;
use MongoDB::Cursor;
use MongoDB::Op::_ChangeStream;
use MongoDB::Error;
use Safe::Isa;
use BSON::Timestamp;
use MongoDB::_Types qw(
    MongoDBCollection
    ArrayOfHashRef
    Boolish
    BSONTimestamp
    ClientSession
);
use Types::Standard qw(
    InstanceOf
    HashRef
    Maybe
    Str
    Num
);

use namespace::clean -except => 'meta';

has _result => (
    is => 'rw',
    isa => InstanceOf['MongoDB::QueryResult'],
    init_arg => undef,
);

has _client => (
    is => 'ro',
    isa => InstanceOf['MongoDB::MongoClient'],
    init_arg => 'client',
    required => 1,
);

has _op_args => (
    is => 'ro',
    isa => HashRef,
    init_arg => 'op_args',
    required => 1,
);

has _pipeline => (
    is => 'ro',
    isa => ArrayOfHashRef,
    init_arg => 'pipeline',
    required => 1,
);

has _full_document => (
    is => 'ro',
    isa => Str,
    init_arg => 'full_document',
    predicate => '_has_full_document',
);

has _resume_after => (
    is => 'ro',
    init_arg => 'resume_after',
    predicate => '_has_resume_after',
);

has _start_after => (
    is => 'ro',
    init_arg => 'start_after',
    predicate => '_has_start_after',
);

has _all_changes_for_cluster => (
    is => 'ro',
    isa => Boolish,
    init_arg => 'all_changes_for_cluster',
    default => sub { 0 },
);

has _start_at_operation_time => (
    is => 'ro',
    isa => BSONTimestamp,
    init_arg => 'start_at_operation_time',
    predicate => '_has_start_at_operation_time',
    coerce => sub {
        ref($_[0]) ? $_[0] : BSON::Timestamp->new(seconds => $_[0])
    },
);

has _session => (
    is => 'ro',
    isa => Maybe[ClientSession],
    init_arg => 'session',
);

has _options => (
    is => 'ro',
    isa => HashRef,
    init_arg => 'options',
    default => sub { {} },
);

has _max_await_time_ms => (
    is => 'ro',
    isa => Num,
    init_arg => 'max_await_time_ms',
    predicate => '_has_max_await_time_ms',
);

has _last_operation_time => (
    is => 'rw',
    init_arg => undef,
    predicate => '_has_last_operation_time',
);

has _last_resume_token => (
    is => 'rw',
    init_arg => undef,
    predicate => '_has_last_resume_token',
);

sub BUILD {
    my ($self) = @_;

    # starting point is construction time instead of first next call
    $self->_execute_query;
}

sub _execute_query {
    my ($self) = @_;

    my $resume_opt = {};

    # seen prior results, continuing after last resume token
    if ($self->_has_last_resume_token) {
        $resume_opt->{resume_after} = $self->_last_resume_token;
    }
    elsif ( $self->_has_start_after ) {
        $self->_last_resume_token(
            $resume_opt->{start_after} = $self->_start_after
        );
    }
    # no results yet, but we have operation time from prior query
    elsif ($self->_has_last_operation_time) {
        $resume_opt->{start_at_operation_time} = $self->_last_operation_time;
    }
    # no results and no prior operation time, send specified options
    else {
        $resume_opt->{start_at_operation_time} = $self->_start_at_operation_time
            if $self->_has_start_at_operation_time;
        if ( $self->_has_resume_after ) {
            $self->_last_resume_token(
                $resume_opt->{resume_after} = $self->_resume_after
            );
        }
    }

    my $op = MongoDB::Op::_ChangeStream->new(
        pipeline => $self->_pipeline,
        all_changes_for_cluster => $self->_all_changes_for_cluster,
        session => $self->_session,
        options => $self->_options,
        client => $self->_client,
        $self->_has_full_document
            ? (full_document => $self->_full_document)
            : (),
        $self->_has_max_await_time_ms
            ? (maxAwaitTimeMS => $self->_max_await_time_ms)
            : (),
        %$resume_opt,
        %{ $self->_op_args },
    );

    my $res = $self->_client->send_retryable_read_op($op);
    $self->_result($res->{result});
    $self->_last_operation_time($res->{operationTime})
        if exists $res->{operationTime};
}

#pod =head1 STREAM METHODS
#pod
#pod =cut

#pod =head2 next
#pod
#pod     $change_stream = $collection->watch(...);
#pod     $change = $change_stream->next;
#pod
#pod Waits for the next change in the collection and returns it.
#pod
#pod B: This method will wait for the amount of milliseconds passed
#pod as C to L or the server's
#pod default wait-time. It will not wait indefinitely.
#pod
#pod =cut

sub next {
    my ($self) = @_;

    my $change;
    my $retried;
    while (1) {
        last if eval {
            $change = $self->_result->next;
            1; # successfully fetched result
        } or do {
            my $error = $@ || "Unknown error";
            if (
                not($retried)
                and $error->$_isa('MongoDB::Error')
                and $error->_is_resumable
            ) {
                $retried = 1;
                $self->_execute_query;
            }
            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; ## no critic
    }

    if (exists $change->{'postBatchResumeToken'}) {
        $self->_last_resume_token( $change->{'postBatchResumeToken'} );
        return $change;
    }
    elsif (exists $change->{_id}) {
        $self->_last_resume_token( $change->{_id} );
        return $change;
    }
    else {
        MongoDB::InvalidOperationError->throw(
            "Cannot provide resume functionality when the ".
            "resume token is missing");
    }
}

#pod =head2 get_resume_token
#pod
#pod Users can inspect the C<_id> on each C to use as a
#pod resume token. But since MongoDB 4.2, C and C responses
#pod also include a C. Drivers use one or the other
#pod when automatically resuming.
#pod
#pod This method retrieves the same resume token that would be used to
#pod automatically resume. Users intending to store the resume token
#pod should use this method to get the most up to date resume token.
#pod
#pod For instance:
#pod
#pod     if ($local_change) {
#pod         process_change($local_change);
#pod     }
#pod
#pod     eval {
#pod         my $change_stream = $coll->watch([], { resumeAfter => $local_resume_token });
#pod         while ( my $change = $change_stream->next) {
#pod             $local_resume_token = $change_stream->get_resume_token;
#pod             $local_change = $change;
#pod             process_change($local_change);
#pod         }
#pod     };
#pod     if (my $err = $@) {
#pod         $log->error($err);
#pod     }
#pod
#pod In this case the current change is always persisted locally,
#pod including the resume token, such that on restart the application
#pod can still process the change while ensuring that the change stream
#pod continues from the right logical time in the oplog. It is the
#pod application's responsibility to ensure that C is
#pod idempotent, this design merely makes a reasonable effort to process
#pod each change at least once.
#pod
#pod =cut

sub get_resume_token { $_[0]->_last_resume_token }

1;

__END__