diff --git a/src/DotPulsar/Abstractions/IConsumer.cs b/src/DotPulsar/Abstractions/IConsumer.cs
index 48fe671..7fb4e16 100644
--- a/src/DotPulsar/Abstractions/IConsumer.cs
+++ b/src/DotPulsar/Abstractions/IConsumer.cs
@@ -22,97 +22,19 @@
     /// <summary>
     /// A consumer abstraction.
     /// </summary>
-    public interface IConsumer : IAsyncDisposable
+    public interface IConsumer : IGetLastMessageId, IReceive, ISeek, IState<ConsumerState>, IAsyncDisposable
     {
         /// <summary>
-        /// Acknowledge the consumption of a single message.
-        /// </summary>
-        ValueTask Acknowledge(Message message, CancellationToken cancellationToken = default);
-
-        /// <summary>
         /// Acknowledge the consumption of a single message using the MessageId.
         /// </summary>
         ValueTask Acknowledge(MessageId messageId, CancellationToken cancellationToken = default);
 
         /// <summary>
-        /// Acknowledge the consumption of all the messages in the topic up to and including the provided message.
-        /// </summary>
-        ValueTask AcknowledgeCumulative(Message message, CancellationToken cancellationToken = default);
-
-        /// <summary>
         /// Acknowledge the consumption of all the messages in the topic up to and including the provided MessageId.
         /// </summary>
         ValueTask AcknowledgeCumulative(MessageId messageId, CancellationToken cancellationToken = default);
 
         /// <summary>
-        /// Get the MessageId of the last message on the topic.
-        /// </summary>
-        ValueTask<MessageId> GetLastMessageId(CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Ask whether the current state is final, meaning that it will never change.
-        /// </summary>
-        /// <returns>
-        /// True if it's final and False if it's not.
-        /// </returns>
-        bool IsFinalState();
-
-        /// <summary>
-        /// Ask whether the provided state is final, meaning that it will never change.
-        /// </summary>
-        /// <returns>
-        /// True if it's final and False if it's not.
-        /// </returns>
-        bool IsFinalState(ConsumerState state);
-
-        /// <summary>
-        /// Get an IAsyncEnumerable for consuming messages.
-        /// </summary>
-        IAsyncEnumerable<Message> Messages(CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this consumer to a specific MessageId.
-        /// </summary>
-        ValueTask Seek(MessageId messageId, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this consumer to a specific message publish time using unix time in milliseconds.
-        /// </summary>
-        ValueTask Seek(ulong publishTime, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this consumer to a specific message publish time using an UTC DateTime.
-        /// </summary>
-        ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this consumer to a specific message publish time using a DateTimeOffset.
-        /// </summary>
-        ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Wait for the state to change to a specific state.
-        /// </summary>
-        /// <returns>
-        /// The current state.
-        /// </returns>
-        /// <remarks>
-        /// If the state change to a final state, then all awaiting tasks will complete.
-        /// </remarks>
-        ValueTask<ConsumerStateChanged> StateChangedTo(ConsumerState state, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Wait for the state to change from a specific state.
-        /// </summary>
-        /// <returns>
-        /// The current state.
-        /// </returns>
-        /// <remarks>
-        /// If the state change to a final state, then all awaiting tasks will complete.
-        /// </remarks>
-        ValueTask<ConsumerStateChanged> StateChangedFrom(ConsumerState state, CancellationToken cancellationToken = default);
-
-        /// <summary>
         /// The topic of the consumer.
         /// </summary>
         string Topic { get; }
@@ -125,11 +47,11 @@
         /// <summary>
         /// Redeliver the pending messages that were pushed to this consumer that are not yet acknowledged.
         /// </summary>
-        ValueTask RedeliverUnacknowledgedMessages(IEnumerable<MessageId> messageIds, CancellationToken cancellationToken);
+        ValueTask RedeliverUnacknowledgedMessages(IEnumerable<MessageId> messageIds, CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Redeliver all pending messages that were pushed to this consumer that are not yet acknowledged.
         /// </summary>
-        ValueTask RedeliverUnacknowledgedMessages(CancellationToken cancellationToken);
+        ValueTask RedeliverUnacknowledgedMessages(CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/DotPulsar/Internal/Abstractions/IReaderChannelFactory.cs b/src/DotPulsar/Abstractions/IGetLastMessageId.cs
similarity index 63%
copy from src/DotPulsar/Internal/Abstractions/IReaderChannelFactory.cs
copy to src/DotPulsar/Abstractions/IGetLastMessageId.cs
index c651dbf..0c3059a 100644
--- a/src/DotPulsar/Internal/Abstractions/IReaderChannelFactory.cs
+++ b/src/DotPulsar/Abstractions/IGetLastMessageId.cs
@@ -12,13 +12,19 @@
  * limitations under the License.
  */
 
-namespace DotPulsar.Internal.Abstractions
+namespace DotPulsar.Abstractions
 {
     using System.Threading;
     using System.Threading.Tasks;
 
-    public interface IReaderChannelFactory
+    /// <summary>
+    /// An abstraction for getting the last message id.
+    /// </summary>
+    public interface IGetLastMessageId
     {
-        Task<IReaderChannel> Create(CancellationToken cancellationToken = default);
+        /// <summary>
+        /// Get the MessageId of the last message on the topic.
+        /// </summary>
+        ValueTask<MessageId> GetLastMessageId(CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/DotPulsar/Abstractions/IHandleStateChanged.cs b/src/DotPulsar/Abstractions/IHandleStateChanged.cs
index 04d6b60..8be3a99 100644
--- a/src/DotPulsar/Abstractions/IHandleStateChanged.cs
+++ b/src/DotPulsar/Abstractions/IHandleStateChanged.cs
@@ -25,7 +25,7 @@
         /// <summary>
         /// Called after a state has changed.
         /// </summary>
-        ValueTask OnStateChanged(TStateChanged stateChanged, CancellationToken cancellationToken);
+        ValueTask OnStateChanged(TStateChanged stateChanged, CancellationToken cancellationToken = default);
 
         /// <summary>
         /// The cancellation token to use when waiting for and handling state changes.
diff --git a/src/DotPulsar/Abstractions/IProducer.cs b/src/DotPulsar/Abstractions/IProducer.cs
index 03ff03e..27c59ee 100644
--- a/src/DotPulsar/Abstractions/IProducer.cs
+++ b/src/DotPulsar/Abstractions/IProducer.cs
@@ -15,84 +15,13 @@
 namespace DotPulsar.Abstractions
 {
     using System;
-    using System.Buffers;
-    using System.Threading;
-    using System.Threading.Tasks;
 
     /// <summary>
     /// A producer abstraction.
     /// </summary>
-    public interface IProducer : IAsyncDisposable
+    public interface IProducer : ISend, IState<ProducerState>, IAsyncDisposable
     {
         /// <summary>
-        /// Ask whether the current state is final, meaning that it will never change.
-        /// </summary>
-        /// <returns>
-        /// True if it's final and False if it's not.
-        /// </returns>
-        bool IsFinalState();
-
-        /// <summary>
-        /// Ask whether the provided state is final, meaning that it will never change.
-        /// </summary>
-        /// <returns>
-        /// True if it's final and False if it's not.
-        /// </returns>
-        bool IsFinalState(ProducerState state);
-
-        /// <summary>
-        /// Sends a message.
-        /// </summary>
-        ValueTask<MessageId> Send(byte[] data, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Sends a message.
-        /// </summary>
-        ValueTask<MessageId> Send(ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Sends a message.
-        /// </summary>
-        ValueTask<MessageId> Send(ReadOnlySequence<byte> data, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Sends a message with metadata.
-        /// </summary>
-        ValueTask<MessageId> Send(MessageMetadata metadata, byte[] data, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Sends a message with metadata.
-        /// </summary>
-        ValueTask<MessageId> Send(MessageMetadata metadata, ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Sends a message with metadata.
-        /// </summary>
-        ValueTask<MessageId> Send(MessageMetadata metadata, ReadOnlySequence<byte> data, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Wait for the state to change to a specific state.
-        /// </summary>
-        /// <returns>
-        /// The current state.
-        /// </returns>
-        /// <remarks>
-        /// If the state change to a final state, then all awaiting tasks will complete.
-        /// </remarks>
-        ValueTask<ProducerStateChanged> StateChangedTo(ProducerState state, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Wait for the state to change from a specific state.
-        /// </summary>
-        /// <returns>
-        /// The current state.
-        /// </returns>
-        /// <remarks>
-        /// If the state change to a final state, then all awaiting tasks will complete.
-        /// </remarks>
-        ValueTask<ProducerStateChanged> StateChangedFrom(ProducerState state, CancellationToken cancellationToken = default);
-
-        /// <summary>
         /// The topic of the producer.
         /// </summary>
         string Topic { get; }
diff --git a/src/DotPulsar/Abstractions/IReader.cs b/src/DotPulsar/Abstractions/IReader.cs
index 1023419..6b4207b 100644
--- a/src/DotPulsar/Abstractions/IReader.cs
+++ b/src/DotPulsar/Abstractions/IReader.cs
@@ -15,84 +15,13 @@
 namespace DotPulsar.Abstractions
 {
     using System;
-    using System.Collections.Generic;
-    using System.Threading;
-    using System.Threading.Tasks;
 
     /// <summary>
     /// A reader abstraction.
     /// </summary>
-    public interface IReader : IAsyncDisposable
+    public interface IReader : IGetLastMessageId, IReceive, ISeek, IState<ReaderState>, IAsyncDisposable
     {
         /// <summary>
-        /// Ask whether the current state is final, meaning that it will never change.
-        /// </summary>
-        /// <returns>
-        /// True if it's final and False if it's not.
-        /// </returns>
-        bool IsFinalState();
-
-        /// <summary>
-        /// Ask whether the provided state is final, meaning that it will never change.
-        /// </summary>
-        /// <returns>
-        /// True if it's final and False if it's not.
-        /// </returns>
-        bool IsFinalState(ReaderState state);
-
-        /// <summary>
-        /// Get the MessageId of the last message on the topic.
-        /// </summary>
-        ValueTask<MessageId> GetLastMessageId(CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Get an IAsyncEnumerable for reading messages
-        /// </summary>
-        IAsyncEnumerable<Message> Messages(CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this reader to a specific MessageId.
-        /// </summary>
-        ValueTask Seek(MessageId messageId, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this reader to a specific message publish time using unix time in milliseconds.
-        /// </summary>
-        ValueTask Seek(ulong publishTime, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this reader to a specific message publish time using an UTC DateTime.
-        /// </summary>
-        ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Reset the subscription associated with this reader to a specific message publish time using a DateTimeOffset.
-        /// </summary>
-        ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Wait for the state to change to a specific state.
-        /// </summary>
-        /// <returns>
-        /// The current state.
-        /// </returns>
-        /// <remarks>
-        /// If the state change to a final state, then all awaiting tasks will complete.
-        /// </remarks>
-        ValueTask<ReaderStateChanged> StateChangedTo(ReaderState state, CancellationToken cancellationToken = default);
-
-        /// <summary>
-        /// Wait for the state to change from a specific state.
-        /// </summary>
-        /// <returns>
-        /// The current state.
-        /// </returns>
-        /// <remarks>
-        /// If the state change to a final state, then all awaiting tasks will complete.
-        /// </remarks>
-        ValueTask<ReaderStateChanged> StateChangedFrom(ReaderState state, CancellationToken cancellationToken = default);
-
-        /// <summary>
         /// The topic of the reader.
         /// </summary>
         string Topic { get; }
diff --git a/src/DotPulsar/Internal/Abstractions/IReaderChannelFactory.cs b/src/DotPulsar/Abstractions/IReceive.cs
similarity index 66%
rename from src/DotPulsar/Internal/Abstractions/IReaderChannelFactory.cs
rename to src/DotPulsar/Abstractions/IReceive.cs
index c651dbf..2ff275c 100644
--- a/src/DotPulsar/Internal/Abstractions/IReaderChannelFactory.cs
+++ b/src/DotPulsar/Abstractions/IReceive.cs
@@ -12,13 +12,19 @@
  * limitations under the License.
  */
 
-namespace DotPulsar.Internal.Abstractions
+namespace DotPulsar.Abstractions
 {
     using System.Threading;
     using System.Threading.Tasks;
 
-    public interface IReaderChannelFactory
+    /// <summary>
+    /// An abstraction for receiving a single message.
+    /// </summary>
+    public interface IReceive
     {
-        Task<IReaderChannel> Create(CancellationToken cancellationToken = default);
+        /// <summary>
+        /// Receive a single message.
+        /// </summary>
+        ValueTask<Message> Receive(CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/DotPulsar/Abstractions/ISeek.cs b/src/DotPulsar/Abstractions/ISeek.cs
new file mode 100644
index 0000000..373c619
--- /dev/null
+++ b/src/DotPulsar/Abstractions/ISeek.cs
@@ -0,0 +1,35 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Abstractions
+{
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// An abstraction for seeking.
+    /// </summary>
+    public interface ISeek
+    {
+        /// <summary>
+        /// Reset the cursor associated with the consumer or reader to a specific MessageId.
+        /// </summary>
+        ValueTask Seek(MessageId messageId, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the cursor associated with the consumer or reader to a specific message publish time using unix time in milliseconds.
+        /// </summary>
+        ValueTask Seek(ulong publishTime, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/ISend.cs b/src/DotPulsar/Abstractions/ISend.cs
new file mode 100644
index 0000000..2948c34
--- /dev/null
+++ b/src/DotPulsar/Abstractions/ISend.cs
@@ -0,0 +1,36 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Abstractions
+{
+    using System.Buffers;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// An abstraction for sending a message.
+    /// </summary>
+    public interface ISend
+    {
+        /// <summary>
+        /// Sends a message.
+        /// </summary>
+        ValueTask<MessageId> Send(ReadOnlySequence<byte> data, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Sends a message with metadata.
+        /// </summary>
+        ValueTask<MessageId> Send(MessageMetadata metadata, ReadOnlySequence<byte> data, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IState.cs b/src/DotPulsar/Abstractions/IState.cs
new file mode 100644
index 0000000..48a975e
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IState.cs
@@ -0,0 +1,63 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Abstractions
+{
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// A state change monitoring abstraction.
+    /// </summary>
+    public interface IState<TState> where TState : notnull
+    {
+        /// <summary>
+        /// Ask whether the current state is final, meaning that it will never change.
+        /// </summary>
+        /// <returns>
+        /// True if it's final and False if it's not.
+        /// </returns>
+        bool IsFinalState();
+
+        /// <summary>
+        /// Ask whether the provided state is final, meaning that it will never change.
+        /// </summary>
+        /// <returns>
+        /// True if it's final and False if it's not.
+        /// </returns>
+        bool IsFinalState(TState state);
+
+        /// <summary>
+        /// Wait for the state to change to a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        ValueTask<TState> OnStateChangeTo(TState state, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Wait for the state to change from a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        ValueTask<TState> OnStateChangeFrom(TState state, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Extensions/ConsumerExtensions.cs b/src/DotPulsar/Extensions/ConsumerExtensions.cs
new file mode 100644
index 0000000..b61971c
--- /dev/null
+++ b/src/DotPulsar/Extensions/ConsumerExtensions.cs
@@ -0,0 +1,68 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Extensions
+{
+    using DotPulsar.Abstractions;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Extensions for IConsumer.
+    /// </summary>
+    public static class ConsumerExtensions
+    {
+        /// <summary>
+        /// Acknowledge the consumption of a single message.
+        /// </summary>
+        public static async ValueTask Acknowledge(this IConsumer consumer, Message message, CancellationToken cancellationToken = default)
+            => await consumer.Acknowledge(message.MessageId, cancellationToken).ConfigureAwait(false);
+
+        /// <summary>
+        /// Acknowledge the consumption of all the messages in the topic up to and including the provided message.
+        /// </summary>
+        public static async ValueTask AcknowledgeCumulative(this IConsumer consumer, Message message, CancellationToken cancellationToken = default)
+            => await consumer.AcknowledgeCumulative(message.MessageId, cancellationToken).ConfigureAwait(false);
+
+        /// <summary>
+        /// Wait for the state to change to a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        public static async ValueTask<ConsumerStateChanged> StateChangedTo(this IConsumer consumer, ConsumerState state, CancellationToken cancellationToken = default)
+        {
+            var newState = await consumer.OnStateChangeTo(state, cancellationToken).ConfigureAwait(false);
+            return new ConsumerStateChanged(consumer, newState);
+        }
+
+        /// <summary>
+        /// Wait for the state to change from a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        public static async ValueTask<ConsumerStateChanged> StateChangedFrom(this IConsumer consumer, ConsumerState state, CancellationToken cancellationToken = default)
+        {
+            var newState = await consumer.OnStateChangeFrom(state, cancellationToken).ConfigureAwait(false);
+            return new ConsumerStateChanged(consumer, newState);
+        }
+    }
+}
diff --git a/src/DotPulsar/Extensions/ProducerExtensions.cs b/src/DotPulsar/Extensions/ProducerExtensions.cs
index 9c3007c..8f02262 100644
--- a/src/DotPulsar/Extensions/ProducerExtensions.cs
+++ b/src/DotPulsar/Extensions/ProducerExtensions.cs
@@ -16,7 +16,12 @@
 {
     using Abstractions;
     using Internal;
+    using System.Threading;
+    using System.Threading.Tasks;
 
+    /// <summary>
+    /// Extensions for IProducer.
+    /// </summary>
     public static class ProducerExtensions
     {
         /// <summary>
@@ -24,5 +29,35 @@
         /// </summary>
         public static IMessageBuilder NewMessage(this IProducer producer)
             => new MessageBuilder(producer);
+
+        /// <summary>
+        /// Wait for the state to change to a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        public static async ValueTask<ProducerStateChanged> StateChangedTo(this IProducer producer, ProducerState state, CancellationToken cancellationToken = default)
+        {
+            var newState = await producer.OnStateChangeTo(state, cancellationToken).ConfigureAwait(false);
+            return new ProducerStateChanged(producer, newState);
+        }
+
+        /// <summary>
+        /// Wait for the state to change from a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        public static  async ValueTask<ProducerStateChanged> StateChangedFrom(this IProducer producer, ProducerState state, CancellationToken cancellationToken = default)
+        {
+            var newState = await producer.OnStateChangeFrom(state, cancellationToken).ConfigureAwait(false);
+            return new ProducerStateChanged(producer, newState);
+        }
     }
 }
diff --git a/src/DotPulsar/Extensions/PulsarClientExtensions.cs b/src/DotPulsar/Extensions/PulsarClientExtensions.cs
index 77e95aa..574bce1 100644
--- a/src/DotPulsar/Extensions/PulsarClientExtensions.cs
+++ b/src/DotPulsar/Extensions/PulsarClientExtensions.cs
@@ -17,6 +17,9 @@
     using Abstractions;
     using Internal;
 
+    /// <summary>
+    /// Extensions for IPulsarClient.
+    /// </summary>
     public static class PulsarClientExtensions
     {
         /// <summary>
diff --git a/src/DotPulsar/Extensions/ReaderExtensions.cs b/src/DotPulsar/Extensions/ReaderExtensions.cs
new file mode 100644
index 0000000..86d093b
--- /dev/null
+++ b/src/DotPulsar/Extensions/ReaderExtensions.cs
@@ -0,0 +1,56 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Extensions
+{
+    using DotPulsar.Abstractions;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Extensions for IReader.
+    /// </summary>
+    public static class ReaderExtensions
+    {
+        /// <summary>
+        /// Wait for the state to change to a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        public static async ValueTask<ReaderStateChanged> StateChangedTo(this IReader reader, ReaderState state, CancellationToken cancellationToken = default)
+        {
+            var newState = await reader.OnStateChangeTo(state, cancellationToken).ConfigureAwait(false);
+            return new ReaderStateChanged(reader, newState);
+        }
+
+        /// <summary>
+        /// Wait for the state to change from a specific state.
+        /// </summary>
+        /// <returns>
+        /// The current state.
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete.
+        /// </remarks>
+        public static async ValueTask<ReaderStateChanged> StateChangedFrom(this IReader reader, ReaderState state, CancellationToken cancellationToken = default)
+        {
+            var newState = await reader.OnStateChangeFrom(state, cancellationToken).ConfigureAwait(false);
+            return new ReaderStateChanged(reader, newState);
+        }
+    }
+}
diff --git a/src/DotPulsar/Extensions/ReceiveExtensions.cs b/src/DotPulsar/Extensions/ReceiveExtensions.cs
new file mode 100644
index 0000000..c2262c6
--- /dev/null
+++ b/src/DotPulsar/Extensions/ReceiveExtensions.cs
@@ -0,0 +1,36 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Extensions
+{
+    using DotPulsar.Abstractions;
+    using System.Collections.Generic;
+    using System.Runtime.CompilerServices;
+    using System.Threading;
+
+    /// <summary>
+    /// Extensions for IReceive.
+    /// </summary>
+    public static class ReceiveExtensions
+    {
+        /// <summary>
+        /// Get an IAsyncEnumerable for receiving messages.
+        /// </summary>
+        public static async IAsyncEnumerable<Message> Messages(this IReceive receiver, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+                yield return await receiver.Receive(cancellationToken).ConfigureAwait(false);
+        }
+    }
+}
diff --git a/src/DotPulsar/Extensions/SeekExtensions.cs b/src/DotPulsar/Extensions/SeekExtensions.cs
new file mode 100644
index 0000000..b4f0d2c
--- /dev/null
+++ b/src/DotPulsar/Extensions/SeekExtensions.cs
@@ -0,0 +1,39 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Extensions
+{
+    using DotPulsar.Abstractions;
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Extensions for ISeek.
+    /// </summary>
+    public static class SeekExtensions
+    {
+        /// <summary>
+        /// Reset the cursor associated with the consumer or reader to a specific message publish time using an UTC DateTime.
+        /// </summary>
+        public static async ValueTask Seek(this ISeek seeker, DateTime publishTime, CancellationToken cancellationToken = default)
+            => await seeker.Seek((ulong) new DateTimeOffset(publishTime).ToUnixTimeMilliseconds(), cancellationToken).ConfigureAwait(false);
+
+        /// <summary>
+        /// Reset the cursor associated with the consumer or reader to a specific message publish time using a DateTimeOffset.
+        /// </summary>
+        public static async ValueTask Seek(this ISeek seeker, DateTimeOffset publishTime, CancellationToken cancellationToken = default)
+            => await seeker.Seek((ulong) publishTime.ToUnixTimeMilliseconds(), cancellationToken).ConfigureAwait(false);
+    }
+}
diff --git a/src/DotPulsar/Extensions/SendExtensions.cs b/src/DotPulsar/Extensions/SendExtensions.cs
new file mode 100644
index 0000000..41ace1e
--- /dev/null
+++ b/src/DotPulsar/Extensions/SendExtensions.cs
@@ -0,0 +1,52 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Extensions
+{
+    using Abstractions;
+    using System;
+    using System.Buffers;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Extensions for ISend.
+    /// </summary>
+    public static class SendExtensions
+    {
+        /// <summary>
+        /// Sends a message.
+        /// </summary>
+        public static async ValueTask<MessageId> Send(this ISend sender, byte[] data, CancellationToken cancellationToken = default)
+            => await sender.Send(new ReadOnlySequence<byte>(data), cancellationToken).ConfigureAwait(false);
+
+        /// <summary>
+        /// Sends a message.
+        /// </summary>
+        public static async ValueTask<MessageId> Send(this ISend sender, ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default)
+            => await sender.Send(new ReadOnlySequence<byte>(data), cancellationToken).ConfigureAwait(false);
+
+        /// <summary>
+        /// Sends a message with metadata.
+        /// </summary>
+        public static async ValueTask<MessageId> Send(this ISend sender, MessageMetadata metadata, byte[] data, CancellationToken cancellationToken = default)
+            => await sender.Send(metadata, new ReadOnlySequence<byte>(data), cancellationToken).ConfigureAwait(false);
+
+        /// <summary>
+        /// Sends a message with metadata.
+        /// </summary>
+        public static async ValueTask<MessageId> Send(this ISend sender, MessageMetadata metadata, ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default)
+            => await sender.Send(metadata, new ReadOnlySequence<byte>(data), cancellationToken).ConfigureAwait(false);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs b/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs
index 31140b2..ef09431 100644
--- a/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs
+++ b/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs
@@ -23,9 +23,9 @@
     {
         Task Send(CommandAck command, CancellationToken cancellationToken);
         Task Send(CommandRedeliverUnacknowledgedMessages command, CancellationToken cancellationToken);
-        Task<CommandSuccess> Send(CommandUnsubscribe command, CancellationToken cancellationToken);
-        Task<CommandSuccess> Send(CommandSeek command, CancellationToken cancellationToken);
-        Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command, CancellationToken cancellationToken);
+        Task Send(CommandUnsubscribe command, CancellationToken cancellationToken);
+        Task Send(CommandSeek command, CancellationToken cancellationToken);
+        Task<MessageId> Send(CommandGetLastMessageId command, CancellationToken cancellationToken);
         ValueTask<Message> Receive(CancellationToken cancellationToken);
         ValueTask ClosedByClient(CancellationToken cancellationToken);
     }
diff --git a/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs b/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs
deleted file mode 100644
index 55135ea..0000000
--- a/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs
+++ /dev/null
@@ -1,29 +0,0 @@
-﻿/*
- * 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.
- */
-
-namespace DotPulsar.Internal.Abstractions
-{
-    using DotPulsar.Internal.PulsarApi;
-    using System;
-    using System.Threading;
-    using System.Threading.Tasks;
-
-    public interface IReaderChannel : IAsyncDisposable
-    {
-        Task<CommandSuccess> Send(CommandSeek command, CancellationToken cancellationToken);
-        Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command, CancellationToken cancellationToken);
-        ValueTask<Message> Receive(CancellationToken cancellationToken);
-        ValueTask ClosedByClient(CancellationToken cancellationToken);
-    }
-}
diff --git a/src/DotPulsar/Internal/Consumer.cs b/src/DotPulsar/Internal/Consumer.cs
index c6e9dfc..d263df2 100644
--- a/src/DotPulsar/Internal/Consumer.cs
+++ b/src/DotPulsar/Internal/Consumer.cs
@@ -17,13 +17,13 @@
     using Abstractions;
     using DotPulsar.Abstractions;
     using DotPulsar.Exceptions;
+    using DotPulsar.Internal.Extensions;
     using Events;
     using Microsoft.Extensions.ObjectPool;
     using PulsarApi;
     using System;
     using System.Collections.Generic;
     using System.Linq;
-    using System.Runtime.CompilerServices;
     using System.Threading;
     using System.Threading.Tasks;
 
@@ -59,17 +59,11 @@
             _eventRegister.Register(new ConsumerCreated(_correlationId, this));
         }
 
-        public async ValueTask<ConsumerStateChanged> StateChangedTo(ConsumerState state, CancellationToken cancellationToken)
-        {
-            var newState = await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false);
-            return new ConsumerStateChanged(this, newState);
-        }
+        public async ValueTask<ConsumerState> OnStateChangeTo(ConsumerState state, CancellationToken cancellationToken)
+            => await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false);
 
-        public async ValueTask<ConsumerStateChanged> StateChangedFrom(ConsumerState state, CancellationToken cancellationToken)
-        {
-            var newState = await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false);
-            return new ConsumerStateChanged(this, newState);
-        }
+        public async ValueTask<ConsumerState> OnStateChangeFrom(ConsumerState state, CancellationToken cancellationToken)
+            => await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false);
 
         public bool IsFinalState()
             => _state.IsFinalState();
@@ -87,35 +81,28 @@
             await _channel.DisposeAsync().ConfigureAwait(false);
         }
 
-        public async IAsyncEnumerable<Message> Messages([EnumeratorCancellation] CancellationToken cancellationToken)
+        public async ValueTask<Message> Receive(CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
-            while (!cancellationToken.IsCancellationRequested)
-                yield return await _executor.Execute(() => Receive(cancellationToken), cancellationToken).ConfigureAwait(false);
+            return await _executor.Execute(() => ReceiveMessage(cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
-        private async ValueTask<Message> Receive(CancellationToken cancellationToken)
+        private async ValueTask<Message> ReceiveMessage(CancellationToken cancellationToken)
             => await _channel.Receive(cancellationToken).ConfigureAwait(false);
 
-        public async ValueTask Acknowledge(Message message, CancellationToken cancellationToken)
-            => await Acknowledge(message.MessageId.Data, CommandAck.AckType.Individual, cancellationToken).ConfigureAwait(false);
-
         public async ValueTask Acknowledge(MessageId messageId, CancellationToken cancellationToken)
-            => await Acknowledge(messageId.Data, CommandAck.AckType.Individual, cancellationToken).ConfigureAwait(false);
-
-        public async ValueTask AcknowledgeCumulative(Message message, CancellationToken cancellationToken)
-            => await Acknowledge(message.MessageId.Data, CommandAck.AckType.Cumulative, cancellationToken).ConfigureAwait(false);
+            => await Acknowledge(messageId, CommandAck.AckType.Individual, cancellationToken).ConfigureAwait(false);
 
         public async ValueTask AcknowledgeCumulative(MessageId messageId, CancellationToken cancellationToken)
-            => await Acknowledge(messageId.Data, CommandAck.AckType.Cumulative, cancellationToken).ConfigureAwait(false);
+            => await Acknowledge(messageId, CommandAck.AckType.Cumulative, cancellationToken).ConfigureAwait(false);
 
         public async ValueTask RedeliverUnacknowledgedMessages(IEnumerable<MessageId> messageIds, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
             var command = new CommandRedeliverUnacknowledgedMessages();
-            command.MessageIds.AddRange(messageIds.Select(m => m.Data));
+            command.MessageIds.AddRange(messageIds.Select(messageId => messageId.ToMessageIdData()));
             await _executor.Execute(() => RedeliverUnacknowledgedMessages(command, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
@@ -131,16 +118,14 @@
         }
 
         private async ValueTask Unsubscribe(CommandUnsubscribe command, CancellationToken cancellationToken)
-        {
-            _ = await _channel.Send(command, cancellationToken).ConfigureAwait(false);
-        }
+            =>await _channel.Send(command, cancellationToken).ConfigureAwait(false);
 
         public async ValueTask Seek(MessageId messageId, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
-            var seek = new CommandSeek { MessageId = messageId.Data };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+            var seek = new CommandSeek { MessageId = messageId.ToMessageIdData() };
+            await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         public async ValueTask Seek(ulong publishTime, CancellationToken cancellationToken)
@@ -148,23 +133,7 @@
             ThrowIfDisposed();
 
             var seek = new CommandSeek { MessagePublishTime = publishTime };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
-        }
-
-        public async ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken)
-        {
-            ThrowIfDisposed();
-
-            var seek = new CommandSeek { MessagePublishTime = (ulong) new DateTimeOffset(publishTime).ToUnixTimeMilliseconds() };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
-        }
-
-        public async ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken)
-        {
-            ThrowIfDisposed();
-
-            var seek = new CommandSeek { MessagePublishTime = (ulong) publishTime.ToUnixTimeMilliseconds() };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+            await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         public async ValueTask<MessageId> GetLastMessageId(CancellationToken cancellationToken)
@@ -176,22 +145,21 @@
         }
 
         private async ValueTask<MessageId> GetLastMessageId(CommandGetLastMessageId command, CancellationToken cancellationToken)
-        {
-            var response = await _channel.Send(command, cancellationToken).ConfigureAwait(false);
-            return new MessageId(response.LastMessageId);
-        }
-
-        private async ValueTask<CommandSuccess> Seek(CommandSeek command, CancellationToken cancellationToken)
             => await _channel.Send(command, cancellationToken).ConfigureAwait(false);
 
-        private async ValueTask Acknowledge(MessageIdData messageIdData, CommandAck.AckType ackType, CancellationToken cancellationToken)
+        private async Task Seek(CommandSeek command, CancellationToken cancellationToken)
+            => await _channel.Send(command, cancellationToken).ConfigureAwait(false);
+
+        private async ValueTask Acknowledge(MessageId messageId, CommandAck.AckType ackType, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
             var commandAck = _commandAckPool.Get();
             commandAck.Type = ackType;
-            commandAck.MessageIds.Clear();
-            commandAck.MessageIds.Add(messageIdData);
+            if (commandAck.MessageIds.Count == 0)
+                commandAck.MessageIds.Add(messageId.ToMessageIdData());
+            else
+                commandAck.MessageIds[0].MapFrom(messageId);
 
             try
             {
diff --git a/src/DotPulsar/Internal/ConsumerChannel.cs b/src/DotPulsar/Internal/ConsumerChannel.cs
index 775e240..de109dd 100644
--- a/src/DotPulsar/Internal/ConsumerChannel.cs
+++ b/src/DotPulsar/Internal/ConsumerChannel.cs
@@ -21,7 +21,7 @@
     using System.Threading;
     using System.Threading.Tasks;
 
-    public sealed class ConsumerChannel : IConsumerChannel, IReaderChannel
+    public sealed class ConsumerChannel : IConsumerChannel
     {
         private readonly ulong _id;
         private readonly AsyncQueue<MessagePackage> _queue;
@@ -88,7 +88,7 @@
 
                     return metadata.ShouldSerializeNumMessagesInBatch()
                         ? _batchHandler.Add(messageId, redeliveryCount, metadata, data)
-                        : MessageFactory.Create(new MessageId(messageId), redeliveryCount, metadata, data);
+                        : MessageFactory.Create(messageId.ToMessageId(), redeliveryCount, metadata, data);
                 }
             }
         }
@@ -117,29 +117,27 @@
             await _connection.Send(command, cancellationToken).ConfigureAwait(false);
         }
 
-        public async Task<CommandSuccess> Send(CommandUnsubscribe command, CancellationToken cancellationToken)
+        public async Task Send(CommandUnsubscribe command, CancellationToken cancellationToken)
         {
             command.ConsumerId = _id;
             var response = await _connection.Send(command, cancellationToken).ConfigureAwait(false);
             response.Expect(BaseCommand.Type.Success);
-            return response.Success;
         }
 
-        public async Task<CommandSuccess> Send(CommandSeek command, CancellationToken cancellationToken)
+        public async Task Send(CommandSeek command, CancellationToken cancellationToken)
         {
             command.ConsumerId = _id;
             var response = await _connection.Send(command, cancellationToken).ConfigureAwait(false);
             response.Expect(BaseCommand.Type.Success);
             _batchHandler.Clear();
-            return response.Success;
         }
 
-        public async Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command, CancellationToken cancellationToken)
+        public async Task<MessageId> Send(CommandGetLastMessageId command, CancellationToken cancellationToken)
         {
             command.ConsumerId = _id;
             var response = await _connection.Send(command, cancellationToken).ConfigureAwait(false);
             response.Expect(BaseCommand.Type.GetLastMessageIdResponse);
-            return response.GetLastMessageIdResponse;
+            return response.GetLastMessageIdResponse.LastMessageId.ToMessageId();
         }
 
         public async ValueTask DisposeAsync()
diff --git a/src/DotPulsar/Internal/Extensions/MessageIdDataExtensions.cs b/src/DotPulsar/Internal/Extensions/MessageIdDataExtensions.cs
new file mode 100644
index 0000000..857b00c
--- /dev/null
+++ b/src/DotPulsar/Internal/Extensions/MessageIdDataExtensions.cs
@@ -0,0 +1,32 @@
+﻿/*
+ * 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.
+ */
+
+namespace DotPulsar.Internal.Extensions
+{
+    using DotPulsar.Internal.PulsarApi;
+
+    public static class MessageIdDataExtensions
+    {
+        public static MessageId ToMessageId(this MessageIdData messageIdData)
+            => new MessageId(messageIdData.LedgerId, messageIdData.EntryId, messageIdData.Partition, messageIdData.BatchIndex);
+
+        public static void MapFrom(this MessageIdData destination, MessageId source)
+        {
+            destination.LedgerId = source.LedgerId;
+            destination.EntryId = source.EntryId;
+            destination.Partition = source.Partition;
+            destination.BatchIndex = source.BatchIndex;
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/MessageBuilder.cs b/src/DotPulsar/Internal/MessageBuilder.cs
index d7c7b02..c17ec19 100644
--- a/src/DotPulsar/Internal/MessageBuilder.cs
+++ b/src/DotPulsar/Internal/MessageBuilder.cs
@@ -15,6 +15,7 @@
 namespace DotPulsar.Internal
 {
     using DotPulsar.Abstractions;
+    using DotPulsar.Extensions;
     using Extensions;
     using System;
     using System.Threading;
diff --git a/src/DotPulsar/Internal/MonitorState.cs b/src/DotPulsar/Internal/MonitorState.cs
index 28e0f9a..7a50044 100644
--- a/src/DotPulsar/Internal/MonitorState.cs
+++ b/src/DotPulsar/Internal/MonitorState.cs
@@ -15,6 +15,7 @@
 namespace DotPulsar.Internal
 {
     using DotPulsar.Abstractions;
+    using DotPulsar.Extensions;
     using System.Threading.Tasks;
 
     public static class StateMonitor
diff --git a/src/DotPulsar/Internal/NotReadyChannel.cs b/src/DotPulsar/Internal/NotReadyChannel.cs
index e463ff6..75e724f 100644
--- a/src/DotPulsar/Internal/NotReadyChannel.cs
+++ b/src/DotPulsar/Internal/NotReadyChannel.cs
@@ -22,7 +22,7 @@
     using System.Threading;
     using System.Threading.Tasks;
 
-    public sealed class NotReadyChannel : IConsumerChannel, IProducerChannel, IReaderChannel
+    public sealed class NotReadyChannel : IConsumerChannel, IProducerChannel
     {
         public ValueTask DisposeAsync()
             => new ValueTask();
@@ -42,13 +42,13 @@
         public Task Send(CommandRedeliverUnacknowledgedMessages command, CancellationToken cancellationToken)
             => throw GetException();
 
-        public Task<CommandSuccess> Send(CommandUnsubscribe command, CancellationToken cancellationToken)
+        public Task Send(CommandUnsubscribe command, CancellationToken cancellationToken)
             => throw GetException();
 
-        public Task<CommandSuccess> Send(CommandSeek command, CancellationToken cancellationToken)
+        public Task Send(CommandSeek command, CancellationToken cancellationToken)
             => throw GetException();
 
-        public Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command, CancellationToken cancellationToken)
+        public Task<MessageId> Send(CommandGetLastMessageId command, CancellationToken cancellationToken)
             => throw GetException();
 
         private static Exception GetException()
diff --git a/src/DotPulsar/Internal/Producer.cs b/src/DotPulsar/Internal/Producer.cs
index 2347eee..12e4625 100644
--- a/src/DotPulsar/Internal/Producer.cs
+++ b/src/DotPulsar/Internal/Producer.cs
@@ -17,6 +17,7 @@
     using Abstractions;
     using DotPulsar.Abstractions;
     using DotPulsar.Exceptions;
+    using DotPulsar.Internal.Extensions;
     using Events;
     using Microsoft.Extensions.ObjectPool;
     using System;
@@ -60,17 +61,11 @@
             _eventRegister.Register(new ProducerCreated(_correlationId, this));
         }
 
-        public async ValueTask<ProducerStateChanged> StateChangedTo(ProducerState state, CancellationToken cancellationToken)
-        {
-            var newState = await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false);
-            return new ProducerStateChanged(this, newState);
-        }
+        public async ValueTask<ProducerState> OnStateChangeTo(ProducerState state, CancellationToken cancellationToken)
+            => await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false);
 
-        public async ValueTask<ProducerStateChanged> StateChangedFrom(ProducerState state, CancellationToken cancellationToken)
-        {
-            var newState = await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false);
-            return new ProducerStateChanged(this, newState);
-        }
+        public async ValueTask<ProducerState> OnStateChangeFrom(ProducerState state, CancellationToken cancellationToken)
+            => await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false);
 
         public bool IsFinalState()
             => _state.IsFinalState();
@@ -88,12 +83,6 @@
             await _channel.DisposeAsync().ConfigureAwait(false);
         }
 
-        public ValueTask<MessageId> Send(byte[] data, CancellationToken cancellationToken)
-            => Send(new ReadOnlySequence<byte>(data), cancellationToken);
-
-        public ValueTask<MessageId> Send(ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
-            => Send(new ReadOnlySequence<byte>(data), cancellationToken);
-
         public async ValueTask<MessageId> Send(ReadOnlySequence<byte> data, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
@@ -110,12 +99,6 @@
             }
         }
 
-        public ValueTask<MessageId> Send(MessageMetadata metadata, byte[] data, CancellationToken cancellationToken)
-            => Send(metadata, new ReadOnlySequence<byte>(data), cancellationToken);
-
-        public ValueTask<MessageId> Send(MessageMetadata metadata, ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
-            => Send(metadata, new ReadOnlySequence<byte>(data), cancellationToken);
-
         public async ValueTask<MessageId> Send(MessageMetadata metadata, ReadOnlySequence<byte> data, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
@@ -138,7 +121,7 @@
         private async ValueTask<MessageId> Send(PulsarApi.MessageMetadata metadata, ReadOnlySequence<byte> data, CancellationToken cancellationToken)
         {
             var response = await _channel.Send(metadata, data, cancellationToken).ConfigureAwait(false);
-            return new MessageId(response.MessageId);
+            return response.MessageId.ToMessageId();
         }
 
         internal async ValueTask SetChannel(IProducerChannel channel)
diff --git a/src/DotPulsar/Internal/Reader.cs b/src/DotPulsar/Internal/Reader.cs
index 571c5b6..ccbea59 100644
--- a/src/DotPulsar/Internal/Reader.cs
+++ b/src/DotPulsar/Internal/Reader.cs
@@ -20,8 +20,6 @@
     using DotPulsar.Internal.PulsarApi;
     using Events;
     using System;
-    using System.Collections.Generic;
-    using System.Runtime.CompilerServices;
     using System.Threading;
     using System.Threading.Tasks;
 
@@ -29,7 +27,7 @@
     {
         private readonly Guid _correlationId;
         private readonly IRegisterEvent _eventRegister;
-        private IReaderChannel _channel;
+        private IConsumerChannel _channel;
         private readonly IExecute _executor;
         private readonly IStateChanged<ReaderState> _state;
         private int _isDisposed;
@@ -40,7 +38,7 @@
             Guid correlationId,
             string topic,
             IRegisterEvent eventRegister,
-            IReaderChannel initialChannel,
+            IConsumerChannel initialChannel,
             IExecute executor,
             IStateChanged<ReaderState> state)
         {
@@ -55,17 +53,11 @@
             _eventRegister.Register(new ReaderCreated(_correlationId, this));
         }
 
-        public async ValueTask<ReaderStateChanged> StateChangedTo(ReaderState state, CancellationToken cancellationToken)
-        {
-            var newState = await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false);
-            return new ReaderStateChanged(this, newState);
-        }
+        public async ValueTask<ReaderState> OnStateChangeTo(ReaderState state, CancellationToken cancellationToken)
+            => await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false);
 
-        public async ValueTask<ReaderStateChanged> StateChangedFrom(ReaderState state, CancellationToken cancellationToken)
-        {
-            var newState = await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false);
-            return new ReaderStateChanged(this, newState);
-        }
+        public async ValueTask<ReaderState> OnStateChangeFrom(ReaderState state, CancellationToken cancellationToken)
+            => await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false);
 
         public bool IsFinalState()
             => _state.IsFinalState();
@@ -82,28 +74,24 @@
         }
 
         private async ValueTask<MessageId> GetLastMessageId(CommandGetLastMessageId command, CancellationToken cancellationToken)
-        {
-            var response = await _channel.Send(command, cancellationToken).ConfigureAwait(false);
-            return new MessageId(response.LastMessageId);
-        }
+            => await _channel.Send(command, cancellationToken).ConfigureAwait(false);
 
-        public async IAsyncEnumerable<Message> Messages([EnumeratorCancellation] CancellationToken cancellationToken)
+        public async ValueTask<Message> Receive(CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
-            while (!cancellationToken.IsCancellationRequested)
-                yield return await _executor.Execute(() => Receive(cancellationToken), cancellationToken).ConfigureAwait(false);
+            return await _executor.Execute(() => ReceiveMessage(cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
-        private async ValueTask<Message> Receive(CancellationToken cancellationToken)
+        private async ValueTask<Message> ReceiveMessage(CancellationToken cancellationToken)
             => await _channel.Receive(cancellationToken).ConfigureAwait(false);
 
         public async ValueTask Seek(MessageId messageId, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
-            var seek = new CommandSeek { MessageId = messageId.Data };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+            var seek = new CommandSeek { MessageId = messageId.ToMessageIdData() };
+            await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         public async ValueTask Seek(ulong publishTime, CancellationToken cancellationToken)
@@ -111,23 +99,7 @@
             ThrowIfDisposed();
 
             var seek = new CommandSeek { MessagePublishTime = publishTime };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
-        }
-
-        public async ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken)
-        {
-            ThrowIfDisposed();
-
-            var seek = new CommandSeek { MessagePublishTime = (ulong) new DateTimeOffset(publishTime).ToUnixTimeMilliseconds() };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
-        }
-
-        public async ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken)
-        {
-            ThrowIfDisposed();
-
-            var seek = new CommandSeek { MessagePublishTime = (ulong) publishTime.ToUnixTimeMilliseconds() };
-            _ = await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+            await _executor.Execute(() => Seek(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         public async ValueTask DisposeAsync()
@@ -140,10 +112,10 @@
             await _channel.DisposeAsync().ConfigureAwait(false);
         }
 
-        private async ValueTask<CommandSuccess> Seek(CommandSeek command, CancellationToken cancellationToken)
+        private async Task Seek(CommandSeek command, CancellationToken cancellationToken)
             => await _channel.Send(command, cancellationToken).ConfigureAwait(false);
 
-        internal async ValueTask SetChannel(IReaderChannel channel)
+        internal async ValueTask SetChannel(IConsumerChannel channel)
         {
             if (_isDisposed != 0)
             {
diff --git a/src/DotPulsar/Internal/ReaderChannelFactory.cs b/src/DotPulsar/Internal/ReaderChannelFactory.cs
index 9efe033..58b31e1 100644
--- a/src/DotPulsar/Internal/ReaderChannelFactory.cs
+++ b/src/DotPulsar/Internal/ReaderChannelFactory.cs
@@ -20,7 +20,7 @@
     using System.Threading;
     using System.Threading.Tasks;
 
-    public sealed class ReaderChannelFactory : IReaderChannelFactory
+    public sealed class ReaderChannelFactory : IConsumerChannelFactory
     {
         private readonly Guid _correlationId;
         private readonly IRegisterEvent _eventRegister;
@@ -48,7 +48,7 @@
                 ConsumerName = options.ReaderName,
                 Durable = false,
                 ReadCompacted = options.ReadCompacted,
-                StartMessageId = options.StartMessageId.Data,
+                StartMessageId = options.StartMessageId.ToMessageIdData(),
                 Subscription = $"Reader-{Guid.NewGuid():N}",
                 Topic = options.Topic
             };
@@ -56,10 +56,10 @@
             _batchHandler = new BatchHandler(false);
         }
 
-        public async Task<IReaderChannel> Create(CancellationToken cancellationToken)
+        public async Task<IConsumerChannel> Create(CancellationToken cancellationToken)
             => await _executor.Execute(() => GetChannel(cancellationToken), cancellationToken).ConfigureAwait(false);
 
-        private async ValueTask<IReaderChannel> GetChannel(CancellationToken cancellationToken)
+        private async ValueTask<IConsumerChannel> GetChannel(CancellationToken cancellationToken)
         {
             var connection = await _connectionPool.FindConnectionForTopic(_subscribe.Topic, cancellationToken).ConfigureAwait(false);
             var messageQueue = new AsyncQueue<MessagePackage>();
diff --git a/src/DotPulsar/Internal/ReaderProcess.cs b/src/DotPulsar/Internal/ReaderProcess.cs
index 67763c5..404de7e 100644
--- a/src/DotPulsar/Internal/ReaderProcess.cs
+++ b/src/DotPulsar/Internal/ReaderProcess.cs
@@ -21,13 +21,13 @@
     public sealed class ReaderProcess : Process
     {
         private readonly IStateManager<ReaderState> _stateManager;
-        private readonly IReaderChannelFactory _factory;
+        private readonly IConsumerChannelFactory _factory;
         private readonly Reader _reader;
 
         public ReaderProcess(
             Guid correlationId,
             IStateManager<ReaderState> stateManager,
-            IReaderChannelFactory factory,
+            IConsumerChannelFactory factory,
             Reader reader) : base(correlationId)
         {
             _stateManager = stateManager;
diff --git a/src/DotPulsar/MessageId.cs b/src/DotPulsar/MessageId.cs
index cdcc2eb..bc32039 100644
--- a/src/DotPulsar/MessageId.cs
+++ b/src/DotPulsar/MessageId.cs
@@ -14,6 +14,7 @@
 
 namespace DotPulsar
 {
+    using DotPulsar.Internal.Extensions;
     using Internal.PulsarApi;
     using System;
 
@@ -38,42 +39,36 @@
         /// </summary>
         public static MessageId Latest { get; }
 
-        internal MessageId(MessageIdData messageIdData)
-            => Data = messageIdData;
-
         /// <summary>
         /// Initializes a new instance using the specified ledgerId, entryId, partition and batchIndex.
         /// </summary>
         public MessageId(ulong ledgerId, ulong entryId, int partition, int batchIndex)
-            => Data = new MessageIdData
-            {
-                LedgerId = ledgerId,
-                EntryId = entryId,
-                Partition = partition,
-                BatchIndex = batchIndex
-            };
-
-        internal MessageIdData Data { get; }
+        {
+            LedgerId = ledgerId;
+            EntryId = entryId;
+            Partition = partition;
+            BatchIndex = batchIndex;
+        }
 
         /// <summary>
         /// The id of the ledger.
         /// </summary>
-        public ulong LedgerId => Data.LedgerId;
+        public ulong LedgerId { get; }
 
         /// <summary>
         /// The id of the entry.
         /// </summary>
-        public ulong EntryId => Data.EntryId;
+        public ulong EntryId { get; }
 
         /// <summary>
         /// The partition.
         /// </summary>
-        public int Partition => Data.Partition;
+        public int Partition { get; }
 
         /// <summary>
         /// The batch index.
         /// </summary>
-        public int BatchIndex => Data.BatchIndex;
+        public int BatchIndex { get; }
 
         public int CompareTo(MessageId? other)
         {
@@ -124,5 +119,12 @@
 
         public override string ToString()
             => $"{LedgerId}:{EntryId}:{Partition}:{BatchIndex}";
+
+        internal MessageIdData ToMessageIdData()
+        {
+            var messageIdData = new MessageIdData();
+            messageIdData.MapFrom(this);
+            return messageIdData;
+        }
     }
 }