[Merged by Bors] - Fix some typo Event change_listener

crates/fluvio-sc/src/stores/spu.rs

@@ -65,7 +65,7 @@ mod health_check {
         }
 
         pub fn listener(&self) -> OffsetChangeListener {
-            self.event.change_listner()
+            self.event.change_listener()
         }
 
         /// update health check

crates/fluvio-spu/src/replication/follower/controller.rs

@@ -189,7 +189,7 @@ mod inner {
             let (mut sink, mut stream) = socket.split();
             let mut api_stream = stream.api_stream::<FollowerPeerRequest, FollowerPeerApiEnum>();
 
-            let mut event_listener = self.group.events.change_listner();
+            let mut event_listener = self.group.events.change_listener();
 
             // starts initial sync
             debug!("performing initial offset sync to leader");

crates/fluvio-spu/src/replication/leader/spu.rs

@@ -87,7 +87,7 @@ pub struct FollowerSpuPendingUpdates {
 
 impl FollowerSpuPendingUpdates {
     pub fn listener(&self) -> OffsetChangeListener {
-        self.event.change_listner()
+        self.event.change_listener()
     }
 
     ///  add replica to be updated

crates/fluvio-spu/src/services/public/stream_fetch.rs

@@ -61,7 +61,7 @@ impl StreamFetchHandler {
         if let Some(leader_state) = ctx.leaders_state().get(&replica).await {
             let (stream_id, offset_publisher) =
                 ctx.stream_publishers().create_new_publisher().await;
-            let consumer_offset_listener = offset_publisher.change_listner();
+            let consumer_offset_listener = offset_publisher.change_listener();
 
             spawn(async move {
                 if let Err(err) = StreamFetchHandler::fetch(

crates/fluvio-spu/src/storage/mod.rs

@@ -79,8 +79,8 @@ where
     /// listen to offset based on isolation
     pub fn offset_listener(&self, isolation: &Isolation) -> OffsetChangeListener {
         match isolation {
-            Isolation::ReadCommitted => self.hw.change_listner(),
-            Isolation::ReadUncommitted => self.leo.change_listner(),
+            Isolation::ReadCommitted => self.hw.change_listener(),
+            Isolation::ReadUncommitted => self.leo.change_listener(),
         }
     }
 

crates/fluvio-types/src/event.rs

@@ -102,7 +102,12 @@ pub mod offsets {
             self.update(self.current_value() + 1);
         }
 
+        #[deprecated = "Replace by change_listener"]
         pub fn change_listner(self: &Arc<Self>) -> OffsetChangeListener {
+            self.change_listener()
+        }
+
+        pub fn change_listener(self: &Arc<Self>) -> OffsetChangeListener {
             OffsetChangeListener::new(self.clone())
         }
     }
@@ -238,7 +243,7 @@ mod test {
     #[fluvio_future::test]
     async fn test_offset_listener_no_wait() {
         let publisher = OffsetPublisher::shared(0);
-        let listener = publisher.change_listner();
+        let listener = publisher.change_listener();
         let status = Arc::new(AtomicBool::new(false));
 
         TestController::start(listener, status.clone());
@@ -265,7 +270,7 @@ mod test {
     #[fluvio_future::test]
     async fn test_offset_listener_wait() {
         let publisher = OffsetPublisher::shared(0);
-        let listener = publisher.change_listner();
+        let listener = publisher.change_listener();
         let status = Arc::new(AtomicBool::new(false));
 
         TestController::start(listener, status.clone());

crates/fluvio/src/consumer.rs

@@ -359,7 +359,7 @@ where
                 );
 
                 let publisher = OffsetPublisher::shared(0);
-                let mut listener = publisher.change_listner();
+                let mut listener = publisher.change_listener();
 
                 // update stream with received offsets
                 spawn(async move {