Synthesized command monitoring events for legacy kill cursors

lib/MongoDB/Op/_KillCursors.pm

@@ -64,12 +64,102 @@ sub execute {
             )->execute($link);
         };
     }
+    # Server never sends a reply, so ignoring failure here is automatic.
     else {
-        # Server never sends a reply, so ignoring failure here is automatic.
-        $link->write( MongoDB::_Protocol::write_kill_cursors( @{ $self->cursor_ids } ) );
+        my ($msg, $request_id) = MongoDB::_Protocol::write_kill_cursors(
+            @{ $self->cursor_ids },
+        );
+
+        my $start_event;
+        $start_event = $self->_legacy_publish_command_started(
+            $link,
+            $request_id,
+        ) if $self->monitoring_callback;
+        my $start = time;
+
+        eval {
+            $link->write($msg);
+        };
+
+        my $duration = time - $start;
+        if (my $err = $@) {
+            $self->_legacy_publish_command_exception(
+                $start_event,
+                $duration,
+                $err,
+            ) if $self->monitoring_callback;
+            die $err;
+        }
+
+        $self->_legacy_publish_command_reply($start_event, $duration)
+            if $self->monitoring_callback;
     }
 
     return 1;
 }
 
+sub _legacy_publish_command_started {
+    my ($self, $link, $request_id) = @_;
+
+    my %cmd;
+    tie %cmd, "Tie::IxHash", (
+        killCursors => $self->coll_name,
+        cursors     => $self->cursor_ids,
+    );
+
+    my $event = {
+        type         => 'command_started',
+        databaseName => $self->db_name,
+        commandName  => 'killCursors',
+        command      => \%cmd,
+        requestId    => $request_id,
+        connectionId => $link->address,
+    };
+
+    eval { $self->monitoring_callback->($event) };
+
+    return $event;
+}
+
+sub _legacy_publish_command_exception {
+    my ($self, $start_event, $duration, $err) = @_;
+
+    my $event = {
+        type         => 'command_failed',
+        databaseName => $start_event->{databaseName},
+        commandName  => $start_event->{commandName},
+        requestId    => $start_event->{requestId},
+        connectionId => $start_event->{connectionId},
+        durationSecs => $duration,
+        reply        => {},
+        failure      => "$err",
+        eval_error   => $err,
+    };
+
+    eval { $self->monitoring_callback->($event) };
+
+    return;
+}
+
+sub _legacy_publish_command_reply {
+    my ($self, $start_event, $duration) = @_;
+
+    my $event = {
+        type         => 'command_succeeded',
+        databaseName => $start_event->{databaseName},
+        commandName  => $start_event->{commandName},
+        requestId    => $start_event->{requestId},
+        connectionId => $start_event->{connectionId},
+        durationSecs => $duration,
+        reply        => {
+            ok => 1,
+            cursorsUnknown => $self->cursor_ids,
+        },
+    };
+
+    eval { $self->monitoring_callback->($event) };
+
+    return;
+}
+
 1;

t/monitoring_spec.t

@@ -354,6 +354,17 @@ sub check_reply_field {
         }
     }
 
+    # special case for $event.command.cursorsUnknown on killCursors
+    if ($event->{commandName} eq 'killCursors'
+        and defined $exp_reply->{cursorsUnknown}
+    ) {
+        for my $index (0 .. $#{ $exp_reply->{cursorsUnknown} }) {
+            $exp_reply->{cursorsUnknown}[$index]
+                = code(\&_verify_is_positive_num)
+                if $exp_reply->{cursorsUnknown}[$index] eq 42;
+        }
+    }
+
     for my $exp_key (sort keys %$exp_reply) {
         cmp_deeply
             $event_reply->{$exp_key},
@@ -381,6 +392,17 @@ sub check_command_field {
         $exp_command->{writeConcern}{wtimeout} = ignore();
     }
 
+    # special case for $event.command.cursors on killCursors
+    if ($event->{commandName} eq 'killCursors'
+        and defined $exp_command->{cursors}
+    ) {
+        for my $index (0 .. $#{ $exp_command->{cursors} }) {
+            $exp_command->{cursors}[$index]
+                = code(\&_verify_is_positive_num)
+                if $exp_command->{cursors}[$index] eq 42;
+        }
+    }
+
     for my $exp_key (sort keys %$exp_command) {
         my $event_value = $event_command->{$exp_key};
         my $exp_value = prepare_data_spec($exp_command->{$exp_key});