Getting offset and sequence number from event data rather than amqp annotations.

Author: j7nw4r

sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/partition_client.rs

@@ -127,58 +127,15 @@ impl PartitionClient {
     /// # Errors
     /// Returns an error if the sequence number or offset is invalid, or if updating the checkpoint fails.
     pub async fn update_checkpoint(&self, event_data: &ReceivedEventData) -> Result<()> {
-        let mut sequence_number: Option<i64> = None;
-        let mut offset: Option<String> = None;
-
-        let amqp_message = event_data.raw_amqp_message();
-        if let Some(message_annotations) = &amqp_message.message_annotations {
-            for (key, value) in message_annotations.0.iter() {
-                if *key == crate::consumer::SEQUENCE_NUMBER_ANNOTATION {
-                    match value {
-                        azure_core_amqp::AmqpValue::UInt(value) => {
-                            sequence_number = Some(*value as i64);
-                        }
-                        azure_core_amqp::AmqpValue::ULong(value) => {
-                            sequence_number = Some(*value as i64);
-                        }
-                        azure_core_amqp::AmqpValue::Long(value) => {
-                            sequence_number = Some(*value);
-                        }
-                        azure_core_amqp::AmqpValue::Int(value) => {
-                            sequence_number = Some(*value as i64);
-                        }
-                        _ => {
-                            return Err(azure_core::error::Error::with_message(
-                                azure_core::error::ErrorKind::Other,
-                                "Invalid sequence number",
-                            ));
-                        }
-                    }
-                } else if *key == crate::consumer::OFFSET_ANNOTATION {
-                    match value {
-                        azure_core_amqp::AmqpValue::String(value) => {
-                            offset = Some(value.to_string());
-                        }
-                        _ => {
-                            return Err(azure_core::error::Error::with_message(
-                                azure_core::error::ErrorKind::Other,
-                                "Invalid offset",
-                            ));
-                        }
-                    }
-                }
-            }
-            let checkpoint = Checkpoint {
-                fully_qualified_namespace: self.client_details.fully_qualified_namespace.clone(),
-                event_hub_name: self.client_details.eventhub_name.clone(),
-                consumer_group: self.client_details.consumer_group.clone(),
-                partition_id: self.partition_id.clone(),
-                offset,
-                sequence_number,
-            };
-            self.checkpoint_store.update_checkpoint(checkpoint).await?;
-        }
-        Ok(())
+        let checkpoint = Checkpoint {
+            fully_qualified_namespace: self.client_details.fully_qualified_namespace.clone(),
+            event_hub_name: self.client_details.eventhub_name.clone(),
+            consumer_group: self.client_details.consumer_group.clone(),
+            partition_id: self.partition_id.clone(),
+            offset: event_data.offset().clone(),
+            sequence_number: event_data.sequence_number().clone(),
+        };
+        self.checkpoint_store.update_checkpoint(checkpoint).await
     }
 
     pub(crate) fn set_event_receiver(&self, event_receiver: EventReceiver) -> Result<()> {