chore(core): Replace EventArray::for_each_X with an iterator (#13601)

The `EventArray` functions `for_each_X` are equivalent to composing an `iter_X` iterator with that
iterator's `for_each` method. Since the composable solution provides for more possible uses, drop
the `for_each_X` special cases in favour of new `iter_X_mut` iterator generators.

Author: bruceg

lib/vector-core/src/event/array.rs

@@ -137,42 +137,29 @@ pub enum EventArray {
 }
 
 impl EventArray {
-    /// Call the given update function over each `LogEvent` in this array.
-    pub fn for_each_log(&mut self, update: impl FnMut(&mut LogEvent)) {
-        if let Self::Logs(logs) = self {
-            logs.iter_mut().for_each(update);
-        }
-    }
-
-    /// Call the given update function over each `Metric` in this array.
-    pub fn for_each_metric(&mut self, update: impl FnMut(&mut Metric)) {
-        if let Self::Metrics(metrics) = self {
-            metrics.iter_mut().for_each(update);
-        }
-    }
-
-    /// Run the given update function over each `Trace` in this array.
-    pub fn for_each_trace(&mut self, update: impl FnMut(&mut TraceEvent)) {
-        if let Self::Traces(traces) = self {
-            traces.iter_mut().for_each(update);
+    /// Iterate over references to this array's events.
+    pub fn iter_events(&self) -> impl Iterator<Item = EventRef> {
+        match self {
+            Self::Logs(array) => EventArrayIter::Logs(array.iter()),
+            Self::Metrics(array) => EventArrayIter::Metrics(array.iter()),
+            Self::Traces(array) => EventArrayIter::Traces(array.iter()),
         }
     }
 
-    /// Call the given update function over each event in this array.
-    pub fn for_each_event(&mut self, mut update: impl FnMut(EventMutRef<'_>)) {
+    /// Iterate over mutable references to this array's events.
+    pub fn iter_events_mut(&mut self) -> impl Iterator<Item = EventMutRef> {
         match self {
-            Self::Logs(array) => array.iter_mut().for_each(|log| update(log.into())),
-            Self::Metrics(array) => array.iter_mut().for_each(|metric| update(metric.into())),
-            Self::Traces(array) => array.iter_mut().for_each(|trace| update(trace.into())),
+            Self::Logs(array) => EventArrayIterMut::Logs(array.iter_mut()),
+            Self::Metrics(array) => EventArrayIterMut::Metrics(array.iter_mut()),
+            Self::Traces(array) => EventArrayIterMut::Traces(array.iter_mut()),
         }
     }
 
-    /// Iterate over this array's events.
-    pub fn iter_events(&self) -> impl Iterator<Item = EventRef> {
+    /// Iterate over references to the logs in this array.
+    pub fn iter_logs_mut(&mut self) -> impl Iterator<Item = &mut LogEvent> {
         match self {
-            Self::Logs(array) => EventArrayIter::Logs(array.iter()),
-            Self::Metrics(array) => EventArrayIter::Metrics(array.iter()),
-            Self::Traces(array) => EventArrayIter::Traces(array.iter()),
+            Self::Logs(array) => TypedArrayIterMut(Some(array.iter_mut())),
+            _ => TypedArrayIterMut(None),
         }
     }
 }
@@ -348,6 +335,29 @@ impl<'a> Iterator for EventArrayIter<'a> {
     }
 }
 
+/// The iterator type for `EventArray::iter_events_mut`.
+#[derive(Debug)]
+pub enum EventArrayIterMut<'a> {
+    /// An iterator over type `LogEvent`.
+    Logs(slice::IterMut<'a, LogEvent>),
+    /// An iterator over type `Metric`.
+    Metrics(slice::IterMut<'a, Metric>),
+    /// An iterator over type `Trace`.
+    Traces(slice::IterMut<'a, TraceEvent>),
+}
+
+impl<'a> Iterator for EventArrayIterMut<'a> {
+    type Item = EventMutRef<'a>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self {
+            Self::Logs(i) => i.next().map(EventMutRef::from),
+            Self::Metrics(i) => i.next().map(EventMutRef::from),
+            Self::Traces(i) => i.next().map(EventMutRef::from),
+        }
+    }
+}
+
 /// The iterator type for `EventArray::into_events`.
 #[derive(Debug)]
 pub enum EventArrayIntoIter {
@@ -371,6 +381,15 @@ impl Iterator for EventArrayIntoIter {
     }
 }
 
+struct TypedArrayIterMut<'a, T>(Option<slice::IterMut<'a, T>>);
+
+impl<'a, T> Iterator for TypedArrayIterMut<'a, T> {
+    type Item = &'a mut T;
+    fn next(&mut self) -> Option<Self::Item> {
+        self.0.as_mut().and_then(Iterator::next)
+    }
+}
+
 /// Intermediate buffer for conversion of a sequence of individual
 /// `Event`s into a sequence of `EventArray`s by coalescing contiguous
 /// events of the same type into one array. This is used by

src/sinks/aws_cloudwatch_logs/integration_tests.rs

@@ -119,7 +119,7 @@ async fn cloudwatch_insert_log_events_sorted() {
         if doit {
             let timestamp = chrono::Utc::now() - chrono::Duration::days(1);
 
-            events.for_each_log(|log| {
+            events.iter_logs_mut().for_each(|log| {
                 log.insert(log_schema().timestamp_key(), Value::Timestamp(timestamp));
             });
         }

src/sinks/datadog/events/tests.rs

@@ -31,7 +31,7 @@ fn random_events_with_stream(
     (
         lines,
         stream.map(|mut events| {
-            events.for_each_log(|log| {
+            events.iter_logs_mut().for_each(|log| {
                 log.insert("title", "All!");
                 log.insert("invalid", "Tik");
             });
@@ -128,7 +128,7 @@ async fn api_key_in_metadata() {
     let (expected, events) = random_events_with_stream(100, 10, None);
 
     let events = events.map(|mut events| {
-        events.for_each_log(|log| {
+        events.iter_logs_mut().for_each(|log| {
             log.metadata_mut()
                 .set_datadog_api_key(Arc::from("from_metadata"));
         });

src/sinks/datadog/logs/tests.rs

@@ -220,7 +220,7 @@ async fn api_key_in_metadata_inner(api_status: ApiStatus) {
     let api_key = "0xDECAFBAD";
     let events = events.map(|mut e| {
         println!("EVENT: {:?}", e);
-        e.for_each_log(|log| {
+        e.iter_logs_mut().for_each(|log| {
             log.metadata_mut().set_datadog_api_key(Arc::from(api_key));
         });
         e
@@ -360,7 +360,7 @@ async fn enterprise_headers_inner(api_status: ApiStatus) {
     let api_key = "0xDECAFBAD";
     let events = events.map(|mut e| {
         println!("EVENT: {:?}", e);
-        e.for_each_log(|log| {
+        e.iter_logs_mut().for_each(|log| {
             log.metadata_mut().set_datadog_api_key(Arc::from(api_key));
         });
         e
@@ -423,7 +423,7 @@ async fn no_enterprise_headers_inner(api_status: ApiStatus) {
     let api_key = "0xDECAFBAD";
     let events = events.map(|mut e| {
         println!("EVENT: {:?}", e);
-        e.for_each_log(|log| {
+        e.iter_logs_mut().for_each(|log| {
             log.metadata_mut().set_datadog_api_key(Arc::from(api_key));
         });
         e

src/sinks/elasticsearch/integration_tests.rs

@@ -394,7 +394,7 @@ async fn run_insert_tests_with_config(
         let mut doit = false;
         let events = events.map(move |mut events| {
             if doit {
-                events.for_each_log(|log| {
+                events.iter_logs_mut().for_each(|log| {
                     log.insert("_type", 1);
                 });
             }

src/sinks/kafka/tests.rs

@@ -271,7 +271,7 @@ mod integration_test {
                 header_1_key.to_owned(),
                 Value::Bytes(Bytes::from(header_1_value)),
             );
-            events.for_each_log(move |log| {
+            events.iter_logs_mut().for_each(move |log| {
                 log.insert(headers_key.as_str(), header_values.clone());
             });
             events

src/source_sender/mod.rs

@@ -102,7 +102,7 @@ impl SourceSender {
         // events, so we have to add a map to the receiver to handle the
         // finalization.
         let recv = recv.into_stream().flat_map(move |mut events| {
-            events.for_each_event(|mut event| {
+            events.iter_events_mut().for_each(|mut event| {
                 let metadata = event.metadata_mut();
                 metadata.update_status(status);
                 metadata.update_sources();
@@ -126,7 +126,7 @@ impl SourceSender {
                 EventStatus::Delivered
             };
             count += 1;
-            events.for_each_event(|mut event| {
+            events.iter_events_mut().for_each(|mut event| {
                 let metadata = event.metadata_mut();
                 metadata.update_status(status);
                 metadata.update_sources();
@@ -144,7 +144,7 @@ impl SourceSender {
     ) -> impl Stream<Item = EventArray> + Unpin {
         let (inner, recv) = Inner::new_with_buffer(100, name.clone());
         let recv = recv.into_stream().map(move |mut events| {
-            events.for_each_event(|mut event| {
+            events.iter_events_mut().for_each(|mut event| {
                 let metadata = event.metadata_mut();
                 metadata.update_status(status);
                 metadata.update_sources();