WIP: A couple of more ChannelMojo speedup ideas

third_party/mojo/src/mojo/edk/system/local_message_pipe_endpoint.cc

Index: third_party/mojo/src/mojo/edk/system/local_message_pipe_endpoint.cc
diff --git a/third_party/mojo/src/mojo/edk/system/local_message_pipe_endpoint.cc b/third_party/mojo/src/mojo/edk/system/local_message_pipe_endpoint.cc
index 1800aa48db7d49989bdeccfcfd9a107d7a739f6b..0f93efd4359fe26897bef4d669f102c23f26c68e 100644
--- a/third_party/mojo/src/mojo/edk/system/local_message_pipe_endpoint.cc
+++ b/third_party/mojo/src/mojo/edk/system/local_message_pipe_endpoint.cc
@@ -54,6 +54,17 @@ void LocalMessagePipeEndpoint::EnqueueMessage(
     awakable_list_.AwakeForStateChange(GetHandleSignalsState());
 }
 
+void LocalMessagePipeEndpoint::OnReadMessage(
+    MessageInTransit::ReadContext& reading_message) {
+  if (!message_callback_.is_null() && !reading_message.has_dispatchers()) {
+    if (message_callback_.Run(reading_message.view().bytes(),
+                              reading_message.view().num_bytes()))
+      return;
+  }
+
+  MessagePipeEndpoint::OnReadMessage(reading_message);
+}
+
 void LocalMessagePipeEndpoint::Close() {
   DCHECK(is_open_);
   is_open_ = false;
@@ -179,5 +190,11 @@ void LocalMessagePipeEndpoint::RemoveAwakable(
     *signals_state = GetHandleSignalsState();
 }
 
+MojoResult LocalMessagePipeEndpoint::SetAsyncMessageCallback(
+    const AsyncMessageCallback& callback) {
+  message_callback_ = callback;
+  return MOJO_RESULT_OK;
+}
+
 }  // namespace system
 }  // namespace mojo