libmojo: Uprev the library to r456626 from Chromium

Pulled the latest and greatest version of libmojo from Chromium.

The merge was done against r456626 which corresponds to git commit
08266b3fca707804065a2cfd60331722ade41969 of Mar 14, 2017

Notable changes are:
- generated binding files are now split in 2 files, interface.mojom.h
  and interface.mojom-shared.h

Change-Id: Idcfd27310e2c9d3c452b671c7ff7a755c3963618

62 files changed, 2488 insertions, 2260 deletions

diff --git a/mojo/public/cpp/bindings/lib/array_internal.h b/mojo/public/cpp/bindings/lib/array_internal.h
index ba6d16e..eecfcfb 100644
--- a/mojo/public/cpp/bindings/lib/array_internal.h
+++ b/mojo/public/cpp/bindings/lib/array_internal.h
@@ -13,6 +13,7 @@
 
 #include "base/logging.h"
 #include "mojo/public/c/system/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
 #include "mojo/public/cpp/bindings/lib/buffer.h"
 #include "mojo/public/cpp/bindings/lib/serialization_util.h"
@@ -28,13 +29,13 @@ namespace internal {
 template <typename K, typename V>
 class Map_Data;
 
-std::string MakeMessageWithArrayIndex(const char* message,
-                                      size_t size,
-                                      size_t index);
+MOJO_CPP_BINDINGS_EXPORT std::string
+MakeMessageWithArrayIndex(const char* message, size_t size, size_t index);
 
-std::string MakeMessageWithExpectedArraySize(const char* message,
-                                             size_t size,
-                                             size_t expected_size);
+MOJO_CPP_BINDINGS_EXPORT std::string MakeMessageWithExpectedArraySize(
+    const char* message,
+    size_t size,
+    size_t expected_size);
 
 template <typename T>
 struct ArrayDataTraits {
@@ -67,7 +68,7 @@ template <>
 struct ArrayDataTraits<bool> {
   // Helper class to emulate a reference to a bool, used for direct element
   // access.
-  class BitRef {
+  class MOJO_CPP_BINDINGS_EXPORT BitRef {
    public:
     ~BitRef();
     BitRef& operator=(bool value);
@@ -109,7 +110,7 @@ struct ArrayDataTraits<bool> {
 //
 // TODO(yzshen): Validation code should be organzied in a way similar to
 // Serializer<>, or merged into it. It should be templatized with the mojo
-// wrapper type instead of the data type, that way we can use MojomTypeTraits
+// data view type instead of the data type, that way we can use MojomTypeTraits
 // to determine the categories.
 
 template <typename T, bool is_union, bool is_handle_or_interface>
@@ -262,7 +263,7 @@ class Array_Data {
       T,
       IsUnionDataType<T>::value,
       std::is_same<T, AssociatedInterface_Data>::value ||
-          std::is_same<T, AssociatedInterfaceRequest_Data>::value ||
+          std::is_same<T, AssociatedEndpointHandle_Data>::value ||
           std::is_same<T, Interface_Data>::value ||
           std::is_same<T, Handle_Data>::value>;
   using Element = T;
diff --git a/mojo/public/cpp/bindings/lib/array_serialization.h b/mojo/public/cpp/bindings/lib/array_serialization.h
index 5db27a5..d2f8ecf 100644
--- a/mojo/public/cpp/bindings/lib/array_serialization.h
+++ b/mojo/public/cpp/bindings/lib/array_serialization.h
@@ -14,12 +14,11 @@
 #include <vector>
 
 #include "base/logging.h"
-#include "mojo/public/cpp/bindings/array.h"
+#include "mojo/public/cpp/bindings/array_data_view.h"
 #include "mojo/public/cpp/bindings/lib/array_internal.h"
 #include "mojo/public/cpp/bindings/lib/serialization_forward.h"
 #include "mojo/public/cpp/bindings/lib/template_util.h"
 #include "mojo/public/cpp/bindings/lib/validation_errors.h"
-#include "mojo/public/cpp/bindings/map.h"
 
 namespace mojo {
 namespace internal {
@@ -46,7 +45,7 @@ class ArrayIterator<Traits, MaybeConstUserType, true> {
   using GetNextResult =
       decltype(Traits::GetValue(std::declval<IteratorType&>()));
   GetNextResult GetNext() {
-    auto& value = Traits::GetValue(iter_);
+    GetNextResult value = Traits::GetValue(iter_);
     Traits::AdvanceIterator(iter_);
     return value;
   }
@@ -287,13 +286,19 @@ struct ArraySerializer<
   using Element = typename MojomType::Element;
   using Traits = ArrayTraits<UserType>;
 
-  static_assert(std::is_same<Element, typename Traits::Element>::value,
-                "Incorrect array serializer");
-
   static size_t GetSerializedSize(UserTypeIterator* input,
                                   SerializationContext* context) {
-    return sizeof(Data) +
-           Align(input->GetSize() * sizeof(typename Data::Element));
+    size_t element_count = input->GetSize();
+    if (BelongsTo<Element,
+                  MojomTypeCategory::ASSOCIATED_INTERFACE |
+                      MojomTypeCategory::ASSOCIATED_INTERFACE_REQUEST>::value) {
+      for (size_t i = 0; i < element_count; ++i) {
+        typename UserTypeIterator::GetNextResult next = input->GetNext();
+        size_t size = PrepareToSerialize<Element>(next, context);
+        DCHECK_EQ(size, 0u);
+      }
+    }
+    return sizeof(Data) + Align(element_count * sizeof(typename Data::Element));
   }
 
   static void SerializeElements(UserTypeIterator* input,
@@ -306,7 +311,8 @@ struct ArraySerializer<
 
     size_t size = input->GetSize();
     for (size_t i = 0; i < size; ++i) {
-      Serialize<Element>(input->GetNext(), &output->at(i), context);
+      typename UserTypeIterator::GetNextResult next = input->GetNext();
+      Serialize<Element>(next, &output->at(i), context);
 
       static const ValidationError kError =
           BelongsTo<Element,
@@ -361,8 +367,10 @@ struct ArraySerializer<MojomType,
                                   SerializationContext* context) {
     size_t element_count = input->GetSize();
     size_t size = sizeof(Data) + element_count * sizeof(typename Data::Element);
-    for (size_t i = 0; i < element_count; ++i)
-      size += PrepareToSerialize<Element>(input->GetNext(), context);
+    for (size_t i = 0; i < element_count; ++i) {
+      typename UserTypeIterator::GetNextResult next = input->GetNext();
+      size += PrepareToSerialize<Element>(next, context);
+    }
     return size;
   }
 
@@ -374,7 +382,8 @@ struct ArraySerializer<MojomType,
     size_t size = input->GetSize();
     for (size_t i = 0; i < size; ++i) {
       DataElementPtr data_ptr;
-      SerializeCaller<Element>::Run(input->GetNext(), buf, &data_ptr,
+      typename UserTypeIterator::GetNextResult next = input->GetNext();
+      SerializeCaller<Element>::Run(next, buf, &data_ptr,
                                     validate_params->element_validate_params,
                                     context);
       output->at(i).Set(data_ptr);
@@ -444,10 +453,6 @@ struct ArraySerializer<
   using Element = typename MojomType::Element;
   using Traits = ArrayTraits<UserType>;
 
-  static_assert(std::is_same<typename MojomType::Element,
-                             typename Traits::Element>::value,
-                "Incorrect array serializer");
-
   static size_t GetSerializedSize(UserTypeIterator* input,
                                   SerializationContext* context) {
     size_t element_count = input->GetSize();
@@ -455,7 +460,8 @@ struct ArraySerializer<
     for (size_t i = 0; i < element_count; ++i) {
       // Call with |inlined| set to false, so that it will account for both the
       // data in the union and the space in the array used to hold the union.
-      size += PrepareToSerialize<Element>(input->GetNext(), false, context);
+      typename UserTypeIterator::GetNextResult next = input->GetNext();
+      size += PrepareToSerialize<Element>(next, false, context);
     }
     return size;
   }
@@ -468,7 +474,8 @@ struct ArraySerializer<
     size_t size = input->GetSize();
     for (size_t i = 0; i < size; ++i) {
       typename Data::Element* result = output->storage() + i;
-      Serialize<Element>(input->GetNext(), buf, &result, true, context);
+      typename UserTypeIterator::GetNextResult next = input->GetNext();
+      Serialize<Element>(next, buf, &result, true, context);
       MOJO_INTERNAL_DLOG_SERIALIZATION_WARNING(
           !validate_params->element_is_nullable && output->at(i).is_null(),
           VALIDATION_ERROR_UNEXPECTED_NULL_POINTER,
@@ -492,13 +499,13 @@ struct ArraySerializer<
 };
 
 template <typename Element, typename MaybeConstUserType>
-struct Serializer<Array<Element>, MaybeConstUserType> {
+struct Serializer<ArrayDataView<Element>, MaybeConstUserType> {
   using UserType = typename std::remove_const<MaybeConstUserType>::type;
   using Traits = ArrayTraits<UserType>;
-  using Impl = ArraySerializer<Array<Element>,
+  using Impl = ArraySerializer<ArrayDataView<Element>,
                                MaybeConstUserType,
                                ArrayIterator<Traits, MaybeConstUserType>>;
-  using Data = typename MojomTypeTraits<Array<Element>>::Data;
+  using Data = typename MojomTypeTraits<ArrayDataView<Element>>::Data;
 
   static size_t PrepareToSerialize(MaybeConstUserType& input,
                                    SerializationContext* context) {
diff --git a/mojo/public/cpp/bindings/lib/associated_group.cc b/mojo/public/cpp/bindings/lib/associated_group.cc
index a9c53b5..3e95eeb 100644
--- a/mojo/public/cpp/bindings/lib/associated_group.cc
+++ b/mojo/public/cpp/bindings/lib/associated_group.cc
@@ -8,28 +8,27 @@
 
 namespace mojo {
 
-AssociatedGroup::AssociatedGroup() {}
+AssociatedGroup::AssociatedGroup() = default;
 
-AssociatedGroup::AssociatedGroup(const AssociatedGroup& other)
-    : controller_(other.controller_) {}
+AssociatedGroup::AssociatedGroup(
+    scoped_refptr<AssociatedGroupController> controller)
+    : controller_(std::move(controller)) {}
 
-AssociatedGroup::~AssociatedGroup() {}
+AssociatedGroup::AssociatedGroup(const ScopedInterfaceEndpointHandle& handle)
+    : controller_getter_(handle.CreateGroupControllerGetter()) {}
 
-AssociatedGroup& AssociatedGroup::operator=(const AssociatedGroup& other) {
-  if (this == &other)
-    return *this;
+AssociatedGroup::AssociatedGroup(const AssociatedGroup& other) = default;
 
-  controller_ = other.controller_;
-  return *this;
-}
+AssociatedGroup::~AssociatedGroup() = default;
+
+AssociatedGroup& AssociatedGroup::operator=(const AssociatedGroup& other) =
+    default;
 
-void AssociatedGroup::CreateEndpointHandlePair(
-    ScopedInterfaceEndpointHandle* local_endpoint,
-    ScopedInterfaceEndpointHandle* remote_endpoint) {
-  if (!controller_)
-    return;
+AssociatedGroupController* AssociatedGroup::GetController() {
+  if (controller_)
+    return controller_.get();
 
-  controller_->CreateEndpointHandlePair(local_endpoint, remote_endpoint);
+  return controller_getter_.Run();
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/associated_group_controller.cc b/mojo/public/cpp/bindings/lib/associated_group_controller.cc
index 42db9b3..f4a9aa2 100644
--- a/mojo/public/cpp/bindings/lib/associated_group_controller.cc
+++ b/mojo/public/cpp/bindings/lib/associated_group_controller.cc
@@ -8,25 +8,17 @@
 
 namespace mojo {
 
-AssociatedGroupController::AssociatedGroupController(
-    scoped_refptr<base::SingleThreadTaskRunner> task_runner)
-    : base::RefCountedDeleteOnMessageLoop<AssociatedGroupController>(
-          task_runner) {}
-
 AssociatedGroupController::~AssociatedGroupController() {}
 
-std::unique_ptr<AssociatedGroup>
-AssociatedGroupController::CreateAssociatedGroup() {
-  std::unique_ptr<AssociatedGroup> group(new AssociatedGroup);
-  group->controller_ = this;
-  return group;
+ScopedInterfaceEndpointHandle
+AssociatedGroupController::CreateScopedInterfaceEndpointHandle(InterfaceId id) {
+  return ScopedInterfaceEndpointHandle(id, this);
 }
 
-ScopedInterfaceEndpointHandle
-AssociatedGroupController::CreateScopedInterfaceEndpointHandle(
-    InterfaceId id,
-    bool is_local) {
-  return ScopedInterfaceEndpointHandle(id, is_local, this);
+bool AssociatedGroupController::NotifyAssociation(
+    ScopedInterfaceEndpointHandle* handle_to_send,
+    InterfaceId id) {
+  return handle_to_send->NotifyAssociation(id, this);
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/associated_interface_ptr_state.h b/mojo/public/cpp/bindings/lib/associated_interface_ptr_state.h
index c7f74fb..72f7960 100644
--- a/mojo/public/cpp/bindings/lib/associated_interface_ptr_state.h
+++ b/mojo/public/cpp/bindings/lib/associated_interface_ptr_state.h
@@ -9,6 +9,7 @@
 
 #include <algorithm>  // For |std::swap()|.
 #include <memory>
+#include <string>
 #include <utility>
 
 #include "base/bind.h"
@@ -18,11 +19,10 @@
 #include "base/memory/ref_counted.h"
 #include "base/single_thread_task_runner.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
-#include "mojo/public/cpp/bindings/associated_group_controller.h"
 #include "mojo/public/cpp/bindings/associated_interface_ptr_info.h"
+#include "mojo/public/cpp/bindings/connection_error_callback.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_id.h"
-#include "mojo/public/cpp/bindings/lib/control_message_proxy.h"
 #include "mojo/public/cpp/bindings/scoped_interface_endpoint_handle.h"
 #include "mojo/public/cpp/system/message_pipe.h"
 
@@ -46,18 +46,12 @@ class AssociatedInterfacePtrState {
 
   uint32_t version() const { return version_; }
 
-  uint32_t interface_id() const {
-    DCHECK(is_bound());
-    return endpoint_client_->interface_id();
-  }
-
   void QueryVersion(const base::Callback<void(uint32_t)>& callback) {
-    // Do a static cast in case the interface contains methods with the same
-    // name. It is safe to capture |this| because the callback won't be run
-    // after this object goes away.
-    static_cast<ControlMessageProxy*>(proxy_.get())
-        ->QueryVersion(base::Bind(&AssociatedInterfacePtrState::OnQueryVersion,
-                                  base::Unretained(this), callback));
+    // It is safe to capture |this| because the callback won't be run after this
+    // object goes away.
+    endpoint_client_->QueryVersion(
+        base::Bind(&AssociatedInterfacePtrState::OnQueryVersion,
+                   base::Unretained(this), callback));
   }
 
   void RequireVersion(uint32_t version) {
@@ -65,9 +59,13 @@ class AssociatedInterfacePtrState {
       return;
 
     version_ = version;
-    // Do a static cast in case the interface contains methods with the same
-    // name.
-    static_cast<ControlMessageProxy*>(proxy_.get())->RequireVersion(version);
+    endpoint_client_->RequireVersion(version);
+  }
+
+  void FlushForTesting() { endpoint_client_->FlushForTesting(); }
+
+  void CloseWithReason(uint32_t custom_reason, const std::string& description) {
+    endpoint_client_->CloseWithReason(custom_reason, description);
   }
 
   void Swap(AssociatedInterfacePtrState* other) {
@@ -85,13 +83,13 @@ class AssociatedInterfacePtrState {
     DCHECK(info.is_valid());
 
     version_ = info.version();
+    // The version is only queried from the client so the value passed here
+    // will not be used.
     endpoint_client_.reset(new InterfaceEndpointClient(
         info.PassHandle(), nullptr,
         base::WrapUnique(new typename Interface::ResponseValidator_()), false,
-        std::move(runner)));
+        std::move(runner), 0u));
     proxy_.reset(new Proxy(endpoint_client_.get()));
-    proxy_->serialization_context()->group_controller =
-        endpoint_client_->group_controller();
   }
 
   // After this method is called, the object is in an invalid state and
@@ -114,6 +112,12 @@ class AssociatedInterfacePtrState {
     endpoint_client_->set_connection_error_handler(error_handler);
   }
 
+  void set_connection_error_with_reason_handler(
+      const ConnectionErrorWithReasonCallback& error_handler) {
+    DCHECK(endpoint_client_);
+    endpoint_client_->set_connection_error_with_reason_handler(error_handler);
+  }
+
   // Returns true if bound and awaiting a response to a message.
   bool has_pending_callbacks() const {
     return endpoint_client_ && endpoint_client_->has_pending_responders();
@@ -123,6 +127,13 @@ class AssociatedInterfacePtrState {
     return endpoint_client_ ? endpoint_client_->associated_group() : nullptr;
   }
 
+  void ForwardMessage(Message message) { endpoint_client_->Accept(&message); }
+
+  void ForwardMessageWithResponder(Message message,
+                                   std::unique_ptr<MessageReceiver> responder) {
+    endpoint_client_->AcceptWithResponder(&message, responder.release());
+  }
+
  private:
   using Proxy = typename Interface::Proxy_;
 
diff --git a/mojo/public/cpp/bindings/lib/binding_state.h b/mojo/public/cpp/bindings/lib/binding_state.h
index c8d3e83..0b0dbee 100644
--- a/mojo/public/cpp/bindings/lib/binding_state.h
+++ b/mojo/public/cpp/bindings/lib/binding_state.h
@@ -6,6 +6,7 @@
 #define MOJO_PUBLIC_CPP_BINDINGS_LIB_BINDING_STATE_H_
 
 #include <memory>
+#include <string>
 #include <utility>
 
 #include "base/bind.h"
@@ -15,15 +16,15 @@
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.h"
 #include "base/single_thread_task_runner.h"
-#include "mojo/public/cpp/bindings/associated_group.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
+#include "mojo/public/cpp/bindings/connection_error_callback.h"
+#include "mojo/public/cpp/bindings/filter_chain.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_id.h"
 #include "mojo/public/cpp/bindings/interface_ptr.h"
 #include "mojo/public/cpp/bindings/interface_ptr_info.h"
 #include "mojo/public/cpp/bindings/interface_request.h"
-#include "mojo/public/cpp/bindings/lib/filter_chain.h"
 #include "mojo/public/cpp/bindings/lib/multiplex_router.h"
-#include "mojo/public/cpp/bindings/lib/router.h"
 #include "mojo/public/cpp/bindings/message_header_validator.h"
 #include "mojo/public/cpp/bindings/scoped_interface_endpoint_handle.h"
 #include "mojo/public/cpp/system/core.h"
@@ -31,76 +32,34 @@
 namespace mojo {
 namespace internal {
 
-template <typename Interface, bool use_multiplex_router>
-class BindingState;
-
-// Uses a single-threaded, dedicated router. If |Interface| doesn't have any
-// methods to pass associated interface pointers or requests, there won't be
-// multiple interfaces running on the underlying message pipe. In that case, we
-// can use this specialization to reduce cost.
-template <typename Interface>
-class BindingState<Interface, false> {
+class MOJO_CPP_BINDINGS_EXPORT BindingStateBase {
  public:
-  explicit BindingState(Interface* impl) : impl_(impl) {
-    stub_.set_sink(impl_);
-  }
+  BindingStateBase();
+  ~BindingStateBase();
 
-  ~BindingState() { Close(); }
+  void AddFilter(std::unique_ptr<MessageReceiver> filter);
 
-  void Bind(ScopedMessagePipeHandle handle,
-            scoped_refptr<base::SingleThreadTaskRunner> runner) {
-    DCHECK(!router_);
-    internal::FilterChain filters;
-    filters.Append<MessageHeaderValidator>(Interface::Name_);
-    filters.Append<typename Interface::RequestValidator_>();
-
-    router_ =
-        new internal::Router(std::move(handle), std::move(filters),
-                             Interface::HasSyncMethods_, std::move(runner));
-    router_->set_incoming_receiver(&stub_);
-    router_->set_connection_error_handler(
-        base::Bind(&BindingState::RunConnectionErrorHandler,
-                   base::Unretained(this)));
-  }
-
-  bool HasAssociatedInterfaces() const { return false; }
+  bool HasAssociatedInterfaces() const;
 
-  void PauseIncomingMethodCallProcessing() {
-    DCHECK(router_);
-    router_->PauseIncomingMethodCallProcessing();
-  }
-  void ResumeIncomingMethodCallProcessing() {
-    DCHECK(router_);
-    router_->ResumeIncomingMethodCallProcessing();
-  }
+  void PauseIncomingMethodCallProcessing();
+  void ResumeIncomingMethodCallProcessing();
 
   bool WaitForIncomingMethodCall(
-      MojoDeadline deadline = MOJO_DEADLINE_INDEFINITE) {
-    DCHECK(router_);
-    return router_->WaitForIncomingMessage(deadline);
-  }
-
-  void Close() {
-    if (!router_)
-      return;
-
-    router_->CloseMessagePipe();
-    DestroyRouter();
-  }
+      MojoDeadline deadline = MOJO_DEADLINE_INDEFINITE);
 
-  InterfaceRequest<Interface> Unbind() {
-    InterfaceRequest<Interface> request =
-        MakeRequest<Interface>(router_->PassMessagePipe());
-    DestroyRouter();
-    return std::move(request);
-  }
+  void Close();
+  void CloseWithReason(uint32_t custom_reason, const std::string& description);
 
   void set_connection_error_handler(const base::Closure& error_handler) {
     DCHECK(is_bound());
-    connection_error_handler_ = error_handler;
+    endpoint_client_->set_connection_error_handler(error_handler);
   }
 
-  Interface* impl() { return impl_; }
+  void set_connection_error_with_reason_handler(
+      const ConnectionErrorWithReasonCallback& error_handler) {
+    DCHECK(is_bound());
+    endpoint_client_->set_connection_error_with_reason_handler(error_handler);
+  }
 
   bool is_bound() const { return !!router_; }
 
@@ -109,90 +68,42 @@ class BindingState<Interface, false> {
     return router_->handle();
   }
 
-  AssociatedGroup* associated_group() { return nullptr; }
+  void FlushForTesting();
 
-  void EnableTestingMode() {
-    DCHECK(is_bound());
-    router_->EnableTestingMode();
-  }
+  void EnableTestingMode();
 
- private:
-  void DestroyRouter() {
-    router_->set_connection_error_handler(base::Closure());
-    delete router_;
-    router_ = nullptr;
-    connection_error_handler_.Reset();
-  }
-
-  void RunConnectionErrorHandler() {
-    if (!connection_error_handler_.is_null())
-      connection_error_handler_.Run();
-  }
+ protected:
+  void BindInternal(ScopedMessagePipeHandle handle,
+                    scoped_refptr<base::SingleThreadTaskRunner> runner,
+                    const char* interface_name,
+                    std::unique_ptr<MessageReceiver> request_validator,
+                    bool passes_associated_kinds,
+                    bool has_sync_methods,
+                    MessageReceiverWithResponderStatus* stub,
+                    uint32_t interface_version);
 
-  internal::Router* router_ = nullptr;
-  typename Interface::Stub_ stub_;
-  Interface* impl_;
-  base::Closure connection_error_handler_;
-
-  DISALLOW_COPY_AND_ASSIGN(BindingState);
+  scoped_refptr<internal::MultiplexRouter> router_;
+  std::unique_ptr<InterfaceEndpointClient> endpoint_client_;
 };
 
-// Uses a multiplexing router. If |Interface| has methods to pass associated
-// interface pointers or requests, this specialization should be used.
-template <typename Interface>
-class BindingState<Interface, true> {
+template <typename Interface, typename ImplRefTraits>
+class BindingState : public BindingStateBase {
  public:
-  explicit BindingState(Interface* impl) : impl_(impl) {
-    stub_.set_sink(impl_);
+  using ImplPointerType = typename ImplRefTraits::PointerType;
+
+  explicit BindingState(ImplPointerType impl) {
+    stub_.set_sink(std::move(impl));
   }
 
   ~BindingState() { Close(); }
 
   void Bind(ScopedMessagePipeHandle handle,
             scoped_refptr<base::SingleThreadTaskRunner> runner) {
-    DCHECK(!router_);
-
-    router_ = new internal::MultiplexRouter(false, std::move(handle), runner);
-    router_->SetMasterInterfaceName(Interface::Name_);
-    stub_.serialization_context()->group_controller = router_;
-
-    endpoint_client_.reset(new InterfaceEndpointClient(
-        router_->CreateLocalEndpointHandle(kMasterInterfaceId),
-        &stub_, base::WrapUnique(new typename Interface::RequestValidator_()),
-        Interface::HasSyncMethods_, std::move(runner)));
-
-    endpoint_client_->set_connection_error_handler(
-        base::Bind(&BindingState::RunConnectionErrorHandler,
-                   base::Unretained(this)));
-  }
-
-  bool HasAssociatedInterfaces() const {
-    return router_ ? router_->HasAssociatedEndpoints() : false;
-  }
-
-  void PauseIncomingMethodCallProcessing() {
-    DCHECK(router_);
-    router_->PauseIncomingMethodCallProcessing();
-  }
-  void ResumeIncomingMethodCallProcessing() {
-    DCHECK(router_);
-    router_->ResumeIncomingMethodCallProcessing();
-  }
-
-  bool WaitForIncomingMethodCall(
-      MojoDeadline deadline = MOJO_DEADLINE_INDEFINITE) {
-    DCHECK(router_);
-    return router_->WaitForIncomingMessage(deadline);
-  }
-
-  void Close() {
-    if (!router_)
-      return;
-
-    endpoint_client_.reset();
-    router_->CloseMessagePipe();
-    router_ = nullptr;
-    connection_error_handler_.Reset();
+    BindingStateBase::BindInternal(
+        std::move(handle), runner, Interface::Name_,
+        base::MakeUnique<typename Interface::RequestValidator_>(),
+        Interface::PassesAssociatedKinds_, Interface::HasSyncMethods_, &stub_,
+        Interface::Version_);
   }
 
   InterfaceRequest<Interface> Unbind() {
@@ -200,45 +111,13 @@ class BindingState<Interface, true> {
     InterfaceRequest<Interface> request =
         MakeRequest<Interface>(router_->PassMessagePipe());
     router_ = nullptr;
-    connection_error_handler_.Reset();
     return request;
   }
 
-  void set_connection_error_handler(const base::Closure& error_handler) {
-    DCHECK(is_bound());
-    connection_error_handler_ = error_handler;
-  }
-
-  Interface* impl() { return impl_; }
-
-  bool is_bound() const { return !!router_; }
-
-  MessagePipeHandle handle() const {
-    DCHECK(is_bound());
-    return router_->handle();
-  }
-
-  AssociatedGroup* associated_group() {
-    return endpoint_client_ ? endpoint_client_->associated_group() : nullptr;
-  }
-
-  void EnableTestingMode() {
-    DCHECK(is_bound());
-    router_->EnableTestingMode();
-  }
+  Interface* impl() { return ImplRefTraits::GetRawPointer(&stub_.sink()); }
 
  private:
-  void RunConnectionErrorHandler() {
-    if (!connection_error_handler_.is_null())
-      connection_error_handler_.Run();
-  }
-
-  scoped_refptr<internal::MultiplexRouter> router_;
-  std::unique_ptr<InterfaceEndpointClient> endpoint_client_;
-
-  typename Interface::Stub_ stub_;
-  Interface* impl_;
-  base::Closure connection_error_handler_;
+  typename Interface::template Stub_<ImplRefTraits> stub_;
 
   DISALLOW_COPY_AND_ASSIGN(BindingState);
 };
diff --git a/mojo/public/cpp/bindings/lib/bindings_internal.cc b/mojo/public/cpp/bindings/lib/bindings_internal.cc
deleted file mode 100644
index a3bdb1f..0000000
--- a/mojo/public/cpp/bindings/lib/bindings_internal.cc
+++ /dev/null
@@ -1,47 +0,0 @@
-// Copyright 2016 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "mojo/public/cpp/bindings/lib/bindings_internal.h"
-
-namespace mojo {
-namespace internal {
-
-namespace {
-
-const size_t kAlignment = 8;
-
-template <typename T>
-T AlignImpl(T t) {
-  return t + (kAlignment - (t % kAlignment)) % kAlignment;
-}
-
-}  // namespace
-
-size_t Align(size_t size) {
-  return AlignImpl(size);
-}
-
-char* AlignPointer(char* ptr) {
-  return reinterpret_cast<char*>(AlignImpl(reinterpret_cast<uintptr_t>(ptr)));
-}
-
-bool IsAligned(const void* ptr) {
-  return !(reinterpret_cast<uintptr_t>(ptr) % kAlignment);
-}
-
-void EncodePointer(const void* ptr, uint64_t* offset) {
-  if (!ptr) {
-    *offset = 0;
-    return;
-  }
-
-  const char* p_obj = reinterpret_cast<const char*>(ptr);
-  const char* p_slot = reinterpret_cast<const char*>(offset);
-  DCHECK(p_obj > p_slot);
-
-  *offset = static_cast<uint64_t>(p_obj - p_slot);
-}
-
-}  // namespace internal
-}  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/bindings_internal.h b/mojo/public/cpp/bindings/lib/bindings_internal.h
index b37d872..631daec 100644
--- a/mojo/public/cpp/bindings/lib/bindings_internal.h
+++ b/mojo/public/cpp/bindings/lib/bindings_internal.h
@@ -17,30 +17,26 @@
 namespace mojo {
 
 template <typename T>
-class Array;
+class ArrayDataView;
 
 template <typename T>
-class AssociatedInterfacePtrInfo;
+class AssociatedInterfacePtrInfoDataView;
 
 template <typename T>
-class AssociatedInterfaceRequest;
+class AssociatedInterfaceRequestDataView;
 
 template <typename T>
-class InterfacePtr;
+class InterfacePtrDataView;
 
 template <typename T>
-class InterfaceRequest;
+class InterfaceRequestDataView;
 
 template <typename K, typename V>
-class Map;
+class MapDataView;
 
-class String;
+class NativeStructDataView;
 
-template <typename T>
-class StructPtr;
-
-template <typename T>
-class InlinedStructPtr;
+class StringDataView;
 
 namespace internal {
 
@@ -58,12 +54,17 @@ class Array_Data;
 template <typename K, typename V>
 class Map_Data;
 
+class NativeStruct_Data;
+
 using String_Data = Array_Data<char>;
 
-size_t Align(size_t size);
-char* AlignPointer(char* ptr);
+inline size_t Align(size_t size) {
+  return (size + 7) & ~0x7;
+}
 
-bool IsAligned(const void* ptr);
+inline bool IsAligned(const void* ptr) {
+  return !(reinterpret_cast<uintptr_t>(ptr) & 0x7);
+}
 
 // Pointers are encoded as relative offsets. The offsets are relative to the
 // address of where the offset value is stored, such that the pointer may be
@@ -73,7 +74,19 @@ bool IsAligned(const void* ptr);
 //
 // A null pointer is encoded as an offset value of 0.
 //
-void EncodePointer(const void* ptr, uint64_t* offset);
+inline void EncodePointer(const void* ptr, uint64_t* offset) {
+  if (!ptr) {
+    *offset = 0;
+    return;
+  }
+
+  const char* p_obj = reinterpret_cast<const char*>(ptr);
+  const char* p_slot = reinterpret_cast<const char*>(offset);
+  DCHECK(p_obj > p_slot);
+
+  *offset = static_cast<uint64_t>(p_obj - p_slot);
+}
+
 // Note: This function doesn't validate the encoded pointer value.
 inline const void* DecodePointer(const uint64_t* offset) {
   if (!*offset)
@@ -111,6 +124,8 @@ struct Pointer {
 };
 static_assert(sizeof(Pointer<char>) == 8, "Bad_sizeof(Pointer)");
 
+using GenericPointer = Pointer<void>;
+
 struct Handle_Data {
   Handle_Data() = default;
   explicit Handle_Data(uint32_t value) : value(value) {}
@@ -127,19 +142,24 @@ struct Interface_Data {
 };
 static_assert(sizeof(Interface_Data) == 8, "Bad_sizeof(Interface_Data)");
 
+struct AssociatedEndpointHandle_Data {
+  AssociatedEndpointHandle_Data() = default;
+  explicit AssociatedEndpointHandle_Data(uint32_t value) : value(value) {}
+
+  bool is_valid() const { return value != kEncodedInvalidHandleValue; }
+
+  uint32_t value;
+};
+static_assert(sizeof(AssociatedEndpointHandle_Data) == 4,
+              "Bad_sizeof(AssociatedEndpointHandle_Data)");
+
 struct AssociatedInterface_Data {
-  InterfaceId interface_id;
+  AssociatedEndpointHandle_Data handle;
   uint32_t version;
 };
 static_assert(sizeof(AssociatedInterface_Data) == 8,
               "Bad_sizeof(AssociatedInterface_Data)");
 
-struct AssociatedInterfaceRequest_Data {
-  InterfaceId interface_id;
-};
-static_assert(sizeof(AssociatedInterfaceRequest_Data) == 4,
-              "Bad_sizeof(AssociatedInterfaceRequest_Data)");
-
 #pragma pack(pop)
 
 template <typename T>
@@ -203,7 +223,7 @@ struct MojomTypeTraits {
 };
 
 template <typename T>
-struct MojomTypeTraits<Array<T>, false> {
+struct MojomTypeTraits<ArrayDataView<T>, false> {
   using Data = Array_Data<typename MojomTypeTraits<T>::DataAsArrayElement>;
   using DataAsArrayElement = Pointer<Data>;
 
@@ -211,7 +231,7 @@ struct MojomTypeTraits<Array<T>, false> {
 };
 
 template <typename T>
-struct MojomTypeTraits<AssociatedInterfacePtrInfo<T>, false> {
+struct MojomTypeTraits<AssociatedInterfacePtrInfoDataView<T>, false> {
   using Data = AssociatedInterface_Data;
   using DataAsArrayElement = Data;
 
@@ -220,8 +240,8 @@ struct MojomTypeTraits<AssociatedInterfacePtrInfo<T>, false> {
 };
 
 template <typename T>
-struct MojomTypeTraits<AssociatedInterfaceRequest<T>, false> {
-  using Data = AssociatedInterfaceRequest_Data;
+struct MojomTypeTraits<AssociatedInterfaceRequestDataView<T>, false> {
+  using Data = AssociatedEndpointHandle_Data;
   using DataAsArrayElement = Data;
 
   static const MojomTypeCategory category =
@@ -253,7 +273,7 @@ struct MojomTypeTraits<ScopedHandleBase<T>, false> {
 };
 
 template <typename T>
-struct MojomTypeTraits<InterfacePtr<T>, false> {
+struct MojomTypeTraits<InterfacePtrDataView<T>, false> {
   using Data = Interface_Data;
   using DataAsArrayElement = Data;
 
@@ -261,7 +281,7 @@ struct MojomTypeTraits<InterfacePtr<T>, false> {
 };
 
 template <typename T>
-struct MojomTypeTraits<InterfaceRequest<T>, false> {
+struct MojomTypeTraits<InterfaceRequestDataView<T>, false> {
   using Data = Handle_Data;
   using DataAsArrayElement = Data;
 
@@ -270,7 +290,7 @@ struct MojomTypeTraits<InterfaceRequest<T>, false> {
 };
 
 template <typename K, typename V>
-struct MojomTypeTraits<Map<K, V>, false> {
+struct MojomTypeTraits<MapDataView<K, V>, false> {
   using Data = Map_Data<typename MojomTypeTraits<K>::DataAsArrayElement,
                         typename MojomTypeTraits<V>::DataAsArrayElement>;
   using DataAsArrayElement = Pointer<Data>;
@@ -279,37 +299,19 @@ struct MojomTypeTraits<Map<K, V>, false> {
 };
 
 template <>
-struct MojomTypeTraits<String, false> {
-  using Data = String_Data;
+struct MojomTypeTraits<NativeStructDataView, false> {
+  using Data = internal::NativeStruct_Data;
   using DataAsArrayElement = Pointer<Data>;
 
-  static const MojomTypeCategory category = MojomTypeCategory::STRING;
+  static const MojomTypeCategory category = MojomTypeCategory::STRUCT;
 };
 
-template <typename T>
-struct MojomTypeTraits<StructPtr<T>, false> {
-  using Data = typename T::Data_;
-  using DataAsArrayElement =
-      typename std::conditional<IsUnionDataType<Data>::value,
-                                Data,
-                                Pointer<Data>>::type;
-
-  static const MojomTypeCategory category = IsUnionDataType<Data>::value
-                                                ? MojomTypeCategory::UNION
-                                                : MojomTypeCategory::STRUCT;
-};
+template <>
+struct MojomTypeTraits<StringDataView, false> {
+  using Data = String_Data;
+  using DataAsArrayElement = Pointer<Data>;
 
-template <typename T>
-struct MojomTypeTraits<InlinedStructPtr<T>, false> {
-  using Data = typename T::Data_;
-  using DataAsArrayElement =
-      typename std::conditional<IsUnionDataType<Data>::value,
-                                Data,
-                                Pointer<Data>>::type;
-
-  static const MojomTypeCategory category = IsUnionDataType<Data>::value
-                                                ? MojomTypeCategory::UNION
-                                                : MojomTypeCategory::STRUCT;
+  static const MojomTypeCategory category = MojomTypeCategory::STRING;
 };
 
 template <typename T, MojomTypeCategory categories>
diff --git a/mojo/public/cpp/bindings/lib/buffer.h b/mojo/public/cpp/bindings/lib/buffer.h
index c3b570e..213a445 100644
--- a/mojo/public/cpp/bindings/lib/buffer.h
+++ b/mojo/public/cpp/bindings/lib/buffer.h
@@ -7,15 +7,61 @@
 
 #include <stddef.h>
 
+#include "base/logging.h"
+#include "base/macros.h"
+#include "mojo/public/cpp/bindings/lib/bindings_internal.h"
+
 namespace mojo {
 namespace internal {
 
-// Buffer provides a way to allocate memory. Allocations are 8-byte aligned and
-// zero-initialized. Allocations remain valid for the lifetime of the Buffer.
+// Buffer provides an interface to allocate memory blocks which are 8-byte
+// aligned and zero-initialized. It doesn't own the underlying memory. Users
+// must ensure that the memory stays valid while using the allocated blocks from
+// Buffer.
 class Buffer {
  public:
-  virtual ~Buffer() {}
-  virtual void* Allocate(size_t num_bytes) = 0;
+  Buffer() {}
+
+  // The memory must have been zero-initialized. |data| must be 8-byte
+  // aligned.
+  void Initialize(void* data, size_t size) {
+    DCHECK(IsAligned(data));
+
+    data_ = data;
+    size_ = size;
+    cursor_ = reinterpret_cast<uintptr_t>(data);
+    data_end_ = cursor_ + size;
+  }
+
+  size_t size() const { return size_; }
+
+  void* data() const { return data_; }
+
+  // Allocates |num_bytes| from the buffer and returns a pointer to the start of
+  // the allocated block.
+  // The resulting address is 8-byte aligned, and the content of the memory is
+  // zero-filled.
+  void* Allocate(size_t num_bytes) {
+    num_bytes = Align(num_bytes);
+    uintptr_t result = cursor_;
+    cursor_ += num_bytes;
+    if (cursor_ > data_end_ || cursor_ < result) {
+      NOTREACHED();
+      cursor_ -= num_bytes;
+      return nullptr;
+    }
+
+    return reinterpret_cast<void*>(result);
+  }
+
+ private:
+  void* data_ = nullptr;
+  size_t size_ = 0;
+
+  uintptr_t cursor_ = 0;
+  uintptr_t data_end_ = 0;
+
+  DISALLOW_COPY_AND_ASSIGN(Buffer);
 };
 
 }  // namespace internal
diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index 1bb38f0..4426def 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -12,52 +12,20 @@
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/synchronization/lock.h"
+#include "mojo/public/cpp/bindings/lib/may_auto_lock.h"
 #include "mojo/public/cpp/bindings/sync_handle_watcher.h"
 
 namespace mojo {
 
-namespace {
-
-// Similar to base::AutoLock, except that it does nothing if |lock| passed into
-// the constructor is null.
-class MayAutoLock {
- public:
-  explicit MayAutoLock(base::Lock* lock) : lock_(lock) {
-    if (lock_)
-      lock_->Acquire();
-  }
-
-  ~MayAutoLock() {
-    if (lock_) {
-      lock_->AssertAcquired();
-      lock_->Release();
-    }
-  }
-
- private:
-  base::Lock* lock_;
-  DISALLOW_COPY_AND_ASSIGN(MayAutoLock);
-};
-
-}  // namespace
-
-// ----------------------------------------------------------------------------
-
 Connector::Connector(ScopedMessagePipeHandle message_pipe,
                      ConnectorConfig config,
                      scoped_refptr<base::SingleThreadTaskRunner> runner)
     : message_pipe_(std::move(message_pipe)),
-      incoming_receiver_(nullptr),
       task_runner_(std::move(runner)),
-      handle_watcher_(task_runner_),
-      error_(false),
-      drop_writes_(false),
-      enforce_errors_from_incoming_receiver_(true),
-      paused_(false),
-      lock_(config == MULTI_THREADED_SEND ? new base::Lock : nullptr),
-      allow_woken_up_by_others_(false),
-      sync_handle_watcher_callback_count_(0),
       weak_factory_(this) {
+  if (config == MULTI_THREADED_SEND)
+    lock_.emplace();
+
   weak_self_ = weak_factory_.GetWeakPtr();
   // Even though we don't have an incoming receiver, we still want to monitor
   // the message pipe to know if is closed or encounters an error.
@@ -65,25 +33,34 @@ Connector::Connector(ScopedMessagePipeHandle message_pipe,
 }
 
 Connector::~Connector() {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  {
+    // Allow for quick destruction on any thread if the pipe is already closed.
+    base::AutoLock lock(connected_lock_);
+    if (!connected_)
+      return;
+  }
 
+  DCHECK(thread_checker_.CalledOnValidThread());
   CancelWait();
 }
 
 void Connector::CloseMessagePipe() {
-  DCHECK(thread_checker_.CalledOnValidThread());
-
-  CancelWait();
-  MayAutoLock locker(lock_.get());
-  message_pipe_.reset();
+  // Throw away the returned message pipe.
+  PassMessagePipe();
 }
 
 ScopedMessagePipeHandle Connector::PassMessagePipe() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   CancelWait();
-  MayAutoLock locker(lock_.get());
-  return std::move(message_pipe_);
+  internal::MayAutoLock locker(&lock_);
+  ScopedMessagePipeHandle message_pipe = std::move(message_pipe_);
+  weak_factory_.InvalidateWeakPtrs();
+  sync_handle_watcher_callback_count_ = 0;
+
+  base::AutoLock lock(connected_lock_);
+  connected_ = false;
+  return message_pipe;
 }
 
 void Connector::RaiseError() {
@@ -143,7 +120,7 @@ bool Connector::Accept(Message* message) {
   if (error_)
     return false;
 
-  MayAutoLock locker(lock_.get());
+  internal::MayAutoLock locker(&lock_);
 
   if (!message_pipe_.is_valid() || drop_writes_)
     return true;
@@ -204,6 +181,13 @@ bool Connector::SyncWatch(const bool* should_stop) {
   return sync_watcher_->SyncWatch(should_stop);
 }
 
+void Connector::SetWatcherHeapProfilerTag(const char* tag) {
+  heap_profiler_tag_ = tag;
+  if (handle_watcher_) {
+    handle_watcher_->set_heap_profiler_tag(tag);
+  }
+}
+
 void Connector::OnWatcherHandleReady(MojoResult result) {
   OnHandleReadyInternal(result);
 }
@@ -214,8 +198,10 @@ void Connector::OnSyncHandleWatcherHandleReady(MojoResult result) {
   sync_handle_watcher_callback_count_++;
   OnHandleReadyInternal(result);
   // At this point, this object might have been deleted.
-  if (weak_self)
+  if (weak_self) {
+    DCHECK_LT(0u, sync_handle_watcher_callback_count_);
     sync_handle_watcher_callback_count_--;
+  }
 }
 
 void Connector::OnHandleReadyInternal(MojoResult result) {
@@ -231,12 +217,14 @@ void Connector::OnHandleReadyInternal(MojoResult result) {
 
 void Connector::WaitToReadMore() {
   CHECK(!paused_);
-  DCHECK(!handle_watcher_.IsWatching());
+  DCHECK(!handle_watcher_);
 
-  MojoResult rv = handle_watcher_.Start(
+  handle_watcher_.reset(new Watcher(FROM_HERE, task_runner_));
+  if (heap_profiler_tag_)
+    handle_watcher_->set_heap_profiler_tag(heap_profiler_tag_);
+  MojoResult rv = handle_watcher_->Start(
       message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-      base::Bind(&Connector::OnWatcherHandleReady,
-                 base::Unretained(this)));
+      base::Bind(&Connector::OnWatcherHandleReady, base::Unretained(this)));
 
   if (rv != MOJO_RESULT_OK) {
     // If the watch failed because the handle is invalid or its conditions can
@@ -257,8 +245,8 @@ bool Connector::ReadSingleMessage(MojoResult* read_result) {
 
   bool receiver_result = false;
 
-  // Detect if |this| was destroyed during message dispatch. Allow for the
-  // possibility of re-entering ReadMore() through message dispatch.
+  // Detect if |this| was destroyed or the message pipe was closed/transferred
+  // during message dispatch.
   base::WeakPtr<Connector> weak_self = weak_self_;
 
   Message message;
@@ -292,9 +280,11 @@ void Connector::ReadAllAvailableMessages() {
   while (!error_) {
     MojoResult rv;
 
-    // Return immediately if |this| was destroyed. Do not touch any members!
-    if (!ReadSingleMessage(&rv))
+    if (!ReadSingleMessage(&rv)) {
+      // Return immediately without touching any members. |this| may have been
+      // destroyed.
       return;
+    }
 
     if (paused_)
       return;
@@ -305,7 +295,7 @@ void Connector::ReadAllAvailableMessages() {
 }
 
 void Connector::CancelWait() {
-  handle_watcher_.Cancel();
+  handle_watcher_.reset();
   sync_watcher_.reset();
 }
 
@@ -325,7 +315,7 @@ void Connector::HandleError(bool force_pipe_reset, bool force_async_handler) {
 
   if (force_pipe_reset) {
     CancelWait();
-    MayAutoLock locker(lock_.get());
+    internal::MayAutoLock locker(&lock_);
     message_pipe_.reset();
     MessagePipe dummy_pipe;
     message_pipe_ = std::move(dummy_pipe.handle0);
diff --git a/mojo/public/cpp/bindings/lib/control_message_handler.cc b/mojo/public/cpp/bindings/lib/control_message_handler.cc
index 9f44e88..c90aada 100644
--- a/mojo/public/cpp/bindings/lib/control_message_handler.cc
+++ b/mojo/public/cpp/bindings/lib/control_message_handler.cc
@@ -11,15 +11,53 @@
 #include "base/logging.h"
 #include "mojo/public/cpp/bindings/lib/message_builder.h"
 #include "mojo/public/cpp/bindings/lib/serialization.h"
+#include "mojo/public/cpp/bindings/lib/validation_util.h"
 #include "mojo/public/interfaces/bindings/interface_control_messages.mojom.h"
 
 namespace mojo {
 namespace internal {
+namespace {
+
+bool ValidateControlRequestWithResponse(Message* message) {
+  ValidationContext validation_context(message->payload(),
+                                       message->payload_num_bytes(), 0, 0,
+                                       message, "ControlRequestValidator");
+  if (!ValidateMessageIsRequestExpectingResponse(message, &validation_context))
+    return false;
+
+  switch (message->header()->name) {
+    case interface_control::kRunMessageId:
+      return ValidateMessagePayload<
+          interface_control::internal::RunMessageParams_Data>(
+          message, &validation_context);
+  }
+  return false;
+}
+
+bool ValidateControlRequestWithoutResponse(Message* message) {
+  ValidationContext validation_context(message->payload(),
+                                       message->payload_num_bytes(), 0, 0,
+                                       message, "ControlRequestValidator");
+  if (!ValidateMessageIsRequestWithoutResponse(message, &validation_context))
+    return false;
+
+  switch (message->header()->name) {
+    case interface_control::kRunOrClosePipeMessageId:
+      return ValidateMessageIsRequestWithoutResponse(message,
+                                                     &validation_context) &&
+             ValidateMessagePayload<
+                 interface_control::internal::RunOrClosePipeMessageParams_Data>(
+                 message, &validation_context);
+  }
+  return false;
+}
+
+}  // namespace
 
 // static
 bool ControlMessageHandler::IsControlMessage(const Message* message) {
-  return message->header()->name == kRunMessageId ||
-         message->header()->name == kRunOrClosePipeMessageId;
+  return message->header()->name == interface_control::kRunMessageId ||
+         message->header()->name == interface_control::kRunOrClosePipeMessageId;
 }
 
 ControlMessageHandler::ControlMessageHandler(uint32_t interface_version)
@@ -30,7 +68,10 @@ ControlMessageHandler::~ControlMessageHandler() {
 }
 
 bool ControlMessageHandler::Accept(Message* message) {
-  if (message->header()->name == kRunOrClosePipeMessageId)
+  if (!ValidateControlRequestWithoutResponse(message))
+    return false;
+
+  if (message->header()->name == interface_control::kRunOrClosePipeMessageId)
     return RunOrClosePipe(message);
 
   NOTREACHED();
@@ -40,7 +81,10 @@ bool ControlMessageHandler::Accept(Message* message) {
 bool ControlMessageHandler::AcceptWithResponder(
     Message* message,
     MessageReceiverWithStatus* responder) {
-  if (message->header()->name == kRunMessageId)
+  if (!ValidateControlRequestWithResponse(message))
+    return false;
+
+  if (message->header()->name == interface_control::kRunMessageId)
     return Run(message, responder);
 
   NOTREACHED();
@@ -49,20 +93,37 @@ bool ControlMessageHandler::AcceptWithResponder(
 
 bool ControlMessageHandler::Run(Message* message,
                                 MessageReceiverWithStatus* responder) {
-  RunResponseMessageParamsPtr response_params_ptr(
-      RunResponseMessageParams::New());
-  response_params_ptr->reserved0 = 16u;
-  response_params_ptr->reserved1 = 0u;
-  response_params_ptr->query_version_result = QueryVersionResult::New();
-  response_params_ptr->query_version_result->version = interface_version_;
-
-  size_t size = PrepareToSerialize<RunResponseMessageParamsPtr>(
-      response_params_ptr, &context_);
-  ResponseMessageBuilder builder(kRunMessageId, size, message->request_id());
-
-  RunResponseMessageParams_Data* response_params = nullptr;
-  Serialize<RunResponseMessageParamsPtr>(response_params_ptr, builder.buffer(),
-                                         &response_params, &context_);
+  interface_control::internal::RunMessageParams_Data* params =
+      reinterpret_cast<interface_control::internal::RunMessageParams_Data*>(
+          message->mutable_payload());
+  interface_control::RunMessageParamsPtr params_ptr;
+  Deserialize<interface_control::RunMessageParamsDataView>(params, &params_ptr,
+                                                           &context_);
+  auto& input = *params_ptr->input;
+  interface_control::RunOutputPtr output = interface_control::RunOutput::New();
+  if (input.is_query_version()) {
+    output->set_query_version_result(
+        interface_control::QueryVersionResult::New());
+    output->get_query_version_result()->version = interface_version_;
+  } else if (input.is_flush_for_testing()) {
+    output.reset();
+  } else {
+    output.reset();
+  }
+
+  auto response_params_ptr = interface_control::RunResponseMessageParams::New();
+  response_params_ptr->output = std::move(output);
+  size_t size =
+      PrepareToSerialize<interface_control::RunResponseMessageParamsDataView>(
+          response_params_ptr, &context_);
+  MessageBuilder builder(interface_control::kRunMessageId,
+                         Message::kFlagIsResponse, size, 0);
+  builder.message()->set_request_id(message->request_id());
+
+  interface_control::internal::RunResponseMessageParams_Data* response_params =
+      nullptr;
+  Serialize<interface_control::RunResponseMessageParamsDataView>(
+      response_params_ptr, builder.buffer(), &response_params, &context_);
   bool ok = responder->Accept(builder.message());
   ALLOW_UNUSED_LOCAL(ok);
   delete responder;
@@ -71,13 +132,18 @@ bool ControlMessageHandler::Run(Message* message,
 }
 
 bool ControlMessageHandler::RunOrClosePipe(Message* message) {
-  RunOrClosePipeMessageParams_Data* params =
-      reinterpret_cast<RunOrClosePipeMessageParams_Data*>(
+  interface_control::internal::RunOrClosePipeMessageParams_Data* params =
+      reinterpret_cast<
+          interface_control::internal::RunOrClosePipeMessageParams_Data*>(
           message->mutable_payload());
-  RunOrClosePipeMessageParamsPtr params_ptr;
-  Deserialize<RunOrClosePipeMessageParamsPtr>(params, &params_ptr, &context_);
+  interface_control::RunOrClosePipeMessageParamsPtr params_ptr;
+  Deserialize<interface_control::RunOrClosePipeMessageParamsDataView>(
+      params, &params_ptr, &context_);
+  auto& input = *params_ptr->input;
+  if (input.is_require_version())
+    return interface_version_ >= input.get_require_version()->version;
 
-  return interface_version_ >= params_ptr->require_version->version;
+  return false;
 }
 
 }  // namespace internal
diff --git a/mojo/public/cpp/bindings/lib/control_message_handler.h b/mojo/public/cpp/bindings/lib/control_message_handler.h
index 13b5aa6..3c385e4 100644
--- a/mojo/public/cpp/bindings/lib/control_message_handler.h
+++ b/mojo/public/cpp/bindings/lib/control_message_handler.h
@@ -7,7 +7,9 @@
 
 #include <stdint.h>
 
+#include "base/compiler_specific.h"
 #include "base/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/serialization_context.h"
 #include "mojo/public/cpp/bindings/message.h"
 
@@ -15,7 +17,8 @@ namespace mojo {
 namespace internal {
 
 // Handlers for request messages defined in interface_control_messages.mojom.
-class ControlMessageHandler : public MessageReceiverWithResponderStatus {
+class MOJO_CPP_BINDINGS_EXPORT ControlMessageHandler
+    : NON_EXPORTED_BASE(public MessageReceiverWithResponderStatus) {
  public:
   static bool IsControlMessage(const Message* message);
 
diff --git a/mojo/public/cpp/bindings/lib/control_message_proxy.cc b/mojo/public/cpp/bindings/lib/control_message_proxy.cc
index 7af409d..23de991 100644
--- a/mojo/public/cpp/bindings/lib/control_message_proxy.cc
+++ b/mojo/public/cpp/bindings/lib/control_message_proxy.cc
@@ -9,9 +9,12 @@
 #include <utility>
 
 #include "base/bind.h"
+#include "base/callback_helpers.h"
 #include "base/macros.h"
+#include "base/run_loop.h"
 #include "mojo/public/cpp/bindings/lib/message_builder.h"
 #include "mojo/public/cpp/bindings/lib/serialization.h"
+#include "mojo/public/cpp/bindings/lib/validation_util.h"
 #include "mojo/public/cpp/bindings/message.h"
 #include "mojo/public/interfaces/bindings/interface_control_messages.mojom.h"
 
@@ -20,11 +23,28 @@ namespace internal {
 
 namespace {
 
-using RunCallback = base::Callback<void(QueryVersionResultPtr)>;
+bool ValidateControlResponse(Message* message) {
+  ValidationContext validation_context(message->payload(),
+                                       message->payload_num_bytes(), 0, 0,
+                                       message, "ControlResponseValidator");
+  if (!ValidateMessageIsResponse(message, &validation_context))
+    return false;
+
+  switch (message->header()->name) {
+    case interface_control::kRunMessageId:
+      return ValidateMessagePayload<
+          interface_control::internal::RunResponseMessageParams_Data>(
+          message, &validation_context);
+  }
+  return false;
+}
+
+using RunCallback =
+    base::Callback<void(interface_control::RunResponseMessageParamsPtr)>;
 
 class RunResponseForwardToCallback : public MessageReceiver {
  public:
-  RunResponseForwardToCallback(const RunCallback& callback)
+  explicit RunResponseForwardToCallback(const RunCallback& callback)
       : callback_(callback) {}
   bool Accept(Message* message) override;
 
@@ -34,59 +54,83 @@ class RunResponseForwardToCallback : public MessageReceiver {
 };
 
 bool RunResponseForwardToCallback::Accept(Message* message) {
-  RunResponseMessageParams_Data* params =
-      reinterpret_cast<RunResponseMessageParams_Data*>(
+  if (!ValidateControlResponse(message))
+    return false;
+
+  interface_control::internal::RunResponseMessageParams_Data* params =
+      reinterpret_cast<
+          interface_control::internal::RunResponseMessageParams_Data*>(
           message->mutable_payload());
-  RunResponseMessageParamsPtr params_ptr;
+  interface_control::RunResponseMessageParamsPtr params_ptr;
   SerializationContext context;
-  Deserialize<RunResponseMessageParamsPtr>(params, &params_ptr, &context);
+  Deserialize<interface_control::RunResponseMessageParamsDataView>(
+      params, &params_ptr, &context);
 
-  callback_.Run(std::move(params_ptr->query_version_result));
+  callback_.Run(std::move(params_ptr));
   return true;
 }
 
 void SendRunMessage(MessageReceiverWithResponder* receiver,
-                    QueryVersionPtr query_version,
-                    const RunCallback& callback,
-                    SerializationContext* context) {
-  RunMessageParamsPtr params_ptr(RunMessageParams::New());
-  params_ptr->reserved0 = 16u;
-  params_ptr->reserved1 = 0u;
-  params_ptr->query_version = std::move(query_version);
-
-  size_t size = PrepareToSerialize<RunMessageParamsPtr>(params_ptr, context);
-  RequestMessageBuilder builder(kRunMessageId, size);
-
-  RunMessageParams_Data* params = nullptr;
-  Serialize<RunMessageParamsPtr>(params_ptr, builder.buffer(), &params,
-                                 context);
+                    interface_control::RunInputPtr input_ptr,
+                    const RunCallback& callback) {
+  SerializationContext context;
+
+  auto params_ptr = interface_control::RunMessageParams::New();
+  params_ptr->input = std::move(input_ptr);
+  size_t size = PrepareToSerialize<interface_control::RunMessageParamsDataView>(
+      params_ptr, &context);
+  MessageBuilder builder(interface_control::kRunMessageId,
+                         Message::kFlagExpectsResponse, size, 0);
+
+  interface_control::internal::RunMessageParams_Data* params = nullptr;
+  Serialize<interface_control::RunMessageParamsDataView>(
+      params_ptr, builder.buffer(), &params, &context);
   MessageReceiver* responder = new RunResponseForwardToCallback(callback);
   if (!receiver->AcceptWithResponder(builder.message(), responder))
     delete responder;
 }
 
-void SendRunOrClosePipeMessage(MessageReceiverWithResponder* receiver,
-                               RequireVersionPtr require_version,
-                               SerializationContext* context) {
-  RunOrClosePipeMessageParamsPtr params_ptr(RunOrClosePipeMessageParams::New());
-  params_ptr->reserved0 = 16u;
-  params_ptr->reserved1 = 0u;
-  params_ptr->require_version = std::move(require_version);
-
-  size_t size =
-      PrepareToSerialize<RunOrClosePipeMessageParamsPtr>(params_ptr, context);
-  MessageBuilder builder(kRunOrClosePipeMessageId, size);
-
-  RunOrClosePipeMessageParams_Data* params = nullptr;
-  Serialize<RunOrClosePipeMessageParamsPtr>(params_ptr, builder.buffer(),
-                                            &params, context);
-  bool ok = receiver->Accept(builder.message());
+Message ConstructRunOrClosePipeMessage(
+    interface_control::RunOrClosePipeInputPtr input_ptr) {
+  SerializationContext context;
+
+  auto params_ptr = interface_control::RunOrClosePipeMessageParams::New();
+  params_ptr->input = std::move(input_ptr);
+
+  size_t size = PrepareToSerialize<
+      interface_control::RunOrClosePipeMessageParamsDataView>(params_ptr,
+                                                              &context);
+  MessageBuilder builder(interface_control::kRunOrClosePipeMessageId, 0, size,
+                         0);
+
+  interface_control::internal::RunOrClosePipeMessageParams_Data* params =
+      nullptr;
+  Serialize<interface_control::RunOrClosePipeMessageParamsDataView>(
+      params_ptr, builder.buffer(), &params, &context);
+  return std::move(*builder.message());
+}
+
+void SendRunOrClosePipeMessage(
+    MessageReceiverWithResponder* receiver,
+    interface_control::RunOrClosePipeInputPtr input_ptr) {
+  Message message(ConstructRunOrClosePipeMessage(std::move(input_ptr)));
+
+  bool ok = receiver->Accept(&message);
   ALLOW_UNUSED_LOCAL(ok);
 }
 
-void RunVersionCallback(const base::Callback<void(uint32_t)>& callback,
-                        QueryVersionResultPtr query_version_result) {
-  callback.Run(query_version_result->version);
+void RunVersionCallback(
+    const base::Callback<void(uint32_t)>& callback,
+    interface_control::RunResponseMessageParamsPtr run_response) {
+  uint32_t version = 0u;
+  if (run_response->output && run_response->output->is_query_version_result())
+    version = run_response->output->get_query_version_result()->version;
+  callback.Run(version);
+}
+
+void RunClosure(const base::Closure& callback,
+                interface_control::RunResponseMessageParamsPtr run_response) {
+  callback.Run();
 }
 
 }  // namespace
@@ -95,16 +139,49 @@ ControlMessageProxy::ControlMessageProxy(MessageReceiverWithResponder* receiver)
     : receiver_(receiver) {
 }
 
+ControlMessageProxy::~ControlMessageProxy() = default;
+
 void ControlMessageProxy::QueryVersion(
     const base::Callback<void(uint32_t)>& callback) {
-  SendRunMessage(receiver_, QueryVersion::New(),
-                 base::Bind(&RunVersionCallback, callback), &context_);
+  auto input_ptr = interface_control::RunInput::New();
+  input_ptr->set_query_version(interface_control::QueryVersion::New());
+  SendRunMessage(receiver_, std::move(input_ptr),
+                 base::Bind(&RunVersionCallback, callback));
 }
 
 void ControlMessageProxy::RequireVersion(uint32_t version) {
-  RequireVersionPtr require_version(RequireVersion::New());
+  auto require_version = interface_control::RequireVersion::New();
   require_version->version = version;
-  SendRunOrClosePipeMessage(receiver_, std::move(require_version), &context_);
+  auto input_ptr = interface_control::RunOrClosePipeInput::New();
+  input_ptr->set_require_version(std::move(require_version));
+  SendRunOrClosePipeMessage(receiver_, std::move(input_ptr));
+}
+
+void ControlMessageProxy::FlushForTesting() {
+  if (encountered_error_)
+    return;
+
+  auto input_ptr = interface_control::RunInput::New();
+  input_ptr->set_flush_for_testing(interface_control::FlushForTesting::New());
+  base::RunLoop run_loop;
+  run_loop_quit_closure_ = run_loop.QuitClosure();
+  SendRunMessage(
+      receiver_, std::move(input_ptr),
+      base::Bind(&RunClosure,
+                 base::Bind(&ControlMessageProxy::RunFlushForTestingClosure,
+                            base::Unretained(this))));
+  run_loop.Run();
+}
+
+void ControlMessageProxy::RunFlushForTestingClosure() {
+  DCHECK(!run_loop_quit_closure_.is_null());
+  base::ResetAndReturn(&run_loop_quit_closure_).Run();
+}
+
+void ControlMessageProxy::OnConnectionError() {
+  encountered_error_ = true;
+  if (!run_loop_quit_closure_.is_null())
+    RunFlushForTestingClosure();
 }
 
 }  // namespace internal
diff --git a/mojo/public/cpp/bindings/lib/control_message_proxy.h b/mojo/public/cpp/bindings/lib/control_message_proxy.h
index 5ec6ddc..2f9314e 100644
--- a/mojo/public/cpp/bindings/lib/control_message_proxy.h
+++ b/mojo/public/cpp/bindings/lib/control_message_proxy.h
@@ -7,8 +7,9 @@
 
 #include <stdint.h>
 
-#include "base/callback_forward.h"
+#include "base/callback.h"
 #include "base/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/serialization_context.h"
 
 namespace mojo {
@@ -18,18 +19,26 @@ class MessageReceiverWithResponder;
 namespace internal {
 
 // Proxy for request messages defined in interface_control_messages.mojom.
-class ControlMessageProxy {
+class MOJO_CPP_BINDINGS_EXPORT ControlMessageProxy {
  public:
   // Doesn't take ownership of |receiver|. It must outlive this object.
   explicit ControlMessageProxy(MessageReceiverWithResponder* receiver);
+  ~ControlMessageProxy();
 
   void QueryVersion(const base::Callback<void(uint32_t)>& callback);
   void RequireVersion(uint32_t version);
 
- protected:
+  void FlushForTesting();
+  void OnConnectionError();
+
+ private:
+  void RunFlushForTestingClosure();
+
   // Not owned.
   MessageReceiverWithResponder* receiver_;
-  SerializationContext context_;
+  bool encountered_error_ = false;
+
+  base::Closure run_loop_quit_closure_;
 
   DISALLOW_COPY_AND_ASSIGN(ControlMessageProxy);
 };
diff --git a/mojo/public/cpp/bindings/lib/clone_equals_util.h b/mojo/public/cpp/bindings/lib/equals_traits.h
index f7bd898..53c7dce 100644
--- a/mojo/public/cpp/bindings/lib/clone_equals_util.h
+++ b/mojo/public/cpp/bindings/lib/equals_traits.h
@@ -2,8 +2,8 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_CLONE_EQUALS_UTIL_H_
-#define MOJO_PUBLIC_CPP_BINDINGS_LIB_CLONE_EQUALS_UTIL_H_
+#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_EQUALS_TRAITS_H_
+#define MOJO_PUBLIC_CPP_BINDINGS_LIB_EQUALS_TRAITS_H_
 
 #include <type_traits>
 #include <unordered_map>
@@ -16,73 +16,6 @@ namespace mojo {
 namespace internal {
 
 template <typename T>
-struct HasCloneMethod {
-  template <typename U>
-  static char Test(decltype(&U::Clone));
-  template <typename U>
-  static int Test(...);
-  static const bool value = sizeof(Test<T>(0)) == sizeof(char);
-
- private:
-  EnsureTypeIsComplete<T> check_t_;
-};
-
-template <typename T, bool has_clone_method = HasCloneMethod<T>::value>
-struct CloneTraits;
-
-template <typename T>
-T Clone(const T& input);
-
-template <typename T>
-struct CloneTraits<T, true> {
-  static T Clone(const T& input) { return input.Clone(); }
-};
-
-template <typename T>
-struct CloneTraits<T, false> {
-  static T Clone(const T& input) { return input; }
-};
-
-template <typename T>
-struct CloneTraits<base::Optional<T>, false> {
-  static base::Optional<T> Clone(const base::Optional<T>& input) {
-    if (!input)
-      return base::nullopt;
-
-    return base::Optional<T>(internal::Clone(*input));
-  }
-};
-
-template <typename T>
-struct CloneTraits<std::vector<T>, false> {
-  static std::vector<T> Clone(const std::vector<T>& input) {
-    std::vector<T> result;
-    result.reserve(input.size());
-    for (const auto& element : input)
-      result.push_back(internal::Clone(element));
-
-    return result;
-  }
-};
-
-template <typename K, typename V>
-struct CloneTraits<std::unordered_map<K, V>, false> {
-  static std::unordered_map<K, V> Clone(const std::unordered_map<K, V>& input) {
-    std::unordered_map<K, V> result;
-    for (const auto& element : input) {
-      result.insert(std::make_pair(internal::Clone(element.first),
-                                   internal::Clone(element.second)));
-    }
-    return result;
-  }
-};
-
-template <typename T>
-T Clone(const T& input) {
-  return CloneTraits<T>::Clone(input);
-};
-
-template <typename T>
 struct HasEqualsMethod {
   template <typename U>
   static char Test(decltype(&U::Equals));
@@ -158,4 +91,4 @@ bool Equals(const T& a, const T& b) {
 }  // namespace internal
 }  // namespace mojo
 
-#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_CLONE_EQUALS_UTIL_H_
+#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_EQUALS_TRAITS_H_
diff --git a/mojo/public/cpp/bindings/lib/filter_chain.cc b/mojo/public/cpp/bindings/lib/filter_chain.cc
index 899bac1..5d919fe 100644
--- a/mojo/public/cpp/bindings/lib/filter_chain.cc
+++ b/mojo/public/cpp/bindings/lib/filter_chain.cc
@@ -2,14 +2,13 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "mojo/public/cpp/bindings/lib/filter_chain.h"
+#include "mojo/public/cpp/bindings/filter_chain.h"
 
 #include <algorithm>
 
 #include "base/logging.h"
 
 namespace mojo {
-namespace internal {
 
 FilterChain::FilterChain(MessageReceiver* sink) : sink_(sink) {
 }
@@ -26,24 +25,23 @@ FilterChain& FilterChain::operator=(FilterChain&& other) {
 }
 
 FilterChain::~FilterChain() {
-  for (std::vector<MessageFilter*>::iterator iter = filters_.begin();
-       iter != filters_.end();
-       ++iter) {
-    delete *iter;
-  }
 }
 
 void FilterChain::SetSink(MessageReceiver* sink) {
   DCHECK(!sink_);
   sink_ = sink;
-  if (!filters_.empty())
-    filters_.back()->set_sink(sink);
 }
 
-MessageReceiver* FilterChain::GetHead() {
+bool FilterChain::Accept(Message* message) {
   DCHECK(sink_);
-  return filters_.empty() ? sink_ : filters_.front();
+  for (auto& filter : filters_)
+    if (!filter->Accept(message))
+      return false;
+  return sink_->Accept(message);
+}
+
+void FilterChain::Append(std::unique_ptr<MessageReceiver> filter) {
+  filters_.emplace_back(std::move(filter));
 }
 
-}  // namespace internal
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/filter_chain.h b/mojo/public/cpp/bindings/lib/filter_chain.h
deleted file mode 100644
index 447be3d..0000000
--- a/mojo/public/cpp/bindings/lib/filter_chain.h
+++ /dev/null
@@ -1,66 +0,0 @@
-// Copyright 2014 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_FILTER_CHAIN_H_
-#define MOJO_PUBLIC_CPP_BINDINGS_LIB_FILTER_CHAIN_H_
-
-#include <utility>
-#include <vector>
-
-#include "base/macros.h"
-#include "mojo/public/cpp/bindings/message.h"
-#include "mojo/public/cpp/bindings/message_filter.h"
-
-namespace mojo {
-namespace internal {
-
-class FilterChain {
- public:
-  // Doesn't take ownership of |sink|. Therefore |sink| has to stay alive while
-  // this object is alive.
-  explicit FilterChain(MessageReceiver* sink = nullptr);
-
-  FilterChain(FilterChain&& other);
-  FilterChain& operator=(FilterChain&& other);
-  ~FilterChain();
-
-  template <typename FilterType, typename... Args>
-  inline void Append(Args&&... args);
-
-  // Doesn't take ownership of |sink|. Therefore |sink| has to stay alive while
-  // this object is alive.
-  void SetSink(MessageReceiver* sink);
-
-  // Returns a receiver to accept messages. Messages flow through all filters in
-  // the same order as they were appended to the chain. If all filters allow a
-  // message to pass, it will be forwarded to |sink_|.
-  // The returned value is invalidated when this object goes away.
-  MessageReceiver* GetHead();
-
- private:
-  // Owned by this object.
-  // TODO(dcheng): Use unique_ptr.
-  std::vector<MessageFilter*> filters_;
-
-  MessageReceiver* sink_;
-
-  DISALLOW_COPY_AND_ASSIGN(FilterChain);
-};
-
-template <typename FilterType, typename... Args>
-inline void FilterChain::Append(Args&&... args) {
-  FilterType* filter = new FilterType(std::forward<Args>(args)..., sink_);
-  if (!filters_.empty())
-    filters_.back()->set_sink(filter);
-  filters_.push_back(filter);
-}
-
-template <>
-inline void FilterChain::Append<PassThroughFilter>() {
-}
-
-}  // namespace internal
-}  // namespace mojo
-
-#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_FILTER_CHAIN_H_
diff --git a/mojo/public/cpp/bindings/lib/fixed_buffer.cc b/mojo/public/cpp/bindings/lib/fixed_buffer.cc
index 50b8a21..725a193 100644
--- a/mojo/public/cpp/bindings/lib/fixed_buffer.cc
+++ b/mojo/public/cpp/bindings/lib/fixed_buffer.cc
@@ -4,56 +4,25 @@
 
 #include "mojo/public/cpp/bindings/lib/fixed_buffer.h"
 
-#include <stddef.h>
 #include <stdlib.h>
 
-#include <algorithm>
-
-#include "base/logging.h"
-#include "mojo/public/cpp/bindings/lib/serialization_util.h"
-
 namespace mojo {
 namespace internal {
 
-FixedBuffer::FixedBuffer() : ptr_(nullptr), cursor_(0), size_(0) {}
-
-void FixedBuffer::Initialize(void* memory, size_t size) {
-  DCHECK(size == internal::Align(size));
-
-  ptr_ = static_cast<char*>(memory);
-  cursor_ = 0;
-  size_ = size;
-}
-
-void* FixedBuffer::Allocate(size_t delta) {
-  delta = internal::Align(delta);
-
-  if (delta == 0 || delta > size_ - cursor_) {
-    NOTREACHED();
-    return nullptr;
-  }
-
-  char* result = ptr_ + cursor_;
-  cursor_ += delta;
-
-  return result;
-}
-
 FixedBufferForTesting::FixedBufferForTesting(size_t size) {
-  size_ = internal::Align(size);
+  size = internal::Align(size);
   // Use calloc here to ensure all message memory is zero'd out.
-  ptr_ = static_cast<char*>(calloc(size_, 1));
+  void* ptr = calloc(size, 1);
+  Initialize(ptr, size);
 }
 
 FixedBufferForTesting::~FixedBufferForTesting() {
-  free(ptr_);
+  free(data());
 }
 
 void* FixedBufferForTesting::Leak() {
-  char* ptr = ptr_;
-  ptr_ = nullptr;
-  cursor_ = 0;
-  size_ = 0;
+  void* ptr = data();
+  Initialize(nullptr, 0);
   return ptr;
 }
 
diff --git a/mojo/public/cpp/bindings/lib/fixed_buffer.h b/mojo/public/cpp/bindings/lib/fixed_buffer.h
index 9a5704b..070b0c8 100644
--- a/mojo/public/cpp/bindings/lib/fixed_buffer.h
+++ b/mojo/public/cpp/bindings/lib/fixed_buffer.h
@@ -7,62 +7,21 @@
 
 #include <stddef.h>
 
+#include "base/compiler_specific.h"
 #include "base/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/buffer.h"
 
 namespace mojo {
 namespace internal {
 
-// FixedBuffer provides a simple way to allocate objects within a fixed chunk
-// of memory. Objects are allocated by calling the |Allocate| method, which
-// extends the buffer accordingly. Objects allocated in this way are not freed
-// explicitly. Instead, they remain valid so long as the FixedBuffer remains
-// valid.  The Leak method may be used to steal the underlying memory from the
-// FixedBuffer.
-//
-// Typical usage:
-//
-//   {
-//     FixedBuffer buf(8 + 8);
-//
-//     int* a = static_cast<int*>(buf->Allocate(sizeof(int)));
-//     *a = 2;
-//
-//     double* b = static_cast<double*>(buf->Allocate(sizeof(double)));
-//     *b = 3.14f;
-//
-//     void* data = buf.Leak();
-//     Process(data);
-//
-//     free(data);
-//   }
-
-class FixedBuffer : public Buffer {
- public:
-  FixedBuffer();
-
-  // |size| should be aligned using internal::Align.
-  void Initialize(void* memory, size_t size);
-
-  size_t size() const { return size_; }
-
-  // Grows the buffer by |num_bytes| and returns a pointer to the start of the
-  // addition. The resulting address is 8-byte aligned, and the content of the
-  // memory is zero-filled.
-  void* Allocate(size_t num_bytes) override;
-
- protected:
-  char* ptr_;
-  size_t cursor_;
-  size_t size_;
-
-  DISALLOW_COPY_AND_ASSIGN(FixedBuffer);
-};
-
-class FixedBufferForTesting : public FixedBuffer {
+// FixedBufferForTesting owns its buffer. The Leak method may be used to steal
+// the underlying memory.
+class MOJO_CPP_BINDINGS_EXPORT FixedBufferForTesting
+    : NON_EXPORTED_BASE(public Buffer) {
  public:
   explicit FixedBufferForTesting(size_t size);
-  ~FixedBufferForTesting() override;
+  ~FixedBufferForTesting();
 
   // Returns the internal memory owned by the Buffer to the caller. The Buffer
   // relinquishes its pointer, effectively resetting the state of the Buffer
diff --git a/mojo/public/cpp/bindings/lib/handle_interface_serialization.h b/mojo/public/cpp/bindings/lib/handle_interface_serialization.h
index 344c2ca..14ed21f 100644
--- a/mojo/public/cpp/bindings/lib/handle_interface_serialization.h
+++ b/mojo/public/cpp/bindings/lib/handle_interface_serialization.h
@@ -5,9 +5,12 @@
 #ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_HANDLE_INTERFACE_SERIALIZATION_H_
 #define MOJO_PUBLIC_CPP_BINDINGS_LIB_HANDLE_INTERFACE_SERIALIZATION_H_
 
+#include <type_traits>
+
 #include "mojo/public/cpp/bindings/associated_group_controller.h"
 #include "mojo/public/cpp/bindings/associated_interface_ptr_info.h"
 #include "mojo/public/cpp/bindings/associated_interface_request.h"
+#include "mojo/public/cpp/bindings/interface_data_view.h"
 #include "mojo/public/cpp/bindings/interface_ptr.h"
 #include "mojo/public/cpp/bindings/interface_request.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
@@ -18,52 +21,98 @@
 namespace mojo {
 namespace internal {
 
-template <typename T>
-struct Serializer<AssociatedInterfacePtrInfo<T>,
+template <typename Base, typename T>
+struct Serializer<AssociatedInterfacePtrInfoDataView<Base>,
                   AssociatedInterfacePtrInfo<T>> {
+  static_assert(std::is_base_of<Base, T>::value, "Interface type mismatch.");
+
+  static size_t PrepareToSerialize(const AssociatedInterfacePtrInfo<T>& input,
+                                   SerializationContext* context) {
+    if (input.handle().is_valid())
+      context->associated_endpoint_count++;
+    return 0;
+  }
+
   static void Serialize(AssociatedInterfacePtrInfo<T>& input,
                         AssociatedInterface_Data* output,
                         SerializationContext* context) {
-    DCHECK(!input.handle().is_valid() || !input.handle().is_local());
-    DCHECK_EQ(input.handle().group_controller(),
-              context->group_controller.get());
+    DCHECK(!input.handle().is_valid() || input.handle().pending_association());
+    if (input.handle().is_valid()) {
+      // Set to the index of the element pushed to the back of the vector.
+      output->handle.value =
+          static_cast<uint32_t>(context->associated_endpoint_handles.size());
+      context->associated_endpoint_handles.push_back(input.PassHandle());
+    } else {
+      output->handle.value = kEncodedInvalidHandleValue;
+    }
     output->version = input.version();
-    output->interface_id = input.PassHandle().release();
   }
 
   static bool Deserialize(AssociatedInterface_Data* input,
                           AssociatedInterfacePtrInfo<T>* output,
                           SerializationContext* context) {
-    output->set_handle(context->group_controller->CreateLocalEndpointHandle(
-        FetchAndReset(&input->interface_id)));
+    if (input->handle.is_valid()) {
+      DCHECK_LT(input->handle.value,
+                context->associated_endpoint_handles.size());
+      output->set_handle(
+          std::move(context->associated_endpoint_handles[input->handle.value]));
+    } else {
+      output->set_handle(ScopedInterfaceEndpointHandle());
+    }
     output->set_version(input->version);
     return true;
   }
 };
 
-template <typename T>
-struct Serializer<AssociatedInterfaceRequest<T>,
+template <typename Base, typename T>
+struct Serializer<AssociatedInterfaceRequestDataView<Base>,
                   AssociatedInterfaceRequest<T>> {
+  static_assert(std::is_base_of<Base, T>::value, "Interface type mismatch.");
+
+  static size_t PrepareToSerialize(const AssociatedInterfaceRequest<T>& input,
+                                   SerializationContext* context) {
+    if (input.handle().is_valid())
+      context->associated_endpoint_count++;
+    return 0;
+  }
+
   static void Serialize(AssociatedInterfaceRequest<T>& input,
-                        AssociatedInterfaceRequest_Data* output,
+                        AssociatedEndpointHandle_Data* output,
                         SerializationContext* context) {
-    DCHECK(!input.handle().is_valid() || !input.handle().is_local());
-    DCHECK_EQ(input.handle().group_controller(),
-              context->group_controller.get());
-    output->interface_id = input.PassHandle().release();
+    DCHECK(!input.handle().is_valid() || input.handle().pending_association());
+    if (input.handle().is_valid()) {
+      // Set to the index of the element pushed to the back of the vector.
+      output->value =
+          static_cast<uint32_t>(context->associated_endpoint_handles.size());
+      context->associated_endpoint_handles.push_back(input.PassHandle());
+    } else {
+      output->value = kEncodedInvalidHandleValue;
+    }
   }
 
-  static bool Deserialize(AssociatedInterfaceRequest_Data* input,
+  static bool Deserialize(AssociatedEndpointHandle_Data* input,
                           AssociatedInterfaceRequest<T>* output,
                           SerializationContext* context) {
-    output->Bind(context->group_controller->CreateLocalEndpointHandle(
-        FetchAndReset(&input->interface_id)));
+    if (input->is_valid()) {
+      DCHECK_LT(input->value, context->associated_endpoint_handles.size());
+      output->Bind(
+          std::move(context->associated_endpoint_handles[input->value]));
+    } else {
+      output->Bind(ScopedInterfaceEndpointHandle());
+    }
     return true;
   }
 };
 
-template <typename T>
-struct Serializer<InterfacePtr<T>, InterfacePtr<T>> {
+template <typename Base, typename T>
+struct Serializer<InterfacePtrDataView<Base>, InterfacePtr<T>> {
+  static_assert(std::is_base_of<Base, T>::value, "Interface type mismatch.");
+
+  static size_t PrepareToSerialize(const InterfacePtr<T>& input,
+                                   SerializationContext* context) {
+    return 0;
+  }
+
   static void Serialize(InterfacePtr<T>& input,
                         Interface_Data* output,
                         SerializationContext* context) {
@@ -82,8 +131,15 @@ struct Serializer<InterfacePtr<T>, InterfacePtr<T>> {
   }
 };
 
-template <typename T>
-struct Serializer<InterfaceRequest<T>, InterfaceRequest<T>> {
+template <typename Base, typename T>
+struct Serializer<InterfaceRequestDataView<Base>, InterfaceRequest<T>> {
+  static_assert(std::is_base_of<Base, T>::value, "Interface type mismatch.");
+
+  static size_t PrepareToSerialize(const InterfaceRequest<T>& input,
+                                   SerializationContext* context) {
+    return 0;
+  }
+
   static void Serialize(InterfaceRequest<T>& input,
                         Handle_Data* output,
                         SerializationContext* context) {
@@ -100,6 +156,11 @@ struct Serializer<InterfaceRequest<T>, InterfaceRequest<T>> {
 
 template <typename T>
 struct Serializer<ScopedHandleBase<T>, ScopedHandleBase<T>> {
+  static size_t PrepareToSerialize(const ScopedHandleBase<T>& input,
+                                   SerializationContext* context) {
+    return 0;
+  }
+
   static void Serialize(ScopedHandleBase<T>& input,
                         Handle_Data* output,
                         SerializationContext* context) {
diff --git a/mojo/public/cpp/bindings/lib/hash_util.h b/mojo/public/cpp/bindings/lib/hash_util.h
new file mode 100644
index 0000000..93280d6
--- /dev/null
+++ b/mojo/public/cpp/bindings/lib/hash_util.h
@@ -0,0 +1,84 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_HASH_UTIL_H_
+#define MOJO_PUBLIC_CPP_BINDINGS_LIB_HASH_UTIL_H_
+
+#include <cstring>
+#include <functional>
+#include <type_traits>
+#include <vector>
+
+#include "base/optional.h"
+#include "mojo/public/cpp/bindings/lib/template_util.h"
+
+namespace mojo {
+namespace internal {
+
+template <typename T>
+size_t HashCombine(size_t seed, const T& value) {
+  // Based on proposal in:
+  // http://www.open-std.org/JTC1/SC22/WG21/docs/papers/2005/n1756.pdf
+  return seed ^ (std::hash<T>()(value) + (seed << 6) + (seed >> 2));
+}
+
+template <typename T>
+struct HasHashMethod {
+  template <typename U>
+  static char Test(decltype(&U::Hash));
+  template <typename U>
+  static int Test(...);
+  static const bool value = sizeof(Test<T>(0)) == sizeof(char);
+
+ private:
+  EnsureTypeIsComplete<T> check_t_;
+};
+
+template <typename T, bool has_hash_method = HasHashMethod<T>::value>
+struct HashTraits;
+
+template <typename T>
+size_t Hash(size_t seed, const T& value);
+
+template <typename T>
+struct HashTraits<T, true> {
+  static size_t Hash(size_t seed, const T& value) { return value.Hash(seed); }
+};
+
+template <typename T>
+struct HashTraits<T, false> {
+  static size_t Hash(size_t seed, const T& value) {
+    return HashCombine(seed, value);
+  }
+};
+
+template <typename T>
+struct HashTraits<std::vector<T>, false> {
+  static size_t Hash(size_t seed, const std::vector<T>& value) {
+    for (const auto& element : value) {
+      seed = HashCombine(seed, element);
+    }
+    return seed;
+  }
+};
+
+template <typename T>
+struct HashTraits<base::Optional<std::vector<T>>, false> {
+  static size_t Hash(size_t seed, const base::Optional<std::vector<T>>& value) {
+    if (!value)
+      return HashCombine(seed, 0);
+
+    return Hash(seed, *value);
+  }
+};
+
+template <typename T>
+size_t Hash(size_t seed, const T& value) {
+  return HashTraits<T>::Hash(seed, value);
+}
+
+}  // namespace internal
+}  // namespace mojo
+
+#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_HASH_UTIL_H_
diff --git a/mojo/public/cpp/bindings/lib/interface_endpoint_client.cc b/mojo/public/cpp/bindings/lib/interface_endpoint_client.cc
index e1f388a..3eca5a1 100644
--- a/mojo/public/cpp/bindings/lib/interface_endpoint_client.cc
+++ b/mojo/public/cpp/bindings/lib/interface_endpoint_client.cc
@@ -10,6 +10,7 @@
 
 #include "base/bind.h"
 #include "base/location.h"
+#include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/single_thread_task_runner.h"
@@ -17,6 +18,7 @@
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/associated_group_controller.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
+#include "mojo/public/cpp/bindings/lib/validation_util.h"
 #include "mojo/public/cpp/bindings/sync_call_restrictions.h"
 
 namespace mojo {
@@ -45,7 +47,7 @@ class ResponderThunk : public MessageReceiverWithStatus {
         task_runner_(std::move(runner)) {}
   ~ResponderThunk() override {
     if (!accept_was_invoked_) {
-      // The Mojo application handled a message that was expecting a response
+      // The Service handled a message that was expecting a response
       // but did not send a response.
       // We raise an error to signal the calling application that an error
       // condition occurred. Without this the calling application would have no
@@ -132,47 +134,47 @@ bool InterfaceEndpointClient::HandleIncomingMessageThunk::Accept(
 InterfaceEndpointClient::InterfaceEndpointClient(
     ScopedInterfaceEndpointHandle handle,
     MessageReceiverWithResponderStatus* receiver,
-    std::unique_ptr<MessageFilter> payload_validator,
+    std::unique_ptr<MessageReceiver> payload_validator,
     bool expect_sync_requests,
-    scoped_refptr<base::SingleThreadTaskRunner> runner)
-    : handle_(std::move(handle)),
+    scoped_refptr<base::SingleThreadTaskRunner> runner,
+    uint32_t interface_version)
+    : expect_sync_requests_(expect_sync_requests),
+      handle_(std::move(handle)),
       incoming_receiver_(receiver),
-      payload_validator_(std::move(payload_validator)),
       thunk_(this),
-      next_request_id_(1),
-      encountered_error_(false),
+      filters_(&thunk_),
       task_runner_(std::move(runner)),
+      control_message_proxy_(this),
+      control_message_handler_(interface_version),
       weak_ptr_factory_(this) {
   DCHECK(handle_.is_valid());
-  DCHECK(handle_.is_local());
 
   // TODO(yzshen): the way to use validator (or message filter in general)
   // directly is a little awkward.
-  payload_validator_->set_sink(&thunk_);
+  if (payload_validator)
+    filters_.Append(std::move(payload_validator));
 
-  controller_ = handle_.group_controller()->AttachEndpointClient(
-      handle_, this, task_runner_);
-  if (expect_sync_requests)
-    controller_->AllowWokenUpBySyncWatchOnSameThread();
+  if (handle_.pending_association()) {
+    handle_.SetAssociationEventHandler(base::Bind(
+        &InterfaceEndpointClient::OnAssociationEvent, base::Unretained(this)));
+  } else {
+    InitControllerIfNecessary();
+  }
 }
 
 InterfaceEndpointClient::~InterfaceEndpointClient() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  handle_.group_controller()->DetachEndpointClient(handle_);
+  if (controller_)
+    handle_.group_controller()->DetachEndpointClient(handle_);
 }
 
 AssociatedGroup* InterfaceEndpointClient::associated_group() {
   if (!associated_group_)
-    associated_group_ = handle_.group_controller()->CreateAssociatedGroup();
+    associated_group_ = base::MakeUnique<AssociatedGroup>(handle_);
   return associated_group_.get();
 }
 
-uint32_t InterfaceEndpointClient::interface_id() const {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  return handle_.id();
-}
-
 ScopedInterfaceEndpointHandle InterfaceEndpointClient::PassHandle() {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(!has_pending_responders());
@@ -180,38 +182,73 @@ ScopedInterfaceEndpointHandle InterfaceEndpointClient::PassHandle() {
   if (!handle_.is_valid())
     return ScopedInterfaceEndpointHandle();
 
-  controller_ = nullptr;
-  handle_.group_controller()->DetachEndpointClient(handle_);
+  handle_.SetAssociationEventHandler(
+      ScopedInterfaceEndpointHandle::AssociationEventCallback());
+
+  if (controller_) {
+    controller_ = nullptr;
+    handle_.group_controller()->DetachEndpointClient(handle_);
+  }
 
   return std::move(handle_);
 }
 
+void InterfaceEndpointClient::AddFilter(
+    std::unique_ptr<MessageReceiver> filter) {
+  filters_.Append(std::move(filter));
+}
+
 void InterfaceEndpointClient::RaiseError() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  handle_.group_controller()->RaiseError();
+  if (!handle_.pending_association())
+    handle_.group_controller()->RaiseError();
+}
+
+void InterfaceEndpointClient::CloseWithReason(uint32_t custom_reason,
+                                              const std::string& description) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  auto handle = PassHandle();
+  handle.ResetWithReason(custom_reason, description);
 }
 
 bool InterfaceEndpointClient::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  DCHECK(controller_);
   DCHECK(!message->has_flag(Message::kFlagExpectsResponse));
+  DCHECK(!handle_.pending_association());
+
+  // This has to been done even if connection error has occurred. For example,
+  // the message contains a pending associated request. The user may try to use
+  // the corresponding associated interface pointer after sending this message.
+  // That associated interface pointer has to join an associated group in order
+  // to work properly.
+  if (!message->associated_endpoint_handles()->empty())
+    message->SerializeAssociatedEndpointHandles(handle_.group_controller());
 
   if (encountered_error_)
     return false;
 
+  InitControllerIfNecessary();
+
   return controller_->SendMessage(message);
 }
 
 bool InterfaceEndpointClient::AcceptWithResponder(Message* message,
                                                   MessageReceiver* responder) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  DCHECK(controller_);
   DCHECK(message->has_flag(Message::kFlagExpectsResponse));
+  DCHECK(!handle_.pending_association());
+
+  // Please see comments in Accept().
+  if (!message->associated_endpoint_handles()->empty())
+    message->SerializeAssociatedEndpointHandles(handle_.group_controller());
 
   if (encountered_error_)
     return false;
 
+  InitControllerIfNecessary();
+
   // Reserve 0 in case we want it to convey special meaning in the future.
   uint64_t request_id = next_request_id_++;
   if (request_id == 0)
@@ -234,20 +271,18 @@ bool InterfaceEndpointClient::AcceptWithResponder(Message* message,
   bool response_received = false;
   std::unique_ptr<MessageReceiver> sync_responder(responder);
   sync_responses_.insert(std::make_pair(
-      request_id, base::WrapUnique(new SyncResponseInfo(&response_received))));
+      request_id, base::MakeUnique<SyncResponseInfo>(&response_received)));
 
   base::WeakPtr<InterfaceEndpointClient> weak_self =
       weak_ptr_factory_.GetWeakPtr();
   controller_->SyncWatch(&response_received);
   // Make sure that this instance hasn't been destroyed.
   if (weak_self) {
-    DCHECK(ContainsKey(sync_responses_, request_id));
+    DCHECK(base::ContainsKey(sync_responses_, request_id));
     auto iter = sync_responses_.find(request_id);
     DCHECK_EQ(&response_received, iter->second->response_received);
-    if (response_received) {
-      std::unique_ptr<Message> response = std::move(iter->second->response);
-      ignore_result(sync_responder->Accept(response.get()));
-    }
+    if (response_received)
+      ignore_result(sync_responder->Accept(&iter->second->response));
     sync_responses_.erase(iter);
   }
 
@@ -257,30 +292,97 @@ bool InterfaceEndpointClient::AcceptWithResponder(Message* message,
 
 bool InterfaceEndpointClient::HandleIncomingMessage(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
-
-  return payload_validator_->Accept(message);
+  return filters_.Accept(message);
 }
 
-void InterfaceEndpointClient::NotifyError() {
+void InterfaceEndpointClient::NotifyError(
+    const base::Optional<DisconnectReason>& reason) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   if (encountered_error_)
     return;
   encountered_error_ = true;
-  if (!error_handler_.is_null())
-    error_handler_.Run();
+
+  // Response callbacks may hold on to resource, and there's no need to keep
+  // them alive any longer. Note that it's allowed that a pending response
+  // callback may own this endpoint, so we simply move the responders onto the
+  // stack here and let them be destroyed when the stack unwinds.
+  AsyncResponderMap responders = std::move(async_responders_);
+
+  control_message_proxy_.OnConnectionError();
+
+  if (!error_handler_.is_null()) {
+    base::Closure error_handler = std::move(error_handler_);
+    error_handler.Run();
+  } else if (!error_with_reason_handler_.is_null()) {
+    ConnectionErrorWithReasonCallback error_with_reason_handler =
+        std::move(error_with_reason_handler_);
+    if (reason) {
+      error_with_reason_handler.Run(reason->custom_reason, reason->description);
+    } else {
+      error_with_reason_handler.Run(0, std::string());
+    }
+  }
+}
+
+void InterfaceEndpointClient::QueryVersion(
+    const base::Callback<void(uint32_t)>& callback) {
+  control_message_proxy_.QueryVersion(callback);
+}
+
+void InterfaceEndpointClient::RequireVersion(uint32_t version) {
+  control_message_proxy_.RequireVersion(version);
+}
+
+void InterfaceEndpointClient::FlushForTesting() {
+  control_message_proxy_.FlushForTesting();
+}
+
+void InterfaceEndpointClient::InitControllerIfNecessary() {
+  if (controller_ || handle_.pending_association())
+    return;
+
+  controller_ = handle_.group_controller()->AttachEndpointClient(handle_, this,
+                                                                 task_runner_);
+  if (expect_sync_requests_)
+    controller_->AllowWokenUpBySyncWatchOnSameThread();
+}
+
+void InterfaceEndpointClient::OnAssociationEvent(
+    ScopedInterfaceEndpointHandle::AssociationEvent event) {
+  if (event == ScopedInterfaceEndpointHandle::ASSOCIATED) {
+    InitControllerIfNecessary();
+  } else if (event ==
+             ScopedInterfaceEndpointHandle::PEER_CLOSED_BEFORE_ASSOCIATION) {
+    task_runner_->PostTask(FROM_HERE,
+                           base::Bind(&InterfaceEndpointClient::NotifyError,
+                                      weak_ptr_factory_.GetWeakPtr(),
+                                      handle_.disconnect_reason()));
+  }
 }
 
 bool InterfaceEndpointClient::HandleValidatedMessage(Message* message) {
   DCHECK_EQ(handle_.id(), message->interface_id());
 
-  if (message->has_flag(Message::kFlagExpectsResponse)) {
-    if (!incoming_receiver_)
-      return false;
+  if (encountered_error_) {
+    // This message is received after error has been encountered. For associated
+    // interfaces, this means the remote side sends a
+    // PeerAssociatedEndpointClosed event but continues to send more messages
+    // for the same interface. Close the pipe because this shouldn't happen.
+    DVLOG(1) << "A message is received for an interface after it has been "
+             << "disconnected. Closing the pipe.";
+    return false;
+  }
 
+  if (message->has_flag(Message::kFlagExpectsResponse)) {
     MessageReceiverWithStatus* responder =
         new ResponderThunk(weak_ptr_factory_.GetWeakPtr(), task_runner_);
-    bool ok = incoming_receiver_->AcceptWithResponder(message, responder);
+    bool ok = false;
+    if (mojo::internal::ControlMessageHandler::IsControlMessage(message)) {
+      ok = control_message_handler_.AcceptWithResponder(message, responder);
+    } else {
+      ok = incoming_receiver_->AcceptWithResponder(message, responder);
+    }
     if (!ok)
       delete responder;
     return ok;
@@ -291,8 +393,7 @@ bool InterfaceEndpointClient::HandleValidatedMessage(Message* message) {
       auto it = sync_responses_.find(request_id);
       if (it == sync_responses_.end())
         return false;
-      it->second->response.reset(new Message());
-      message->MoveTo(it->second->response.get());
+      it->second->response = std::move(*message);
       *it->second->response_received = true;
       return true;
     }
@@ -304,8 +405,8 @@ bool InterfaceEndpointClient::HandleValidatedMessage(Message* message) {
     async_responders_.erase(it);
     return responder->Accept(message);
   } else {
-    if (!incoming_receiver_)
-      return false;
+    if (mojo::internal::ControlMessageHandler::IsControlMessage(message))
+      return control_message_handler_.Accept(message);
 
     return incoming_receiver_->Accept(message);
   }
diff --git a/mojo/public/cpp/bindings/lib/interface_ptr_state.h b/mojo/public/cpp/bindings/lib/interface_ptr_state.h
index 584933e..8f5b4ff 100644
--- a/mojo/public/cpp/bindings/lib/interface_ptr_state.h
+++ b/mojo/public/cpp/bindings/lib/interface_ptr_state.h
@@ -9,6 +9,7 @@
 
 #include <algorithm>  // For |std::swap()|.
 #include <memory>
+#include <string>
 #include <utility>
 
 #include "base/bind.h"
@@ -19,174 +20,20 @@
 #include "base/memory/ref_counted.h"
 #include "base/single_thread_task_runner.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
+#include "mojo/public/cpp/bindings/connection_error_callback.h"
+#include "mojo/public/cpp/bindings/filter_chain.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_id.h"
 #include "mojo/public/cpp/bindings/interface_ptr_info.h"
-#include "mojo/public/cpp/bindings/lib/control_message_proxy.h"
-#include "mojo/public/cpp/bindings/lib/filter_chain.h"
 #include "mojo/public/cpp/bindings/lib/multiplex_router.h"
-#include "mojo/public/cpp/bindings/lib/router.h"
 #include "mojo/public/cpp/bindings/message_header_validator.h"
 #include "mojo/public/cpp/bindings/scoped_interface_endpoint_handle.h"
 
 namespace mojo {
 namespace internal {
 
-template <typename Interface, bool use_multiplex_router>
-class InterfacePtrState;
-
-// Uses a single-threaded, dedicated router. If |Interface| doesn't have any
-// methods to pass associated interface pointers or requests, there won't be
-// multiple interfaces running on the underlying message pipe. In that case, we
-// can use this specialization to reduce cost.
-template <typename Interface>
-class InterfacePtrState<Interface, false> {
- public:
-  InterfacePtrState() : proxy_(nullptr), router_(nullptr), version_(0u) {}
-
-  ~InterfacePtrState() {
-    // Destruction order matters here. We delete |proxy_| first, even though
-    // |router_| may have a reference to it, so that destructors for any request
-    // callbacks still pending can interact with the InterfacePtr.
-    delete proxy_;
-    delete router_;
-  }
-
-  Interface* instance() {
-    ConfigureProxyIfNecessary();
-
-    // This will be null if the object is not bound.
-    return proxy_;
-  }
-
-  uint32_t version() const { return version_; }
-
-  void QueryVersion(const base::Callback<void(uint32_t)>& callback) {
-    ConfigureProxyIfNecessary();
-
-    // Do a static cast in case the interface contains methods with the same
-    // name. It is safe to capture |this| because the callback won't be run
-    // after this object goes away.
-    static_cast<ControlMessageProxy*>(proxy_)->QueryVersion(
-        base::Bind(&InterfacePtrState::OnQueryVersion, base::Unretained(this),
-                   callback));
-  }
-
-  void RequireVersion(uint32_t version) {
-    ConfigureProxyIfNecessary();
-
-    if (version <= version_)
-      return;
-
-    version_ = version;
-    // Do a static cast in case the interface contains methods with the same
-    // name.
-    static_cast<ControlMessageProxy*>(proxy_)->RequireVersion(version);
-  }
-
-  void Swap(InterfacePtrState* other) {
-    using std::swap;
-    swap(other->proxy_, proxy_);
-    swap(other->router_, router_);
-    handle_.swap(other->handle_);
-    runner_.swap(other->runner_);
-    swap(other->version_, version_);
-  }
-
-  void Bind(InterfacePtrInfo<Interface> info,
-            scoped_refptr<base::SingleThreadTaskRunner> runner) {
-    DCHECK(!proxy_);
-    DCHECK(!router_);
-    DCHECK(!handle_.is_valid());
-    DCHECK_EQ(0u, version_);
-    DCHECK(info.is_valid());
-
-    handle_ = info.PassHandle();
-    version_ = info.version();
-    runner_ = std::move(runner);
-  }
-
-  bool HasAssociatedInterfaces() const { return false; }
-
-  // After this method is called, the object is in an invalid state and
-  // shouldn't be reused.
-  InterfacePtrInfo<Interface> PassInterface() {
-    return InterfacePtrInfo<Interface>(
-        router_ ? router_->PassMessagePipe() : std::move(handle_), version_);
-  }
-
-  bool is_bound() const { return handle_.is_valid() || router_; }
-
-  bool encountered_error() const {
-    return router_ ? router_->encountered_error() : false;
-  }
-
-  void set_connection_error_handler(const base::Closure& error_handler) {
-    ConfigureProxyIfNecessary();
-
-    DCHECK(router_);
-    router_->set_connection_error_handler(error_handler);
-  }
-
-  // Returns true if bound and awaiting a response to a message.
-  bool has_pending_callbacks() const {
-    return router_ && router_->has_pending_responders();
-  }
-
-  AssociatedGroup* associated_group() { return nullptr; }
-
-  void EnableTestingMode() {
-    ConfigureProxyIfNecessary();
-    router_->EnableTestingMode();
-  }
-
- private:
-  using Proxy = typename Interface::Proxy_;
-
-  void ConfigureProxyIfNecessary() {
-    // The proxy has been configured.
-    if (proxy_) {
-      DCHECK(router_);
-      return;
-    }
-    // The object hasn't been bound.
-    if (!handle_.is_valid())
-      return;
-
-    FilterChain filters;
-    filters.Append<MessageHeaderValidator>(Interface::Name_);
-    filters.Append<typename Interface::ResponseValidator_>();
-
-    router_ = new Router(std::move(handle_), std::move(filters), false,
-                         std::move(runner_));
-
-    proxy_ = new Proxy(router_);
-  }
-
-  void OnQueryVersion(const base::Callback<void(uint32_t)>& callback,
-                      uint32_t version) {
-    version_ = version;
-    callback.Run(version);
-  }
-
-  Proxy* proxy_;
-  Router* router_;
-
-  // |proxy_| and |router_| are not initialized until read/write with the
-  // message pipe handle is needed. |handle_| is valid between the Bind() call
-  // and the initialization of |proxy_| and |router_|.
-  ScopedMessagePipeHandle handle_;
-  scoped_refptr<base::SingleThreadTaskRunner> runner_;
-
-  uint32_t version_;
-
-  DISALLOW_COPY_AND_ASSIGN(InterfacePtrState);
-};
-
-// Uses a multiplexing router. If |Interface| has methods to pass associated
-// interface pointers or requests, this specialization should be used.
 template <typename Interface>
-class InterfacePtrState<Interface, true> {
+class InterfacePtrState {
  public:
   InterfacePtrState() : version_(0u) {}
 
@@ -209,13 +56,10 @@ class InterfacePtrState<Interface, true> {
   void QueryVersion(const base::Callback<void(uint32_t)>& callback) {
     ConfigureProxyIfNecessary();
 
-
-    // Do a static cast in case the interface contains methods with the same
-    // name. It is safe to capture |this| because the callback won't be run
-    // after this object goes away.
-    static_cast<ControlMessageProxy*>(proxy_.get())->QueryVersion(
-        base::Bind(&InterfacePtrState::OnQueryVersion, base::Unretained(this),
-                   callback));
+    // It is safe to capture |this| because the callback won't be run after this
+    // object goes away.
+    endpoint_client_->QueryVersion(base::Bind(
+        &InterfacePtrState::OnQueryVersion, base::Unretained(this), callback));
   }
 
   void RequireVersion(uint32_t version) {
@@ -225,9 +69,17 @@ class InterfacePtrState<Interface, true> {
       return;
 
     version_ = version;
-    // Do a static cast in case the interface contains methods with the same
-    // name.
-    static_cast<ControlMessageProxy*>(proxy_.get())->RequireVersion(version);
+    endpoint_client_->RequireVersion(version);
+  }
+
+  void FlushForTesting() {
+    ConfigureProxyIfNecessary();
+    endpoint_client_->FlushForTesting();
+  }
+
+  void CloseWithReason(uint32_t custom_reason, const std::string& description) {
+    ConfigureProxyIfNecessary();
+    endpoint_client_->CloseWithReason(custom_reason, description);
   }
 
   void Swap(InterfacePtrState* other) {
@@ -280,6 +132,14 @@ class InterfacePtrState<Interface, true> {
     endpoint_client_->set_connection_error_handler(error_handler);
   }
 
+  void set_connection_error_with_reason_handler(
+      const ConnectionErrorWithReasonCallback& error_handler) {
+    ConfigureProxyIfNecessary();
+
+    DCHECK(endpoint_client_);
+    endpoint_client_->set_connection_error_with_reason_handler(error_handler);
+  }
+
   // Returns true if bound and awaiting a response to a message.
   bool has_pending_callbacks() const {
     return endpoint_client_ && endpoint_client_->has_pending_responders();
@@ -295,6 +155,17 @@ class InterfacePtrState<Interface, true> {
     router_->EnableTestingMode();
   }
 
+  void ForwardMessage(Message message) {
+    ConfigureProxyIfNecessary();
+    endpoint_client_->Accept(&message);
+  }
+
+  void ForwardMessageWithResponder(Message message,
+                                   std::unique_ptr<MessageReceiver> responder) {
+    ConfigureProxyIfNecessary();
+    endpoint_client_->AcceptWithResponder(&message, responder.release());
+  }
+
  private:
   using Proxy = typename Interface::Proxy_;
 
@@ -309,15 +180,22 @@ class InterfacePtrState<Interface, true> {
     if (!handle_.is_valid())
       return;
 
-    router_ = new MultiplexRouter(true, std::move(handle_), runner_);
+    MultiplexRouter::Config config =
+        Interface::PassesAssociatedKinds_
+            ? MultiplexRouter::MULTI_INTERFACE
+            : (Interface::HasSyncMethods_
+                   ? MultiplexRouter::SINGLE_INTERFACE_WITH_SYNC_METHODS
+                   : MultiplexRouter::SINGLE_INTERFACE);
+    router_ = new MultiplexRouter(std::move(handle_), config, true, runner_);
     router_->SetMasterInterfaceName(Interface::Name_);
     endpoint_client_.reset(new InterfaceEndpointClient(
         router_->CreateLocalEndpointHandle(kMasterInterfaceId), nullptr,
         base::WrapUnique(new typename Interface::ResponseValidator_()), false,
-        std::move(runner_)));
+        std::move(runner_),
+        // The version is only queried from the client so the value passed here
+        // will not be used.
+        0u));
     proxy_.reset(new Proxy(endpoint_client_.get()));
-    proxy_->serialization_context()->group_controller =
-        endpoint_client_->group_controller();
   }
 
   void OnQueryVersion(const base::Callback<void(uint32_t)>& callback,
diff --git a/mojo/public/cpp/bindings/lib/map_serialization.h b/mojo/public/cpp/bindings/lib/map_serialization.h
index c28b835..718a763 100644
--- a/mojo/public/cpp/bindings/lib/map_serialization.h
+++ b/mojo/public/cpp/bindings/lib/map_serialization.h
@@ -8,11 +8,11 @@
 #include <type_traits>
 #include <vector>
 
-#include "mojo/public/cpp/bindings/array.h"
+#include "mojo/public/cpp/bindings/array_data_view.h"
 #include "mojo/public/cpp/bindings/lib/array_serialization.h"
 #include "mojo/public/cpp/bindings/lib/map_data_internal.h"
 #include "mojo/public/cpp/bindings/lib/serialization_forward.h"
-#include "mojo/public/cpp/bindings/map.h"
+#include "mojo/public/cpp/bindings/map_data_view.h"
 
 namespace mojo {
 namespace internal {
@@ -46,12 +46,15 @@ class MapKeyReader : public MapReaderBase<MaybeConstUserType> {
  public:
   using Base = MapReaderBase<MaybeConstUserType>;
   using Traits = typename Base::Traits;
+  using MaybeConstIterator = typename Base::MaybeConstIterator;
 
   explicit MapKeyReader(MaybeConstUserType& input) : Base(input) {}
   ~MapKeyReader() {}
 
-  const typename Traits::Key& GetNext() {
-    const typename Traits::Key& key = Traits::GetKey(this->iter_);
+  using GetNextResult =
+      decltype(Traits::GetKey(std::declval<MaybeConstIterator&>()));
+  GetNextResult GetNext() {
+    GetNextResult key = Traits::GetKey(this->iter_);
     Traits::AdvanceIterator(this->iter_);
     return key;
   }
@@ -78,17 +81,17 @@ class MapValueReader : public MapReaderBase<MaybeConstUserType> {
 };
 
 template <typename Key, typename Value, typename MaybeConstUserType>
-struct Serializer<Map<Key, Value>, MaybeConstUserType> {
+struct Serializer<MapDataView<Key, Value>, MaybeConstUserType> {
   using UserType = typename std::remove_const<MaybeConstUserType>::type;
   using Traits = MapTraits<UserType>;
   using UserKey = typename Traits::Key;
   using UserValue = typename Traits::Value;
-  using Data = typename MojomTypeTraits<Map<Key, Value>>::Data;
-  using KeyArraySerializer = ArraySerializer<Array<Key>,
+  using Data = typename MojomTypeTraits<MapDataView<Key, Value>>::Data;
+  using KeyArraySerializer = ArraySerializer<ArrayDataView<Key>,
                                              std::vector<UserKey>,
                                              MapKeyReader<MaybeConstUserType>>;
   using ValueArraySerializer =
-      ArraySerializer<Array<Value>,
+      ArraySerializer<ArrayDataView<Value>,
                       std::vector<UserValue>,
                       MapValueReader<MaybeConstUserType>>;
 
@@ -122,8 +125,8 @@ struct Serializer<Map<Key, Value>, MaybeConstUserType> {
 
     auto result = Data::New(buf);
     if (result) {
-      auto keys_ptr =
-          MojomTypeTraits<Array<Key>>::Data::New(Traits::GetSize(input), buf);
+      auto keys_ptr = MojomTypeTraits<ArrayDataView<Key>>::Data::New(
+          Traits::GetSize(input), buf);
       if (keys_ptr) {
         MapKeyReader<MaybeConstUserType> key_reader(input);
         KeyArraySerializer::SerializeElements(
@@ -132,8 +135,8 @@ struct Serializer<Map<Key, Value>, MaybeConstUserType> {
         result->keys.Set(keys_ptr);
       }
 
-      auto values_ptr =
-          MojomTypeTraits<Array<Value>>::Data::New(Traits::GetSize(input), buf);
+      auto values_ptr = MojomTypeTraits<ArrayDataView<Value>>::Data::New(
+          Traits::GetSize(input), buf);
       if (values_ptr) {
         MapValueReader<MaybeConstUserType> value_reader(input);
         ValueArraySerializer::SerializeElements(
diff --git a/mojo/public/cpp/bindings/lib/may_auto_lock.h b/mojo/public/cpp/bindings/lib/may_auto_lock.h
new file mode 100644
index 0000000..06091fe
--- /dev/null
+++ b/mojo/public/cpp/bindings/lib/may_auto_lock.h
@@ -0,0 +1,62 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_MAY_AUTO_LOCK_H_
+#define MOJO_PUBLIC_CPP_BINDINGS_LIB_MAY_AUTO_LOCK_H_
+
+#include "base/macros.h"
+#include "base/optional.h"
+#include "base/synchronization/lock.h"
+
+namespace mojo {
+namespace internal {
+
+// Similar to base::AutoLock, except that it does nothing if |lock| passed into
+// the constructor is null.
+class MayAutoLock {
+ public:
+  explicit MayAutoLock(base::Optional<base::Lock>* lock)
+      : lock_(lock->has_value() ? &lock->value() : nullptr) {
+    if (lock_)
+      lock_->Acquire();
+  }
+
+  ~MayAutoLock() {
+    if (lock_) {
+      lock_->AssertAcquired();
+      lock_->Release();
+    }
+  }
+
+ private:
+  base::Lock* lock_;
+  DISALLOW_COPY_AND_ASSIGN(MayAutoLock);
+};
+
+// Similar to base::AutoUnlock, except that it does nothing if |lock| passed
+// into the constructor is null.
+class MayAutoUnlock {
+ public:
+  explicit MayAutoUnlock(base::Optional<base::Lock>* lock)
+      : lock_(lock->has_value() ? &lock->value() : nullptr) {
+    if (lock_) {
+      lock_->AssertAcquired();
+      lock_->Release();
+    }
+  }
+
+  ~MayAutoUnlock() {
+    if (lock_)
+      lock_->Acquire();
+  }
+
+ private:
+  base::Lock* lock_;
+  DISALLOW_COPY_AND_ASSIGN(MayAutoUnlock);
+};
+
+}  // namespace internal
+}  // namespace mojo
+
+#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_MAY_AUTO_LOCK_H_
diff --git a/mojo/public/cpp/bindings/lib/message.cc b/mojo/public/cpp/bindings/lib/message.cc
index 939e064..e5f3808 100644
--- a/mojo/public/cpp/bindings/lib/message.cc
+++ b/mojo/public/cpp/bindings/lib/message.cc
@@ -11,18 +11,58 @@
 #include <algorithm>
 #include <utility>
 
+#include "base/bind.h"
+#include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/strings/stringprintf.h"
+#include "base/threading/thread_local.h"
+#include "mojo/public/cpp/bindings/associated_group_controller.h"
+#include "mojo/public/cpp/bindings/lib/array_internal.h"
 
 namespace mojo {
 
+namespace {
+
+base::LazyInstance<base::ThreadLocalPointer<internal::MessageDispatchContext>>::
+    DestructorAtExit g_tls_message_dispatch_context = LAZY_INSTANCE_INITIALIZER;
+
+base::LazyInstance<base::ThreadLocalPointer<SyncMessageResponseContext>>::
+    DestructorAtExit g_tls_sync_response_context = LAZY_INSTANCE_INITIALIZER;
+
+void DoNotifyBadMessage(Message message, const std::string& error) {
+  message.NotifyBadMessage(error);
+}
+
+}  // namespace
+
 Message::Message() {
 }
 
+Message::Message(Message&& other)
+    : buffer_(std::move(other.buffer_)),
+      handles_(std::move(other.handles_)),
+      associated_endpoint_handles_(
+          std::move(other.associated_endpoint_handles_)) {}
+
 Message::~Message() {
   CloseHandles();
 }
 
+Message& Message::operator=(Message&& other) {
+  Reset();
+  std::swap(other.buffer_, buffer_);
+  std::swap(other.handles_, handles_);
+  std::swap(other.associated_endpoint_handles_, associated_endpoint_handles_);
+  return *this;
+}
+
+void Message::Reset() {
+  CloseHandles();
+  handles_.clear();
+  associated_endpoint_handles_.clear();
+  buffer_.reset();
+}
+
 void Message::Initialize(size_t capacity, bool zero_initialized) {
   DCHECK(!buffer_);
   buffer_.reset(new internal::MessageBuffer(capacity, zero_initialized));
@@ -36,19 +76,52 @@ void Message::InitializeFromMojoMessage(ScopedMessageHandle message,
   handles_.swap(*handles);
 }
 
-void Message::MoveTo(Message* destination) {
-  DCHECK(this != destination);
+const uint8_t* Message::payload() const {
+  if (version() < 2)
+    return data() + header()->num_bytes;
 
-  // No copy needed.
-  std::swap(destination->buffer_, buffer_);
-  std::swap(destination->handles_, handles_);
+  return static_cast<const uint8_t*>(header_v2()->payload.Get());
+}
 
-  CloseHandles();
-  handles_.clear();
-  buffer_.reset();
+uint32_t Message::payload_num_bytes() const {
+  DCHECK_GE(data_num_bytes(), header()->num_bytes);
+  size_t num_bytes;
+  if (version() < 2) {
+    num_bytes = data_num_bytes() - header()->num_bytes;
+  } else {
+    auto payload = reinterpret_cast<uintptr_t>(header_v2()->payload.Get());
+    if (!payload) {
+      num_bytes = 0;
+    } else {
+      auto payload_end =
+          reinterpret_cast<uintptr_t>(header_v2()->payload_interface_ids.Get());
+      if (!payload_end)
+        payload_end = reinterpret_cast<uintptr_t>(data() + data_num_bytes());
+      DCHECK_GE(payload_end, payload);
+      num_bytes = payload_end - payload;
+    }
+  }
+  DCHECK_LE(num_bytes, std::numeric_limits<uint32_t>::max());
+  return static_cast<uint32_t>(num_bytes);
+}
+
+uint32_t Message::payload_num_interface_ids() const {
+  auto* array_pointer =
+      version() < 2 ? nullptr : header_v2()->payload_interface_ids.Get();
+  return array_pointer ? static_cast<uint32_t>(array_pointer->size()) : 0;
+}
+
+const uint32_t* Message::payload_interface_ids() const {
+  auto* array_pointer =
+      version() < 2 ? nullptr : header_v2()->payload_interface_ids.Get();
+  return array_pointer ? array_pointer->storage() : nullptr;
 }
 
 ScopedMessageHandle Message::TakeMojoMessage() {
+  // If there are associated endpoints transferred,
+  // SerializeAssociatedEndpointHandles() must be called before this method.
+  DCHECK(associated_endpoint_handles_.empty());
+
   if (handles_.empty())  // Fast path for the common case: No handles.
     return buffer_->TakeMessage();
 
@@ -80,6 +153,7 @@ ScopedMessageHandle Message::TakeMojoMessage() {
 }
 
 void Message::NotifyBadMessage(const std::string& error) {
+  DCHECK(buffer_);
   buffer_->NotifyBadMessage(error);
 }
 
@@ -91,6 +165,88 @@ void Message::CloseHandles() {
   }
 }
 
+void Message::SerializeAssociatedEndpointHandles(
+    AssociatedGroupController* group_controller) {
+  if (associated_endpoint_handles_.empty())
+    return;
+
+  DCHECK_GE(version(), 2u);
+  DCHECK(header_v2()->payload_interface_ids.is_null());
+
+  size_t size = associated_endpoint_handles_.size();
+  auto* data = internal::Array_Data<uint32_t>::New(size, buffer());
+  header_v2()->payload_interface_ids.Set(data);
+
+  for (size_t i = 0; i < size; ++i) {
+    ScopedInterfaceEndpointHandle& handle = associated_endpoint_handles_[i];
+
+    DCHECK(handle.pending_association());
+    data->storage()[i] =
+        group_controller->AssociateInterface(std::move(handle));
+  }
+  associated_endpoint_handles_.clear();
+}
+
+bool Message::DeserializeAssociatedEndpointHandles(
+    AssociatedGroupController* group_controller) {
+  associated_endpoint_handles_.clear();
+
+  uint32_t num_ids = payload_num_interface_ids();
+  if (num_ids == 0)
+    return true;
+
+  associated_endpoint_handles_.reserve(num_ids);
+  uint32_t* ids = header_v2()->payload_interface_ids.Get()->storage();
+  bool result = true;
+  for (uint32_t i = 0; i < num_ids; ++i) {
+    auto handle = group_controller->CreateLocalEndpointHandle(ids[i]);
+    if (IsValidInterfaceId(ids[i]) && !handle.is_valid()) {
+      // |ids[i]| itself is valid but handle creation failed. In that case, mark
+      // deserialization as failed but continue to deserialize the rest of
+      // handles.
+      result = false;
+    }
+
+    associated_endpoint_handles_.push_back(std::move(handle));
+    ids[i] = kInvalidInterfaceId;
+  }
+  return result;
+}
+
+PassThroughFilter::PassThroughFilter() {}
+
+PassThroughFilter::~PassThroughFilter() {}
+
+bool PassThroughFilter::Accept(Message* message) { return true; }
+
+SyncMessageResponseContext::SyncMessageResponseContext()
+    : outer_context_(current()) {
+  g_tls_sync_response_context.Get().Set(this);
+}
+
+SyncMessageResponseContext::~SyncMessageResponseContext() {
+  DCHECK_EQ(current(), this);
+  g_tls_sync_response_context.Get().Set(outer_context_);
+}
+
+// static
+SyncMessageResponseContext* SyncMessageResponseContext::current() {
+  return g_tls_sync_response_context.Get().Get();
+}
+
+void SyncMessageResponseContext::ReportBadMessage(const std::string& error) {
+  GetBadMessageCallback().Run(error);
+}
+
+const ReportBadMessageCallback&
+SyncMessageResponseContext::GetBadMessageCallback() {
+  if (bad_message_callback_.is_null()) {
+    bad_message_callback_ =
+        base::Bind(&DoNotifyBadMessage, base::Passed(&response_));
+  }
+  return bad_message_callback_;
+}
+
 MojoResult ReadMessage(MessagePipeHandle handle, Message* message) {
   MojoResult rv;
 
@@ -122,4 +278,55 @@ MojoResult ReadMessage(MessagePipeHandle handle, Message* message) {
   return MOJO_RESULT_OK;
 }
 
+void ReportBadMessage(const std::string& error) {
+  internal::MessageDispatchContext* context =
+      internal::MessageDispatchContext::current();
+  DCHECK(context);
+  context->GetBadMessageCallback().Run(error);
+}
+
+ReportBadMessageCallback GetBadMessageCallback() {
+  internal::MessageDispatchContext* context =
+      internal::MessageDispatchContext::current();
+  DCHECK(context);
+  return context->GetBadMessageCallback();
+}
+
+namespace internal {
+
+MessageHeaderV2::MessageHeaderV2() = default;
+
+MessageDispatchContext::MessageDispatchContext(Message* message)
+    : outer_context_(current()), message_(message) {
+  g_tls_message_dispatch_context.Get().Set(this);
+}
+
+MessageDispatchContext::~MessageDispatchContext() {
+  DCHECK_EQ(current(), this);
+  g_tls_message_dispatch_context.Get().Set(outer_context_);
+}
+
+// static
+MessageDispatchContext* MessageDispatchContext::current() {
+  return g_tls_message_dispatch_context.Get().Get();
+}
+
+const ReportBadMessageCallback&
+MessageDispatchContext::GetBadMessageCallback() {
+  if (bad_message_callback_.is_null()) {
+    bad_message_callback_ =
+        base::Bind(&DoNotifyBadMessage, base::Passed(message_));
+  }
+  return bad_message_callback_;
+}
+
+// static
+void SyncMessageResponseSetup::SetCurrentSyncResponseMessage(Message* message) {
+  SyncMessageResponseContext* context = SyncMessageResponseContext::current();
+  if (context)
+    context->response_ = std::move(*message);
+}
+
+}  // namespace internal
+
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/message_buffer.cc b/mojo/public/cpp/bindings/lib/message_buffer.cc
index af79cfd..cc12ef6 100644
--- a/mojo/public/cpp/bindings/lib/message_buffer.cc
+++ b/mojo/public/cpp/bindings/lib/message_buffer.cc
@@ -13,54 +13,35 @@ namespace internal {
 
 MessageBuffer::MessageBuffer(size_t capacity, bool zero_initialized) {
   DCHECK_LE(capacity, std::numeric_limits<uint32_t>::max());
-  data_num_bytes_ = static_cast<uint32_t>(capacity);
 
   MojoResult rv = AllocMessage(capacity, nullptr, 0,
                                MOJO_ALLOC_MESSAGE_FLAG_NONE, &message_);
   CHECK_EQ(rv, MOJO_RESULT_OK);
 
-  if (capacity == 0) {
-    buffer_ = nullptr;
-  } else {
-    rv = GetMessageBuffer(message_.get(), &buffer_);
+  void* buffer = nullptr;
+  if (capacity != 0) {
+    rv = GetMessageBuffer(message_.get(), &buffer);
     CHECK_EQ(rv, MOJO_RESULT_OK);
 
     if (zero_initialized)
-      memset(buffer_, 0, capacity);
+      memset(buffer, 0, capacity);
   }
+  Initialize(buffer, capacity);
 }
 
 MessageBuffer::MessageBuffer(ScopedMessageHandle message, uint32_t num_bytes) {
   message_ = std::move(message);
-  data_num_bytes_ = num_bytes;
 
-  if (num_bytes == 0) {
-    buffer_ = nullptr;
-  } else {
-    MojoResult rv = GetMessageBuffer(message_.get(), &buffer_);
+  void* buffer = nullptr;
+  if (num_bytes != 0) {
+    MojoResult rv = GetMessageBuffer(message_.get(), &buffer);
     CHECK_EQ(rv, MOJO_RESULT_OK);
   }
+  Initialize(buffer, num_bytes);
 }
 
 MessageBuffer::~MessageBuffer() {}
 
-void* MessageBuffer::Allocate(size_t delta) {
-  delta = internal::Align(delta);
-
-  DCHECK_LE(delta, static_cast<size_t>(data_num_bytes_));
-  DCHECK_GT(bytes_claimed_ + static_cast<uint32_t>(delta), bytes_claimed_);
-
-  uint32_t new_bytes_claimed = bytes_claimed_ + static_cast<uint32_t>(delta);
-  if (new_bytes_claimed > data_num_bytes_) {
-    NOTREACHED();
-    return nullptr;
-  }
-
-  char* start = static_cast<char*>(buffer_) + bytes_claimed_;
-  bytes_claimed_ = new_bytes_claimed;
-  return static_cast<void*>(start);
-}
-
 void MessageBuffer::NotifyBadMessage(const std::string& error) {
   DCHECK(message_.is_valid());
   MojoResult result = mojo::NotifyBadMessage(message_.get(), error);
diff --git a/mojo/public/cpp/bindings/lib/message_buffer.h b/mojo/public/cpp/bindings/lib/message_buffer.h
index 0382131..96d5140 100644
--- a/mojo/public/cpp/bindings/lib/message_buffer.h
+++ b/mojo/public/cpp/bindings/lib/message_buffer.h
@@ -5,7 +5,6 @@
 #ifndef MOJO_PUBLIC_CPP_BINDINGS_MESSAGE_LIB_MESSAGE_BUFFER_H_
 #define MOJO_PUBLIC_CPP_BINDINGS_MESSAGE_LIB_MESSAGE_BUFFER_H_
 
-#include <stddef.h>
 #include <stdint.h>
 
 #include <utility>
@@ -17,7 +16,7 @@
 namespace mojo {
 namespace internal {
 
-// A fixed-size Buffer implementation using a Mojo message object for storage.
+// A fixed-size Buffer using a Mojo message object for storage.
 class MessageBuffer : public Buffer {
  public:
   // Initializes this buffer to carry a fixed byte capacity and no handles.
@@ -26,24 +25,14 @@ class MessageBuffer : public Buffer {
   // Initializes this buffer from an existing Mojo MessageHandle.
   MessageBuffer(ScopedMessageHandle message, uint32_t num_bytes);
 
-  ~MessageBuffer() override;
-
-  void* data() const { return buffer_; }
-  uint32_t data_num_bytes() const { return data_num_bytes_; }
-
-  // Buffer:
-  void* Allocate(size_t delta) override;
+  ~MessageBuffer();
 
   ScopedMessageHandle TakeMessage() { return std::move(message_); }
 
   void NotifyBadMessage(const std::string& error);
 
  private:
-  uint32_t data_num_bytes_ = 0;
   ScopedMessageHandle message_;
-  void* buffer_;
-
-  uint32_t bytes_claimed_ = 0;
 
   DISALLOW_COPY_AND_ASSIGN(MessageBuffer);
 };
diff --git a/mojo/public/cpp/bindings/lib/message_builder.cc b/mojo/public/cpp/bindings/lib/message_builder.cc
index 4ffa180..6806a73 100644
--- a/mojo/public/cpp/bindings/lib/message_builder.cc
+++ b/mojo/public/cpp/bindings/lib/message_builder.cc
@@ -4,11 +4,10 @@
 
 #include "mojo/public/cpp/bindings/lib/message_builder.h"
 
-#include <stddef.h>
-#include <stdint.h>
-
-#include "mojo/public/cpp/bindings/lib/serialization_util.h"
-#include "mojo/public/cpp/bindings/message.h"
+#include "mojo/public/cpp/bindings/lib/array_internal.h"
+#include "mojo/public/cpp/bindings/lib/bindings_internal.h"
+#include "mojo/public/cpp/bindings/lib/buffer.h"
+#include "mojo/public/cpp/bindings/lib/message_internal.h"
 
 namespace mojo {
 namespace internal {
@@ -19,38 +18,52 @@ void Allocate(Buffer* buf, Header** header) {
   (*header)->num_bytes = sizeof(Header);
 }
 
-MessageBuilder::MessageBuilder(uint32_t name, size_t payload_size) {
-  InitializeMessage(sizeof(MessageHeader) + payload_size);
+MessageBuilder::MessageBuilder(uint32_t name,
+                               uint32_t flags,
+                               size_t payload_size,
+                               size_t payload_interface_id_count) {
+  if (payload_interface_id_count > 0) {
+    // Version 2
+    InitializeMessage(
+        sizeof(MessageHeaderV2) + Align(payload_size) +
+        ArrayDataTraits<uint32_t>::GetStorageSize(
+            static_cast<uint32_t>(payload_interface_id_count)));
+
+    MessageHeaderV2* header;
+    Allocate(message_.buffer(), &header);
+    header->version = 2;
+    header->name = name;
+    header->flags = flags;
+    // The payload immediately follows the header.
+    header->payload.Set(header + 1);
+  } else if (flags &
+             (Message::kFlagExpectsResponse | Message::kFlagIsResponse)) {
+    // Version 1
+    InitializeMessage(sizeof(MessageHeaderV1) + payload_size);
+
+    MessageHeaderV1* header;
+    Allocate(message_.buffer(), &header);
+    header->version = 1;
+    header->name = name;
+    header->flags = flags;
+  } else {
+    InitializeMessage(sizeof(MessageHeader) + payload_size);
 
-  MessageHeader* header;
-  Allocate(message_.buffer(), &header);
-  header->version = 0;
-  header->name = name;
+    MessageHeader* header;
+    Allocate(message_.buffer(), &header);
+    header->version = 0;
+    header->name = name;
+    header->flags = flags;
+  }
 }
 
 MessageBuilder::~MessageBuilder() {
 }
 
-MessageBuilder::MessageBuilder() {}
-
 void MessageBuilder::InitializeMessage(size_t size) {
   message_.Initialize(static_cast<uint32_t>(Align(size)),
                       true /* zero_initialized */);
 }
 
-MessageWithRequestIDBuilder::MessageWithRequestIDBuilder(uint32_t name,
-                                                         size_t payload_size,
-                                                         uint32_t flags,
-                                                         uint64_t request_id) {
-  InitializeMessage(sizeof(MessageHeaderWithRequestID) + payload_size);
-
-  MessageHeaderWithRequestID* header;
-  Allocate(message_.buffer(), &header);
-  header->version = 1;
-  header->name = name;
-  header->flags = flags;
-  header->request_id = request_id;
-}
-
 }  // namespace internal
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/message_builder.h b/mojo/public/cpp/bindings/lib/message_builder.h
index a5a050f..8a4d5c4 100644
--- a/mojo/public/cpp/bindings/lib/message_builder.h
+++ b/mojo/public/cpp/bindings/lib/message_builder.h
@@ -8,24 +8,30 @@
 #include <stddef.h>
 #include <stdint.h>
 
-#include "mojo/public/cpp/bindings/lib/message_internal.h"
+#include "base/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/message.h"
 
 namespace mojo {
+
 class Message;
 
 namespace internal {
 
-class MessageBuilder {
+class Buffer;
+
+class MOJO_CPP_BINDINGS_EXPORT MessageBuilder {
  public:
-  MessageBuilder(uint32_t name, size_t payload_size);
+  MessageBuilder(uint32_t name,
+                 uint32_t flags,
+                 size_t payload_size,
+                 size_t payload_interface_id_count);
   ~MessageBuilder();
 
   Buffer* buffer() { return message_.buffer(); }
   Message* message() { return &message_; }
 
- protected:
-  MessageBuilder();
+ private:
   void InitializeMessage(size_t size);
 
   Message message_;
@@ -33,51 +39,6 @@ class MessageBuilder {
   DISALLOW_COPY_AND_ASSIGN(MessageBuilder);
 };
 
-class MessageWithRequestIDBuilder : public MessageBuilder {
- public:
-  MessageWithRequestIDBuilder(uint32_t name,
-                              size_t payload_size,
-                              uint32_t flags,
-                              uint64_t request_id);
-};
-
-class RequestMessageBuilder : public MessageWithRequestIDBuilder {
- public:
-  RequestMessageBuilder(uint32_t name, size_t payload_size)
-      : MessageWithRequestIDBuilder(name,
-                                    payload_size,
-                                    Message::kFlagExpectsResponse,
-                                    0) {}
-
-  RequestMessageBuilder(uint32_t name,
-                        size_t payload_size,
-                        uint32_t extra_flags)
-      : MessageWithRequestIDBuilder(name,
-                                    payload_size,
-                                    Message::kFlagExpectsResponse | extra_flags,
-                                    0) {}
-};
-
-class ResponseMessageBuilder : public MessageWithRequestIDBuilder {
- public:
-  ResponseMessageBuilder(uint32_t name,
-                         size_t payload_size,
-                         uint64_t request_id)
-      : MessageWithRequestIDBuilder(name,
-                                    payload_size,
-                                    Message::kFlagIsResponse,
-                                    request_id) {}
-
-  ResponseMessageBuilder(uint32_t name,
-                         size_t payload_size,
-                         uint64_t request_id,
-                         uint32_t extra_flags)
-      : MessageWithRequestIDBuilder(name,
-                                    payload_size,
-                                    Message::kFlagIsResponse | extra_flags,
-                                    request_id) {}
-};
-
 }  // namespace internal
 }  // namespace mojo
 
diff --git a/mojo/public/cpp/bindings/lib/message_filter.cc b/mojo/public/cpp/bindings/lib/message_filter.cc
deleted file mode 100644
index b09f40d..0000000
--- a/mojo/public/cpp/bindings/lib/message_filter.cc
+++ /dev/null
@@ -1,23 +0,0 @@
-// Copyright 2014 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "mojo/public/cpp/bindings/message_filter.h"
-
-namespace mojo {
-
-MessageFilter::MessageFilter(MessageReceiver* sink) : sink_(sink) {
-}
-
-MessageFilter::~MessageFilter() {
-}
-
-PassThroughFilter::PassThroughFilter(MessageReceiver* sink)
-    : MessageFilter(sink) {
-}
-
-bool PassThroughFilter::Accept(Message* message) {
-  return sink_->Accept(message);
-}
-
-}  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/message_header_validator.cc b/mojo/public/cpp/bindings/lib/message_header_validator.cc
index 10f7774..9f8c627 100644
--- a/mojo/public/cpp/bindings/lib/message_header_validator.cc
+++ b/mojo/public/cpp/bindings/lib/message_header_validator.cc
@@ -4,6 +4,8 @@
 
 #include "mojo/public/cpp/bindings/message_header_validator.h"
 
+#include "mojo/public/cpp/bindings/lib/array_internal.h"
+#include "mojo/public/cpp/bindings/lib/validate_params.h"
 #include "mojo/public/cpp/bindings/lib/validation_context.h"
 #include "mojo/public/cpp/bindings/lib/validation_errors.h"
 #include "mojo/public/cpp/bindings/lib/validation_util.h"
@@ -11,40 +13,40 @@
 namespace mojo {
 namespace {
 
+// TODO(yzshen): Define a mojom struct for message header and use the generated
+// validation and data view code.
 bool IsValidMessageHeader(const internal::MessageHeader* header,
                           internal::ValidationContext* validation_context) {
   // NOTE: Our goal is to preserve support for future extension of the message
   // header. If we encounter fields we do not understand, we must ignore them.
 
   // Extra validation of the struct header:
-  if (header->version == 0) {
-    if (header->num_bytes != sizeof(internal::MessageHeader)) {
-      internal::ReportValidationError(
-          validation_context,
-          internal::VALIDATION_ERROR_UNEXPECTED_STRUCT_HEADER);
-      return false;
+  do {
+    if (header->version == 0) {
+      if (header->num_bytes == sizeof(internal::MessageHeader))
+        break;
+    } else if (header->version == 1) {
+      if (header->num_bytes == sizeof(internal::MessageHeaderV1))
+        break;
+    } else if (header->version == 2) {
+      if (header->num_bytes == sizeof(internal::MessageHeaderV2))
+        break;
+    } else if (header->version > 2) {
+      if (header->num_bytes >= sizeof(internal::MessageHeaderV2))
+        break;
     }
-  } else if (header->version == 1) {
-    if (header->num_bytes != sizeof(internal::MessageHeaderWithRequestID)) {
-      internal::ReportValidationError(
-          validation_context,
-          internal::VALIDATION_ERROR_UNEXPECTED_STRUCT_HEADER);
-      return false;
-    }
-  } else if (header->version > 1) {
-    if (header->num_bytes < sizeof(internal::MessageHeaderWithRequestID)) {
-      internal::ReportValidationError(
-          validation_context,
-          internal::VALIDATION_ERROR_UNEXPECTED_STRUCT_HEADER);
-      return false;
-    }
-  }
+    internal::ReportValidationError(
+        validation_context,
+        internal::VALIDATION_ERROR_UNEXPECTED_STRUCT_HEADER);
+    return false;
+  } while (false);
 
   // Validate flags (allow unknown bits):
 
   // These flags require a RequestID.
-  if (header->version < 1 && ((header->flags & Message::kFlagExpectsResponse) ||
-                              (header->flags & Message::kFlagIsResponse))) {
+  constexpr uint32_t kRequestIdFlags =
+      Message::kFlagExpectsResponse | Message::kFlagIsResponse;
+  if (header->version == 0 && (header->flags & kRequestIdFlags)) {
     internal::ReportValidationError(
         validation_context,
         internal::VALIDATION_ERROR_MESSAGE_HEADER_MISSING_REQUEST_ID);
@@ -52,25 +54,60 @@ bool IsValidMessageHeader(const internal::MessageHeader* header,
   }
 
   // These flags are mutually exclusive.
-  if ((header->flags & Message::kFlagExpectsResponse) &&
-      (header->flags & Message::kFlagIsResponse)) {
+  if ((header->flags & kRequestIdFlags) == kRequestIdFlags) {
     internal::ReportValidationError(
         validation_context,
         internal::VALIDATION_ERROR_MESSAGE_HEADER_INVALID_FLAGS);
     return false;
   }
 
+  if (header->version < 2)
+    return true;
+
+  auto* header_v2 = static_cast<const internal::MessageHeaderV2*>(header);
+  // For the payload pointer:
+  // - Check that the pointer can be safely decoded.
+  // - Claim one byte that the pointer points to. It makes sure not only the
+  //   address is within the message, but also the address precedes the array
+  //   storing interface IDs (which is important for safely calculating the
+  //   payload size).
+  // - Validation of the payload contents will be done separately based on the
+  //   payload type.
+  if (!header_v2->payload.is_null() &&
+      (!internal::ValidatePointer(header_v2->payload, validation_context) ||
+       !validation_context->ClaimMemory(header_v2->payload.Get(), 1))) {
+    return false;
+  }
+
+  const internal::ContainerValidateParams validate_params(0, false, nullptr);
+  if (!internal::ValidateContainer(header_v2->payload_interface_ids,
+                                   validation_context, &validate_params)) {
+    return false;
+  }
+
+  if (!header_v2->payload_interface_ids.is_null()) {
+    size_t num_ids = header_v2->payload_interface_ids.Get()->size();
+    const uint32_t* ids = header_v2->payload_interface_ids.Get()->storage();
+    for (size_t i = 0; i < num_ids; ++i) {
+      if (!IsValidInterfaceId(ids[i]) || IsMasterInterfaceId(ids[i])) {
+        internal::ReportValidationError(
+            validation_context,
+            internal::VALIDATION_ERROR_ILLEGAL_INTERFACE_ID);
+        return false;
+      }
+    }
+  }
+
   return true;
 }
 
 }  // namespace
 
-MessageHeaderValidator::MessageHeaderValidator(MessageReceiver* sink)
-    : MessageHeaderValidator("MessageHeaderValidator", sink) {}
+MessageHeaderValidator::MessageHeaderValidator()
+    : MessageHeaderValidator("MessageHeaderValidator") {}
 
-MessageHeaderValidator::MessageHeaderValidator(const std::string& description,
-                                               MessageReceiver* sink)
-    : MessageFilter(sink), description_(description) {
+MessageHeaderValidator::MessageHeaderValidator(const std::string& description)
+    : description_(description) {
 }
 
 void MessageHeaderValidator::SetDescription(const std::string& description) {
@@ -78,10 +115,10 @@ void MessageHeaderValidator::SetDescription(const std::string& description) {
 }
 
 bool MessageHeaderValidator::Accept(Message* message) {
-  // Pass 0 as number of handles because we don't expect any in the header, even
-  // if |message| contains handles.
+  // Pass 0 as number of handles and associated endpoint handles because we
+  // don't expect any in the header, even if |message| contains handles.
   internal::ValidationContext validation_context(
-      message->data(), message->data_num_bytes(), 0, message, description_);
+      message->data(), message->data_num_bytes(), 0, 0, message, description_);
 
   if (!internal::ValidateStructHeaderAndClaimMemory(message->data(),
                                                     &validation_context))
@@ -90,7 +127,7 @@ bool MessageHeaderValidator::Accept(Message* message) {
   if (!IsValidMessageHeader(message->header(), &validation_context))
     return false;
 
-  return sink_->Accept(message);
+  return true;
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/message_internal.h b/mojo/public/cpp/bindings/lib/message_internal.h
index 63edffd..6693198 100644
--- a/mojo/public/cpp/bindings/lib/message_internal.h
+++ b/mojo/public/cpp/bindings/lib/message_internal.h
@@ -7,11 +7,22 @@
 
 #include <stdint.h>
 
+#include <string>
+
+#include "base/callback.h"
+#include "base/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
 
 namespace mojo {
+
+class Message;
+
 namespace internal {
 
+template <typename T>
+class Array_Data;
+
 #pragma pack(push, 1)
 
 struct MessageHeader : internal::StructHeader {
@@ -27,16 +38,44 @@ struct MessageHeader : internal::StructHeader {
 };
 static_assert(sizeof(MessageHeader) == 24, "Bad sizeof(MessageHeader)");
 
-struct MessageHeaderWithRequestID : MessageHeader {
+struct MessageHeaderV1 : MessageHeader {
   // Only used if either kFlagExpectsResponse or kFlagIsResponse is set in
   // order to match responses with corresponding requests.
   uint64_t request_id;
 };
-static_assert(sizeof(MessageHeaderWithRequestID) == 32,
-              "Bad sizeof(MessageHeaderWithRequestID)");
+static_assert(sizeof(MessageHeaderV1) == 32, "Bad sizeof(MessageHeaderV1)");
+
+struct MessageHeaderV2 : MessageHeaderV1 {
+  MessageHeaderV2();
+  GenericPointer payload;
+  Pointer<Array_Data<uint32_t>> payload_interface_ids;
+};
+static_assert(sizeof(MessageHeaderV2) == 48, "Bad sizeof(MessageHeaderV2)");
 
 #pragma pack(pop)
 
+class MOJO_CPP_BINDINGS_EXPORT MessageDispatchContext {
+ public:
+  explicit MessageDispatchContext(Message* message);
+  ~MessageDispatchContext();
+
+  static MessageDispatchContext* current();
+
+  const base::Callback<void(const std::string&)>& GetBadMessageCallback();
+
+ private:
+  MessageDispatchContext* outer_context_;
+  Message* message_;
+  base::Callback<void(const std::string&)> bad_message_callback_;
+
+  DISALLOW_COPY_AND_ASSIGN(MessageDispatchContext);
+};
+
+class MOJO_CPP_BINDINGS_EXPORT SyncMessageResponseSetup {
+ public:
+  static void SetCurrentSyncResponseMessage(Message* message);
+};
+
 }  // namespace internal
 }  // namespace mojo
 
diff --git a/mojo/public/cpp/bindings/lib/multiplex_router.cc b/mojo/public/cpp/bindings/lib/multiplex_router.cc
index dcfbab1..2da459a 100644
--- a/mojo/public/cpp/bindings/lib/multiplex_router.cc
+++ b/mojo/public/cpp/bindings/lib/multiplex_router.cc
@@ -15,9 +15,9 @@
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/threading/thread_task_runner_handle.h"
-#include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
+#include "mojo/public/cpp/bindings/lib/may_auto_lock.h"
 #include "mojo/public/cpp/bindings/sync_handle_watcher.h"
 
 namespace mojo {
@@ -36,6 +36,7 @@ class MultiplexRouter::InterfaceEndpoint
         id_(id),
         closed_(false),
         peer_closed_(false),
+        handle_created_(false),
         client_(nullptr),
         event_signalled_(false) {}
 
@@ -50,16 +51,31 @@ class MultiplexRouter::InterfaceEndpoint
 
   bool closed() const { return closed_; }
   void set_closed() {
-    router_->lock_.AssertAcquired();
+    router_->AssertLockAcquired();
     closed_ = true;
   }
 
   bool peer_closed() const { return peer_closed_; }
   void set_peer_closed() {
-    router_->lock_.AssertAcquired();
+    router_->AssertLockAcquired();
     peer_closed_ = true;
   }
 
+  bool handle_created() const { return handle_created_; }
+  void set_handle_created() {
+    router_->AssertLockAcquired();
+    handle_created_ = true;
+  }
+
+  const base::Optional<DisconnectReason>& disconnect_reason() const {
+    return disconnect_reason_;
+  }
+  void set_disconnect_reason(
+      const base::Optional<DisconnectReason>& disconnect_reason) {
+    router_->AssertLockAcquired();
+    disconnect_reason_ = disconnect_reason;
+  }
+
   base::SingleThreadTaskRunner* task_runner() const {
     return task_runner_.get();
   }
@@ -68,7 +84,7 @@ class MultiplexRouter::InterfaceEndpoint
 
   void AttachClient(InterfaceEndpointClient* client,
                     scoped_refptr<base::SingleThreadTaskRunner> runner) {
-    router_->lock_.AssertAcquired();
+    router_->AssertLockAcquired();
     DCHECK(!client_);
     DCHECK(!closed_);
     DCHECK(runner->BelongsToCurrentThread());
@@ -80,7 +96,7 @@ class MultiplexRouter::InterfaceEndpoint
   // This method must be called on the same thread as the corresponding
   // AttachClient() call.
   void DetachClient() {
-    router_->lock_.AssertAcquired();
+    router_->AssertLockAcquired();
     DCHECK(client_);
     DCHECK(task_runner_->BelongsToCurrentThread());
     DCHECK(!closed_);
@@ -91,18 +107,36 @@ class MultiplexRouter::InterfaceEndpoint
   }
 
   void SignalSyncMessageEvent() {
-    router_->lock_.AssertAcquired();
+    router_->AssertLockAcquired();
     if (event_signalled_)
       return;
 
-    EnsureEventMessagePipeExists();
     event_signalled_ = true;
+    if (!sync_message_event_sender_.is_valid())
+      return;
+
     MojoResult result =
         WriteMessageRaw(sync_message_event_sender_.get(), nullptr, 0, nullptr,
                         0, MOJO_WRITE_MESSAGE_FLAG_NONE);
     DCHECK_EQ(MOJO_RESULT_OK, result);
   }
 
+  void ResetSyncMessageSignal() {
+    router_->AssertLockAcquired();
+
+    if (!event_signalled_)
+      return;
+
+    event_signalled_ = false;
+    if (!sync_message_event_receiver_.is_valid())
+      return;
+
+    MojoResult result =
+        ReadMessageRaw(sync_message_event_receiver_.get(), nullptr, nullptr,
+                       nullptr, nullptr, MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
+    DCHECK_EQ(MOJO_RESULT_OK, result);
+  }
+
   // ---------------------------------------------------------------------------
   // The following public methods (i.e., InterfaceEndpointController
   // implementation) are called by the client on the same thread as the
@@ -132,7 +166,7 @@ class MultiplexRouter::InterfaceEndpoint
   friend class base::RefCounted<InterfaceEndpoint>;
 
   ~InterfaceEndpoint() override {
-    router_->lock_.AssertAcquired();
+    router_->AssertLockAcquired();
 
     DCHECK(!client_);
     DCHECK(closed_);
@@ -142,26 +176,23 @@ class MultiplexRouter::InterfaceEndpoint
 
   void OnHandleReady(MojoResult result) {
     DCHECK(task_runner_->BelongsToCurrentThread());
-    scoped_refptr<InterfaceEndpoint> self_protector(this);
     scoped_refptr<MultiplexRouter> router_protector(router_);
 
     // Because we never close |sync_message_event_{sender,receiver}_| before
     // destruction or set a deadline, |result| should always be MOJO_RESULT_OK.
     DCHECK_EQ(MOJO_RESULT_OK, result);
-    bool reset_sync_watcher = false;
-    {
-      base::AutoLock locker(router_->lock_);
 
-      bool more_to_process = router_->ProcessFirstSyncMessageForEndpoint(id_);
+    MayAutoLock locker(&router_->lock_);
+    scoped_refptr<InterfaceEndpoint> self_protector(this);
 
-      if (!more_to_process)
-        ResetSyncMessageSignal();
+    bool more_to_process = router_->ProcessFirstSyncMessageForEndpoint(id_);
 
-      // Currently there are no queued sync messages and the peer has closed so
-      // there won't be incoming sync messages in the future.
-      reset_sync_watcher = !more_to_process && peer_closed_;
-    }
-    if (reset_sync_watcher) {
+    if (!more_to_process)
+      ResetSyncMessageSignal();
+
+    // Currently there are no queued sync messages and the peer has closed so
+    // there won't be incoming sync messages in the future.
+    if (!more_to_process && peer_closed_) {
       // If a SyncWatch() call (or multiple ones) of this interface endpoint is
       // on the call stack, resetting the sync watcher will allow it to exit
       // when the call stack unwinds to that frame.
@@ -175,12 +206,21 @@ class MultiplexRouter::InterfaceEndpoint
       return;
 
     {
-      base::AutoLock locker(router_->lock_);
-      EnsureEventMessagePipeExists();
-
-      auto iter = router_->sync_message_tasks_.find(id_);
-      if (iter != router_->sync_message_tasks_.end() && !iter->second.empty())
-        SignalSyncMessageEvent();
+      MayAutoLock locker(&router_->lock_);
+
+      if (!sync_message_event_sender_.is_valid()) {
+        MojoResult result =
+            CreateMessagePipe(nullptr, &sync_message_event_sender_,
+                              &sync_message_event_receiver_);
+        DCHECK_EQ(MOJO_RESULT_OK, result);
+
+        if (event_signalled_) {
+          // Reset the flag so that SignalSyncMessageEvent() will actually
+          // signal using the newly-created message pipe.
+          event_signalled_ = false;
+          SignalSyncMessageEvent();
+        }
+      }
     }
 
     sync_watcher_.reset(new SyncHandleWatcher(
@@ -188,31 +228,6 @@ class MultiplexRouter::InterfaceEndpoint
         base::Bind(&InterfaceEndpoint::OnHandleReady, base::Unretained(this))));
   }
 
-  void EnsureEventMessagePipeExists() {
-    router_->lock_.AssertAcquired();
-
-    if (sync_message_event_receiver_.is_valid())
-      return;
-
-    MojoResult result = CreateMessagePipe(nullptr, &sync_message_event_sender_,
-                                          &sync_message_event_receiver_);
-    DCHECK_EQ(MOJO_RESULT_OK, result);
-  }
-
-  void ResetSyncMessageSignal() {
-    router_->lock_.AssertAcquired();
-
-    if (!event_signalled_)
-      return;
-
-    DCHECK(sync_message_event_receiver_.is_valid());
-    MojoResult result = ReadMessageRaw(sync_message_event_receiver_.get(),
-                                       nullptr, nullptr, nullptr, nullptr,
-                                       MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
-    DCHECK_EQ(MOJO_RESULT_OK, result);
-    event_signalled_ = false;
-  }
-
   // ---------------------------------------------------------------------------
   // The following members are safe to access from any threads.
 
@@ -227,6 +242,12 @@ class MultiplexRouter::InterfaceEndpoint
   // Whether the peer endpoint has been closed.
   bool peer_closed_;
 
+  // Whether there is already a ScopedInterfaceEndpointHandle created for this
+  // endpoint.
+  bool handle_created_;
+
+  base::Optional<DisconnectReason> disconnect_reason_;
+
   // The task runner on which |client_|'s methods can be called.
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   // Not owned. It is null if no client is attached to this endpoint.
@@ -250,13 +271,53 @@ class MultiplexRouter::InterfaceEndpoint
   DISALLOW_COPY_AND_ASSIGN(InterfaceEndpoint);
 };
 
+// MessageWrapper objects are always destroyed under the router's lock. On
+// destruction, if the message it wrappers contains
+// ScopedInterfaceEndpointHandles (which cannot be destructed under the
+// router's lock), the wrapper unlocks to clean them up.
+class MultiplexRouter::MessageWrapper {
+ public:
+  MessageWrapper() = default;
+
+  MessageWrapper(MultiplexRouter* router, Message message)
+      : router_(router), value_(std::move(message)) {}
+
+  MessageWrapper(MessageWrapper&& other)
+      : router_(other.router_), value_(std::move(other.value_)) {}
+
+  ~MessageWrapper() {
+    if (value_.associated_endpoint_handles()->empty())
+      return;
+
+    router_->AssertLockAcquired();
+    {
+      MayAutoUnlock unlocker(&router_->lock_);
+      value_.mutable_associated_endpoint_handles()->clear();
+    }
+  }
+
+  MessageWrapper& operator=(MessageWrapper&& other) {
+    router_ = other.router_;
+    value_ = std::move(other.value_);
+    return *this;
+  }
+
+  Message& value() { return value_; }
+
+ private:
+  MultiplexRouter* router_ = nullptr;
+  Message value_;
+
+  DISALLOW_COPY_AND_ASSIGN(MessageWrapper);
+};
+
 struct MultiplexRouter::Task {
  public:
   // Doesn't take ownership of |message| but takes its contents.
-  static std::unique_ptr<Task> CreateMessageTask(Message* message) {
+  static std::unique_ptr<Task> CreateMessageTask(
+      MessageWrapper message_wrapper) {
     Task* task = new Task(MESSAGE);
-    task->message.reset(new Message);
-    message->MoveTo(task->message.get());
+    task->message_wrapper = std::move(message_wrapper);
     return base::WrapUnique(task);
   }
   static std::unique_ptr<Task> CreateNotifyErrorTask(
@@ -271,7 +332,7 @@ struct MultiplexRouter::Task {
   bool IsMessageTask() const { return type == MESSAGE; }
   bool IsNotifyErrorTask() const { return type == NOTIFY_ERROR; }
 
-  std::unique_ptr<Message> message;
+  MessageWrapper message_wrapper;
   scoped_refptr<InterfaceEndpoint> endpoint_to_notify;
 
   enum Type { MESSAGE, NOTIFY_ERROR };
@@ -279,36 +340,56 @@ struct MultiplexRouter::Task {
 
  private:
   explicit Task(Type in_type) : type(in_type) {}
+
+  DISALLOW_COPY_AND_ASSIGN(Task);
 };
 
 MultiplexRouter::MultiplexRouter(
-    bool set_interface_id_namesapce_bit,
     ScopedMessagePipeHandle message_pipe,
+    Config config,
+    bool set_interface_id_namesapce_bit,
     scoped_refptr<base::SingleThreadTaskRunner> runner)
-    : AssociatedGroupController(base::ThreadTaskRunnerHandle::Get()),
-      set_interface_id_namespace_bit_(set_interface_id_namesapce_bit),
-      header_validator_(this),
+    : set_interface_id_namespace_bit_(set_interface_id_namesapce_bit),
+      task_runner_(runner),
+      header_validator_(nullptr),
+      filters_(this),
       connector_(std::move(message_pipe),
-                 Connector::MULTI_THREADED_SEND,
+                 config == MULTI_INTERFACE ? Connector::MULTI_THREADED_SEND
+                                           : Connector::SINGLE_THREADED_SEND,
                  std::move(runner)),
       control_message_handler_(this),
       control_message_proxy_(&connector_),
       next_interface_id_value_(1),
       posted_to_process_tasks_(false),
       encountered_error_(false),
+      paused_(false),
       testing_mode_(false) {
-  // Always participate in sync handle watching, because even if it doesn't
-  // expect sync requests during sync handle watching, it may still need to
-  // dispatch messages to associated endpoints on a different thread.
-  connector_.AllowWokenUpBySyncWatchOnSameThread();
-  connector_.set_incoming_receiver(&header_validator_);
+  DCHECK(task_runner_->BelongsToCurrentThread());
+
+  if (config == MULTI_INTERFACE)
+    lock_.emplace();
+
+  if (config == SINGLE_INTERFACE_WITH_SYNC_METHODS ||
+      config == MULTI_INTERFACE) {
+    // Always participate in sync handle watching in multi-interface mode,
+    // because even if it doesn't expect sync requests during sync handle
+    // watching, it may still need to dispatch messages to associated endpoints
+    // on a different thread.
+    connector_.AllowWokenUpBySyncWatchOnSameThread();
+  }
+  connector_.set_incoming_receiver(&filters_);
   connector_.set_connection_error_handler(
       base::Bind(&MultiplexRouter::OnPipeConnectionError,
                  base::Unretained(this)));
+
+  std::unique_ptr<MessageHeaderValidator> header_validator =
+      base::MakeUnique<MessageHeaderValidator>();
+  header_validator_ = header_validator.get();
+  filters_.Append(std::move(header_validator));
 }
 
 MultiplexRouter::~MultiplexRouter() {
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
 
   sync_message_tasks_.clear();
   tasks_.clear();
@@ -319,40 +400,66 @@ MultiplexRouter::~MultiplexRouter() {
     // because it may remove the corresponding value from the map.
     ++iter;
 
-    DCHECK(endpoint->closed());
-    UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
+    if (!endpoint->closed()) {
+      // This happens when a NotifyPeerEndpointClosed message been received, but
+      // the interface ID hasn't been used to create local endpoint handle.
+      DCHECK(!endpoint->client());
+      DCHECK(endpoint->peer_closed());
+      UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
+    } else {
+      UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
+    }
   }
 
   DCHECK(endpoints_.empty());
 }
 
-void MultiplexRouter::SetMasterInterfaceName(const std::string& name) {
+void MultiplexRouter::SetMasterInterfaceName(const char* name) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  header_validator_.SetDescription(name + " [master] MessageHeaderValidator");
+  header_validator_->SetDescription(
+      std::string(name) + " [master] MessageHeaderValidator");
   control_message_handler_.SetDescription(
-      name + " [master] PipeControlMessageHandler");
+      std::string(name) + " [master] PipeControlMessageHandler");
+  connector_.SetWatcherHeapProfilerTag(name);
 }
 
-void MultiplexRouter::CreateEndpointHandlePair(
-    ScopedInterfaceEndpointHandle* local_endpoint,
-    ScopedInterfaceEndpointHandle* remote_endpoint) {
-  base::AutoLock locker(lock_);
+InterfaceId MultiplexRouter::AssociateInterface(
+    ScopedInterfaceEndpointHandle handle_to_send) {
+  if (!handle_to_send.pending_association())
+    return kInvalidInterfaceId;
+
   uint32_t id = 0;
-  do {
-    if (next_interface_id_value_ >= kInterfaceIdNamespaceMask)
-      next_interface_id_value_ = 1;
-    id = next_interface_id_value_++;
-    if (set_interface_id_namespace_bit_)
-      id |= kInterfaceIdNamespaceMask;
-  } while (ContainsKey(endpoints_, id));
-
-  InterfaceEndpoint* endpoint = new InterfaceEndpoint(this, id);
-  endpoints_[id] = endpoint;
-  if (encountered_error_)
-    UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
+  {
+    MayAutoLock locker(&lock_);
+    do {
+      if (next_interface_id_value_ >= kInterfaceIdNamespaceMask)
+        next_interface_id_value_ = 1;
+      id = next_interface_id_value_++;
+      if (set_interface_id_namespace_bit_)
+        id |= kInterfaceIdNamespaceMask;
+    } while (base::ContainsKey(endpoints_, id));
+
+    InterfaceEndpoint* endpoint = new InterfaceEndpoint(this, id);
+    endpoints_[id] = endpoint;
+    if (encountered_error_)
+      UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
+    endpoint->set_handle_created();
+  }
+
+  if (!NotifyAssociation(&handle_to_send, id)) {
+    // The peer handle of |handle_to_send|, which is supposed to join this
+    // associated group, has been closed.
+    {
+      MayAutoLock locker(&lock_);
+      InterfaceEndpoint* endpoint = FindEndpoint(id);
+      if (endpoint)
+        UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
+    }
 
-  *local_endpoint = CreateScopedInterfaceEndpointHandle(id, true);
-  *remote_endpoint = CreateScopedInterfaceEndpointHandle(id, false);
+    control_message_proxy_.NotifyPeerEndpointClosed(
+        id, handle_to_send.disconnect_reason());
+  }
+  return id;
 }
 
 ScopedInterfaceEndpointHandle MultiplexRouter::CreateLocalEndpointHandle(
@@ -360,10 +467,12 @@ ScopedInterfaceEndpointHandle MultiplexRouter::CreateLocalEndpointHandle(
   if (!IsValidInterfaceId(id))
     return ScopedInterfaceEndpointHandle();
 
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
   bool inserted = false;
   InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
   if (inserted) {
+    DCHECK(!endpoint->handle_created());
+
     if (encountered_error_)
       UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
   } else {
@@ -371,34 +480,32 @@ ScopedInterfaceEndpointHandle MultiplexRouter::CreateLocalEndpointHandle(
     // notification that the peer endpoint has closed.
     CHECK(!endpoint->closed());
     CHECK(endpoint->peer_closed());
+
+    if (endpoint->handle_created())
+      return ScopedInterfaceEndpointHandle();
   }
-  return CreateScopedInterfaceEndpointHandle(id, true);
+
+  endpoint->set_handle_created();
+  return CreateScopedInterfaceEndpointHandle(id);
 }
 
-void MultiplexRouter::CloseEndpointHandle(InterfaceId id, bool is_local) {
+void MultiplexRouter::CloseEndpointHandle(
+    InterfaceId id,
+    const base::Optional<DisconnectReason>& reason) {
   if (!IsValidInterfaceId(id))
     return;
 
-  base::AutoLock locker(lock_);
-
-  if (!is_local) {
-    DCHECK(ContainsKey(endpoints_, id));
-    DCHECK(!IsMasterInterfaceId(id));
-
-    // We will receive a NotifyPeerEndpointClosed message from the other side.
-    control_message_proxy_.NotifyEndpointClosedBeforeSent(id);
-
-    return;
-  }
-
-  DCHECK(ContainsKey(endpoints_, id));
+  MayAutoLock locker(&lock_);
+  DCHECK(base::ContainsKey(endpoints_, id));
   InterfaceEndpoint* endpoint = endpoints_[id].get();
   DCHECK(!endpoint->client());
   DCHECK(!endpoint->closed());
   UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
 
-  if (!IsMasterInterfaceId(id))
-    control_message_proxy_.NotifyPeerEndpointClosed(id);
+  if (!IsMasterInterfaceId(id) || reason) {
+    MayAutoUnlock unlocker(&lock_);
+    control_message_proxy_.NotifyPeerEndpointClosed(id, reason);
+  }
 
   ProcessTasks(NO_DIRECT_CLIENT_CALLS, nullptr);
 }
@@ -412,8 +519,8 @@ InterfaceEndpointController* MultiplexRouter::AttachEndpointClient(
   DCHECK(IsValidInterfaceId(id));
   DCHECK(client);
 
-  base::AutoLock locker(lock_);
-  DCHECK(ContainsKey(endpoints_, id));
+  MayAutoLock locker(&lock_);
+  DCHECK(base::ContainsKey(endpoints_, id));
 
   InterfaceEndpoint* endpoint = endpoints_[id].get();
   endpoint->AttachClient(client, std::move(runner));
@@ -431,8 +538,8 @@ void MultiplexRouter::DetachEndpointClient(
 
   DCHECK(IsValidInterfaceId(id));
 
-  base::AutoLock locker(lock_);
-  DCHECK(ContainsKey(endpoints_, id));
+  MayAutoLock locker(&lock_);
+  DCHECK(base::ContainsKey(endpoints_, id));
 
   InterfaceEndpoint* endpoint = endpoints_[id].get();
   endpoint->DetachClient();
@@ -456,21 +563,51 @@ void MultiplexRouter::CloseMessagePipe() {
   OnPipeConnectionError();
 }
 
+void MultiplexRouter::PauseIncomingMethodCallProcessing() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  connector_.PauseIncomingMethodCallProcessing();
+
+  MayAutoLock locker(&lock_);
+  paused_ = true;
+
+  for (auto iter = endpoints_.begin(); iter != endpoints_.end(); ++iter)
+    iter->second->ResetSyncMessageSignal();
+}
+
+void MultiplexRouter::ResumeIncomingMethodCallProcessing() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  connector_.ResumeIncomingMethodCallProcessing();
+
+  MayAutoLock locker(&lock_);
+  paused_ = false;
+
+  for (auto iter = endpoints_.begin(); iter != endpoints_.end(); ++iter) {
+    auto sync_iter = sync_message_tasks_.find(iter->first);
+    if (iter->second->peer_closed() ||
+        (sync_iter != sync_message_tasks_.end() &&
+         !sync_iter->second.empty())) {
+      iter->second->SignalSyncMessageEvent();
+    }
+  }
+
+  ProcessTasks(NO_DIRECT_CLIENT_CALLS, nullptr);
+}
+
 bool MultiplexRouter::HasAssociatedEndpoints() const {
   DCHECK(thread_checker_.CalledOnValidThread());
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
 
   if (endpoints_.size() > 1)
     return true;
   if (endpoints_.size() == 0)
     return false;
 
-  return !ContainsKey(endpoints_, kMasterInterfaceId);
+  return !base::ContainsKey(endpoints_, kMasterInterfaceId);
 }
 
 void MultiplexRouter::EnableTestingMode() {
   DCHECK(thread_checker_.CalledOnValidThread());
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
 
   testing_mode_ = true;
   connector_.set_enforce_errors_from_incoming_receiver(false);
@@ -479,8 +616,13 @@ void MultiplexRouter::EnableTestingMode() {
 bool MultiplexRouter::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
+  if (!message->DeserializeAssociatedEndpointHandles(this))
+    return false;
+
   scoped_refptr<MultiplexRouter> protector(this);
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
+
+  DCHECK(!paused_);
 
   ClientCallBehavior client_call_behavior =
       connector_.during_sync_handle_watcher_callback()
@@ -494,15 +636,16 @@ bool MultiplexRouter::Accept(Message* message) {
   if (!processed) {
     // Either the task queue is not empty or we cannot process the message
     // directly. In both cases, there is no need to call ProcessTasks().
-    tasks_.push_back(Task::CreateMessageTask(message));
+    tasks_.push_back(
+        Task::CreateMessageTask(MessageWrapper(this, std::move(*message))));
     Task* task = tasks_.back().get();
 
-    if (task->message->has_flag(Message::kFlagIsSync)) {
-      InterfaceId id = task->message->interface_id();
+    if (task->message_wrapper.value().has_flag(Message::kFlagIsSync)) {
+      InterfaceId id = task->message_wrapper.value().interface_id();
       sync_message_tasks_[id].push_back(task);
-      auto iter = endpoints_.find(id);
-      if (iter != endpoints_.end())
-        iter->second->SignalSyncMessageEvent();
+      InterfaceEndpoint* endpoint = FindEndpoint(id);
+      if (endpoint)
+        endpoint->SignalSyncMessageEvent();
     }
   } else if (!tasks_.empty()) {
     // Processing the message may result in new tasks (for error notification)
@@ -516,14 +659,17 @@ bool MultiplexRouter::Accept(Message* message) {
   return true;
 }
 
-bool MultiplexRouter::OnPeerAssociatedEndpointClosed(InterfaceId id) {
-  lock_.AssertAcquired();
-
-  if (IsMasterInterfaceId(id))
-    return false;
+bool MultiplexRouter::OnPeerAssociatedEndpointClosed(
+    InterfaceId id,
+    const base::Optional<DisconnectReason>& reason) {
+  DCHECK(!IsMasterInterfaceId(id) || reason);
 
+  MayAutoLock locker(&lock_);
   InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, nullptr);
 
+  if (reason)
+    endpoint->set_disconnect_reason(reason);
+
   // It is possible that this endpoint has been set as peer closed. That is
   // because when the message pipe is closed, all the endpoints are updated with
   // PEER_ENDPOINT_CLOSED. We continue to process remaining tasks in the queue,
@@ -541,26 +687,11 @@ bool MultiplexRouter::OnPeerAssociatedEndpointClosed(InterfaceId id) {
   return true;
 }
 
-bool MultiplexRouter::OnAssociatedEndpointClosedBeforeSent(InterfaceId id) {
-  lock_.AssertAcquired();
-
-  if (IsMasterInterfaceId(id))
-    return false;
-
-  InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, nullptr);
-  DCHECK(!endpoint->closed());
-  UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
-
-  control_message_proxy_.NotifyPeerEndpointClosed(id);
-
-  return true;
-}
-
 void MultiplexRouter::OnPipeConnectionError() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   scoped_refptr<MultiplexRouter> protector(this);
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
 
   encountered_error_ = true;
 
@@ -585,20 +716,21 @@ void MultiplexRouter::OnPipeConnectionError() {
 void MultiplexRouter::ProcessTasks(
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
-  lock_.AssertAcquired();
+  AssertLockAcquired();
 
   if (posted_to_process_tasks_)
     return;
 
-  while (!tasks_.empty()) {
+  while (!tasks_.empty() && !paused_) {
     std::unique_ptr<Task> task(std::move(tasks_.front()));
     tasks_.pop_front();
 
     InterfaceId id = kInvalidInterfaceId;
-    bool sync_message = task->IsMessageTask() && task->message &&
-                        task->message->has_flag(Message::kFlagIsSync);
+    bool sync_message =
+        task->IsMessageTask() && !task->message_wrapper.value().IsNull() &&
+        task->message_wrapper.value().has_flag(Message::kFlagIsSync);
     if (sync_message) {
-      id = task->message->interface_id();
+      id = task->message_wrapper.value().interface_id();
       auto& sync_message_queue = sync_message_tasks_[id];
       DCHECK_EQ(task.get(), sync_message_queue.front());
       sync_message_queue.pop_front();
@@ -608,8 +740,8 @@ void MultiplexRouter::ProcessTasks(
         task->IsNotifyErrorTask()
             ? ProcessNotifyErrorTask(task.get(), client_call_behavior,
                                      current_task_runner)
-            : ProcessIncomingMessage(task->message.get(), client_call_behavior,
-                                     current_task_runner);
+            : ProcessIncomingMessage(&task->message_wrapper.value(),
+                                     client_call_behavior, current_task_runner);
 
     if (!processed) {
       if (sync_message) {
@@ -629,21 +761,25 @@ void MultiplexRouter::ProcessTasks(
 }
 
 bool MultiplexRouter::ProcessFirstSyncMessageForEndpoint(InterfaceId id) {
-  lock_.AssertAcquired();
+  AssertLockAcquired();
 
   auto iter = sync_message_tasks_.find(id);
   if (iter == sync_message_tasks_.end())
     return false;
 
+  if (paused_)
+    return true;
+
   MultiplexRouter::Task* task = iter->second.front();
   iter->second.pop_front();
 
   DCHECK(task->IsMessageTask());
-  std::unique_ptr<Message> message(std::move(task->message));
+  MessageWrapper message_wrapper = std::move(task->message_wrapper);
 
-  // Note: after this call, |task| and  |iter| may be invalidated.
+  // Note: after this call, |task| and |iter| may be invalidated.
   bool processed = ProcessIncomingMessage(
-      message.get(), ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES, nullptr);
+      &message_wrapper.value(), ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES,
+      nullptr);
   DCHECK(processed);
 
   iter = sync_message_tasks_.find(id);
@@ -663,7 +799,9 @@ bool MultiplexRouter::ProcessNotifyErrorTask(
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
   DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
-  lock_.AssertAcquired();
+  DCHECK(!paused_);
+
+  AssertLockAcquired();
   InterfaceEndpoint* endpoint = task->endpoint_to_notify.get();
   if (!endpoint->client())
     return true;
@@ -677,14 +815,17 @@ bool MultiplexRouter::ProcessNotifyErrorTask(
   DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
 
   InterfaceEndpointClient* client = endpoint->client();
+  base::Optional<DisconnectReason> disconnect_reason(
+      endpoint->disconnect_reason());
+
   {
     // We must unlock before calling into |client| because it may call this
     // object within NotifyError(). Holding the lock will lead to deadlock.
     //
     // It is safe to call into |client| without the lock. Because |client| is
     // always accessed on the same thread, including DetachEndpointClient().
-    base::AutoUnlock unlocker(lock_);
-    client->NotifyError();
+    MayAutoUnlock unlocker(&lock_);
+    client->NotifyError(disconnect_reason);
   }
   return true;
 }
@@ -694,47 +835,35 @@ bool MultiplexRouter::ProcessIncomingMessage(
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
   DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
-  lock_.AssertAcquired();
+  DCHECK(!paused_);
+  DCHECK(message);
+  AssertLockAcquired();
 
-  if (!message) {
+  if (message->IsNull()) {
     // This is a sync message and has been processed during sync handle
     // watching.
     return true;
   }
 
   if (PipeControlMessageHandler::IsPipeControlMessage(message)) {
-    if (!control_message_handler_.Accept(message))
+    bool result = false;
+
+    {
+      MayAutoUnlock unlocker(&lock_);
+      result = control_message_handler_.Accept(message);
+    }
+
+    if (!result)
       RaiseErrorInNonTestingMode();
+
     return true;
   }
 
   InterfaceId id = message->interface_id();
   DCHECK(IsValidInterfaceId(id));
 
-  bool inserted = false;
-  InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
-  if (inserted) {
-    // Currently, it is legitimate to receive messages for an endpoint
-    // that is not registered. For example, the endpoint is transferred in
-    // a message that is discarded. Once we add support to specify all
-    // enclosing endpoints in message header, we should be able to remove
-    // this.
-    UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
-
-    // It is also possible that this newly-inserted endpoint is the master
-    // endpoint. When the master InterfacePtr/Binding goes away, the message
-    // pipe is closed and we explicitly trigger a pipe connection error. The
-    // error updates all the endpoints, including the master endpoint, with
-    // PEER_ENDPOINT_CLOSED and removes the master endpoint from the
-    // registration. We continue to process remaining tasks in the queue, as
-    // long as there are refs keeping the router alive. If there are remaining
-    // messages for the master endpoint, we will get here.
-    if (!IsMasterInterfaceId(id))
-      control_message_proxy_.NotifyPeerEndpointClosed(id);
-    return true;
-  }
-
-  if (endpoint->closed())
+  InterfaceEndpoint* endpoint = FindEndpoint(id);
+  if (!endpoint || endpoint->closed())
     return true;
 
   if (!endpoint->client()) {
@@ -768,7 +897,7 @@ bool MultiplexRouter::ProcessIncomingMessage(
     //
     // It is safe to call into |client| without the lock. Because |client| is
     // always accessed on the same thread, including DetachEndpointClient().
-    base::AutoUnlock unlocker(lock_);
+    MayAutoUnlock unlocker(&lock_);
     result = client->HandleIncomingMessage(message);
   }
   if (!result)
@@ -779,7 +908,7 @@ bool MultiplexRouter::ProcessIncomingMessage(
 
 void MultiplexRouter::MaybePostToProcessTasks(
     base::SingleThreadTaskRunner* task_runner) {
-  lock_.AssertAcquired();
+  AssertLockAcquired();
   if (posted_to_process_tasks_)
     return;
 
@@ -792,7 +921,7 @@ void MultiplexRouter::MaybePostToProcessTasks(
 void MultiplexRouter::LockAndCallProcessTasks() {
   // There is no need to hold a ref to this class in this case because this is
   // always called using base::Bind(), which holds a ref.
-  base::AutoLock locker(lock_);
+  MayAutoLock locker(&lock_);
   posted_to_process_tasks_ = false;
   scoped_refptr<base::SingleThreadTaskRunner> runner(
       std::move(posted_to_task_runner_));
@@ -802,23 +931,20 @@ void MultiplexRouter::LockAndCallProcessTasks() {
 void MultiplexRouter::UpdateEndpointStateMayRemove(
     InterfaceEndpoint* endpoint,
     EndpointStateUpdateType type) {
-  switch (type) {
-    case ENDPOINT_CLOSED:
-      endpoint->set_closed();
-      break;
-    case PEER_ENDPOINT_CLOSED:
-      endpoint->set_peer_closed();
-      // If the interface endpoint is performing a sync watch, this makes sure
-      // it is notified and eventually exits the sync watch.
-      endpoint->SignalSyncMessageEvent();
-      break;
+  if (type == ENDPOINT_CLOSED) {
+    endpoint->set_closed();
+  } else {
+    endpoint->set_peer_closed();
+    // If the interface endpoint is performing a sync watch, this makes sure
+    // it is notified and eventually exits the sync watch.
+    endpoint->SignalSyncMessageEvent();
   }
   if (endpoint->closed() && endpoint->peer_closed())
     endpoints_.erase(endpoint->id());
 }
 
 void MultiplexRouter::RaiseErrorInNonTestingMode() {
-  lock_.AssertAcquired();
+  AssertLockAcquired();
   if (!testing_mode_)
     RaiseError();
 }
@@ -826,24 +952,35 @@ void MultiplexRouter::RaiseErrorInNonTestingMode() {
 MultiplexRouter::InterfaceEndpoint* MultiplexRouter::FindOrInsertEndpoint(
     InterfaceId id,
     bool* inserted) {
-  lock_.AssertAcquired();
+  AssertLockAcquired();
   // Either |inserted| is nullptr or it points to a boolean initialized as
   // false.
   DCHECK(!inserted || !*inserted);
 
-  auto iter = endpoints_.find(id);
-  InterfaceEndpoint* endpoint;
-  if (iter == endpoints_.end()) {
+  InterfaceEndpoint* endpoint = FindEndpoint(id);
+  if (!endpoint) {
     endpoint = new InterfaceEndpoint(this, id);
     endpoints_[id] = endpoint;
     if (inserted)
       *inserted = true;
-  } else {
-    endpoint = iter->second.get();
   }
 
   return endpoint;
 }
 
+MultiplexRouter::InterfaceEndpoint* MultiplexRouter::FindEndpoint(
+    InterfaceId id) {
+  AssertLockAcquired();
+  auto iter = endpoints_.find(id);
+  return iter != endpoints_.end() ? iter->second.get() : nullptr;
+}
+
+void MultiplexRouter::AssertLockAcquired() {
+#if DCHECK_IS_ON()
+  if (lock_)
+    lock_->AssertAcquired();
+#endif
+}
+
 }  // namespace internal
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/multiplex_router.h b/mojo/public/cpp/bindings/lib/multiplex_router.h
index dc66e8e..cac138b 100644
--- a/mojo/public/cpp/bindings/lib/multiplex_router.h
+++ b/mojo/public/cpp/bindings/lib/multiplex_router.h
@@ -12,15 +12,19 @@
 #include <memory>
 #include <string>
 
+#include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/weak_ptr.h"
+#include "base/optional.h"
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/lock.h"
 #include "base/threading/thread_checker.h"
 #include "mojo/public/cpp/bindings/associated_group_controller.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/connector.h"
+#include "mojo/public/cpp/bindings/filter_chain.h"
 #include "mojo/public/cpp/bindings/interface_id.h"
 #include "mojo/public/cpp/bindings/message_header_validator.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_handler.h"
@@ -34,8 +38,6 @@ class SingleThreadTaskRunner;
 
 namespace mojo {
 
-class AssociatedGroup;
-
 namespace internal {
 
 // MultiplexRouter supports routing messages for multiple interfaces over a
@@ -47,31 +49,51 @@ namespace internal {
 // Some public methods are only allowed to be called on the creating thread;
 // while the others are safe to call from any threads. Please see the method
 // comments for more details.
-class MultiplexRouter
-    : public MessageReceiver,
+//
+// NOTE: CloseMessagePipe() or PassMessagePipe() MUST be called on |runner|'s
+// thread before this object is destroyed.
+class MOJO_CPP_BINDINGS_EXPORT MultiplexRouter
+    : NON_EXPORTED_BASE(public MessageReceiver),
       public AssociatedGroupController,
-      public PipeControlMessageHandlerDelegate {
+      NON_EXPORTED_BASE(public PipeControlMessageHandlerDelegate) {
  public:
+  enum Config {
+    // There is only the master interface running on this router. Please note
+    // that because of interface versioning, the other side of the message pipe
+    // may use a newer master interface definition which passes associated
+    // interfaces. In that case, this router may still receive pipe control
+    // messages or messages targetting associated interfaces.
+    SINGLE_INTERFACE,
+    // Similar to the mode above, there is only the master interface running on
+    // this router. Besides, the master interface has sync methods.
+    SINGLE_INTERFACE_WITH_SYNC_METHODS,
+    // There may be associated interfaces running on this router.
+    MULTI_INTERFACE
+  };
+
   // If |set_interface_id_namespace_bit| is true, the interface IDs generated by
   // this router will have the highest bit set.
-  MultiplexRouter(bool set_interface_id_namespace_bit,
-                  ScopedMessagePipeHandle message_pipe,
+  MultiplexRouter(ScopedMessagePipeHandle message_pipe,
+                  Config config,
+                  bool set_interface_id_namespace_bit,
                   scoped_refptr<base::SingleThreadTaskRunner> runner);
 
   // Sets the master interface name for this router. Only used when reporting
   // message header or control message validation errors.
-  void SetMasterInterfaceName(const std::string& name);
+  // |name| must be a string literal.
+  void SetMasterInterfaceName(const char* name);
 
   // ---------------------------------------------------------------------------
   // The following public methods are safe to call from any threads.
 
   // AssociatedGroupController implementation:
-  void CreateEndpointHandlePair(
-      ScopedInterfaceEndpointHandle* local_endpoint,
-      ScopedInterfaceEndpointHandle* remote_endpoint) override;
+  InterfaceId AssociateInterface(
+      ScopedInterfaceEndpointHandle handle_to_send) override;
   ScopedInterfaceEndpointHandle CreateLocalEndpointHandle(
       InterfaceId id) override;
-  void CloseEndpointHandle(InterfaceId id, bool is_local) override;
+  void CloseEndpointHandle(
+      InterfaceId id,
+      const base::Optional<DisconnectReason>& reason) override;
   InterfaceEndpointController* AttachEndpointClient(
       const ScopedInterfaceEndpointHandle& handle,
       InterfaceEndpointClient* endpoint_client,
@@ -102,14 +124,8 @@ class MultiplexRouter
   }
 
   // See Binding for details of pause/resume.
-  void PauseIncomingMethodCallProcessing() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.PauseIncomingMethodCallProcessing();
-  }
-  void ResumeIncomingMethodCallProcessing() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.ResumeIncomingMethodCallProcessing();
-  }
+  void PauseIncomingMethodCallProcessing();
+  void ResumeIncomingMethodCallProcessing();
 
   // Whether there are any associated interfaces running currently.
   bool HasAssociatedEndpoints() const;
@@ -131,8 +147,13 @@ class MultiplexRouter
     return connector_.handle();
   }
 
+  bool SimulateReceivingMessageForTesting(Message* message) {
+    return filters_.Accept(message);
+  }
+
  private:
   class InterfaceEndpoint;
+  class MessageWrapper;
   struct Task;
 
   ~MultiplexRouter() override;
@@ -141,8 +162,9 @@ class MultiplexRouter
   bool Accept(Message* message) override;
 
   // PipeControlMessageHandlerDelegate implementation:
-  bool OnPeerAssociatedEndpointClosed(InterfaceId id) override;
-  bool OnAssociatedEndpointClosedBeforeSent(InterfaceId id) override;
+  bool OnPeerAssociatedEndpointClosed(
+      InterfaceId id,
+      const base::Optional<DisconnectReason>& reason) override;
 
   void OnPipeConnectionError();
 
@@ -202,19 +224,30 @@ class MultiplexRouter
   void RaiseErrorInNonTestingMode();
 
   InterfaceEndpoint* FindOrInsertEndpoint(InterfaceId id, bool* inserted);
+  InterfaceEndpoint* FindEndpoint(InterfaceId id);
+
+  void AssertLockAcquired();
 
   // Whether to set the namespace bit when generating interface IDs. Please see
   // comments of kInterfaceIdNamespaceMask.
   const bool set_interface_id_namespace_bit_;
 
-  MessageHeaderValidator header_validator_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+
+  // Owned by |filters_| below.
+  MessageHeaderValidator* header_validator_;
+
+  FilterChain filters_;
   Connector connector_;
 
   base::ThreadChecker thread_checker_;
 
   // Protects the following members.
-  mutable base::Lock lock_;
+  // Not set in Config::SINGLE_INTERFACE* mode.
+  mutable base::Optional<base::Lock> lock_;
   PipeControlMessageHandler control_message_handler_;
+
+  // NOTE: It is unsafe to call into this object while holding |lock_|.
   PipeControlMessageProxy control_message_proxy_;
 
   std::map<InterfaceId, scoped_refptr<InterfaceEndpoint>> endpoints_;
@@ -229,6 +262,8 @@ class MultiplexRouter
 
   bool encountered_error_;
 
+  bool paused_;
+
   bool testing_mode_;
 
   DISALLOW_COPY_AND_ASSIGN(MultiplexRouter);
diff --git a/mojo/public/cpp/bindings/lib/native_struct.cc b/mojo/public/cpp/bindings/lib/native_struct.cc
index 837b75a..7b1a1a6 100644
--- a/mojo/public/cpp/bindings/lib/native_struct.cc
+++ b/mojo/public/cpp/bindings/lib/native_struct.cc
@@ -4,27 +4,31 @@
 
 #include "mojo/public/cpp/bindings/native_struct.h"
 
+#include "mojo/public/cpp/bindings/lib/hash_util.h"
+
 namespace mojo {
 
 // static
 NativeStructPtr NativeStruct::New() {
-  NativeStructPtr rv;
-  internal::StructHelper<NativeStruct>::Initialize(&rv);
-  return rv;
+  return NativeStructPtr(base::in_place);
 }
 
-NativeStruct::NativeStruct() : data(nullptr) {}
+NativeStruct::NativeStruct() {}
 
 NativeStruct::~NativeStruct() {}
 
 NativeStructPtr NativeStruct::Clone() const {
   NativeStructPtr rv(New());
-  rv->data = data.Clone();
+  rv->data = data;
   return rv;
 }
 
 bool NativeStruct::Equals(const NativeStruct& other) const {
-  return data.Equals(other.data);
+  return data == other.data;
+}
+
+size_t NativeStruct::Hash(size_t seed) const {
+  return internal::Hash(seed, data);
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/native_struct_data.h b/mojo/public/cpp/bindings/lib/native_struct_data.h
index 5c58774..1c7cd81 100644
--- a/mojo/public/cpp/bindings/lib/native_struct_data.h
+++ b/mojo/public/cpp/bindings/lib/native_struct_data.h
@@ -7,16 +7,16 @@
 
 #include <vector>
 
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/array_internal.h"
 #include "mojo/public/cpp/system/handle.h"
 
 namespace mojo {
 namespace internal {
 
-class Buffer;
 class ValidationContext;
 
-class NativeStruct_Data {
+class MOJO_CPP_BINDINGS_EXPORT NativeStruct_Data {
  public:
   static bool Validate(const void* data, ValidationContext* validation_context);
 
diff --git a/mojo/public/cpp/bindings/lib/native_struct_serialization.cc b/mojo/public/cpp/bindings/lib/native_struct_serialization.cc
index ac06059..fa0dbf3 100644
--- a/mojo/public/cpp/bindings/lib/native_struct_serialization.cc
+++ b/mojo/public/cpp/bindings/lib/native_struct_serialization.cc
@@ -15,7 +15,8 @@ size_t UnmappedNativeStructSerializerImpl::PrepareToSerialize(
     SerializationContext* context) {
   if (!input)
     return 0;
-  return internal::PrepareToSerialize<Array<uint8_t>>(input->data, context);
+  return internal::PrepareToSerialize<ArrayDataView<uint8_t>>(input->data,
+                                                              context);
 }
 
 // static
@@ -31,8 +32,8 @@ void UnmappedNativeStructSerializerImpl::Serialize(
 
   Array_Data<uint8_t>* data = nullptr;
   const ContainerValidateParams params(0, false, nullptr);
-  internal::Serialize<Array<uint8_t>>(input->data, buffer, &data, &params,
-                                      context);
+  internal::Serialize<ArrayDataView<uint8_t>>(input->data, buffer, &data,
+                                              &params, context);
   *output = reinterpret_cast<NativeStruct_Data*>(data);
 }
 
@@ -44,7 +45,8 @@ bool UnmappedNativeStructSerializerImpl::Deserialize(
   Array_Data<uint8_t>* data = reinterpret_cast<Array_Data<uint8_t>*>(input);
 
   NativeStructPtr result(NativeStruct::New());
-  if (!internal::Deserialize<Array<uint8_t>>(data, &result->data, context)) {
+  if (!internal::Deserialize<ArrayDataView<uint8_t>>(data, &result->data,
+                                                     context)) {
     output = nullptr;
     return false;
   }
diff --git a/mojo/public/cpp/bindings/lib/native_struct_serialization.h b/mojo/public/cpp/bindings/lib/native_struct_serialization.h
index e64b862..457435b 100644
--- a/mojo/public/cpp/bindings/lib/native_struct_serialization.h
+++ b/mojo/public/cpp/bindings/lib/native_struct_serialization.h
@@ -13,12 +13,14 @@
 #include "base/logging.h"
 #include "base/pickle.h"
 #include "ipc/ipc_param_traits.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/array_internal.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
 #include "mojo/public/cpp/bindings/lib/native_struct_data.h"
 #include "mojo/public/cpp/bindings/lib/serialization_forward.h"
 #include "mojo/public/cpp/bindings/lib/serialization_util.h"
 #include "mojo/public/cpp/bindings/native_struct.h"
+#include "mojo/public/cpp/bindings/native_struct_data_view.h"
 
 namespace mojo {
 namespace internal {
@@ -102,7 +104,7 @@ struct NativeStructSerializerImpl {
   }
 };
 
-struct UnmappedNativeStructSerializerImpl {
+struct MOJO_CPP_BINDINGS_EXPORT UnmappedNativeStructSerializerImpl {
   static size_t PrepareToSerialize(const NativeStructPtr& input,
                                    SerializationContext* context);
   static void Serialize(const NativeStructPtr& input,
@@ -123,7 +125,7 @@ struct NativeStructSerializerImpl<const NativeStructPtr>
     : public UnmappedNativeStructSerializerImpl {};
 
 template <typename MaybeConstUserType>
-struct Serializer<NativeStructPtr, MaybeConstUserType>
+struct Serializer<NativeStructDataView, MaybeConstUserType>
     : public NativeStructSerializerImpl<MaybeConstUserType> {};
 
 }  // namespace internal
diff --git a/mojo/public/cpp/bindings/lib/no_interface.cc b/mojo/public/cpp/bindings/lib/no_interface.cc
deleted file mode 100644
index 9e0945c..0000000
--- a/mojo/public/cpp/bindings/lib/no_interface.cc
+++ /dev/null
@@ -1,20 +0,0 @@
-// Copyright 2014 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "mojo/public/cpp/bindings/no_interface.h"
-
-namespace mojo {
-
-const char* NoInterface::Name_ = "mojo::NoInterface";
-
-bool NoInterfaceStub::Accept(Message* message) {
-  return false;
-}
-
-bool NoInterfaceStub::AcceptWithResponder(Message* message,
-                                          MessageReceiver* responder) {
-  return false;
-}
-
-}  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/pipe_control_message_handler.cc b/mojo/public/cpp/bindings/lib/pipe_control_message_handler.cc
index 7ee9f8a..d451c05 100644
--- a/mojo/public/cpp/bindings/lib/pipe_control_message_handler.cc
+++ b/mojo/public/cpp/bindings/lib/pipe_control_message_handler.cc
@@ -5,8 +5,10 @@
 #include "mojo/public/cpp/bindings/pipe_control_message_handler.h"
 
 #include "base/logging.h"
+#include "mojo/public/cpp/bindings/interface_id.h"
 #include "mojo/public/cpp/bindings/lib/message_builder.h"
 #include "mojo/public/cpp/bindings/lib/serialization.h"
+#include "mojo/public/cpp/bindings/lib/serialization_context.h"
 #include "mojo/public/cpp/bindings/lib/validation_context.h"
 #include "mojo/public/cpp/bindings/lib/validation_util.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_handler_delegate.h"
@@ -41,8 +43,9 @@ bool PipeControlMessageHandler::Accept(Message* message) {
 }
 
 bool PipeControlMessageHandler::Validate(Message* message) {
-  internal::ValidationContext validation_context(
-      message->data(), message->data_num_bytes(), 0, message, description_);
+  internal::ValidationContext validation_context(message->payload(),
+                                                 message->payload_num_bytes(),
+                                                 0, 0, message, description_);
 
   if (message->name() == pipe_control::kRunOrClosePipeMessageId) {
     if (!internal::ValidateMessageIsRequestWithoutResponse(
@@ -58,22 +61,25 @@ bool PipeControlMessageHandler::Validate(Message* message) {
 }
 
 bool PipeControlMessageHandler::RunOrClosePipe(Message* message) {
+  internal::SerializationContext context;
   pipe_control::internal::RunOrClosePipeMessageParams_Data* params =
       reinterpret_cast<
           pipe_control::internal::RunOrClosePipeMessageParams_Data*>(
           message->mutable_payload());
   pipe_control::RunOrClosePipeMessageParamsPtr params_ptr;
-  internal::Deserialize<pipe_control::RunOrClosePipeMessageParamsPtr>(
-      params, &params_ptr, &context_);
+  internal::Deserialize<pipe_control::RunOrClosePipeMessageParamsDataView>(
+      params, &params_ptr, &context);
 
   if (params_ptr->input->is_peer_associated_endpoint_closed_event()) {
-    return delegate_->OnPeerAssociatedEndpointClosed(
-        params_ptr->input->get_peer_associated_endpoint_closed_event()->id);
-  }
-  if (params_ptr->input->is_associated_endpoint_closed_before_sent_event()) {
-    return delegate_->OnAssociatedEndpointClosedBeforeSent(
-        params_ptr->input->get_associated_endpoint_closed_before_sent_event()
-            ->id);
+    const auto& event =
+        params_ptr->input->get_peer_associated_endpoint_closed_event();
+
+    base::Optional<DisconnectReason> reason;
+    if (event->disconnect_reason) {
+      reason.emplace(event->disconnect_reason->custom_reason,
+                     event->disconnect_reason->description);
+    }
+    return delegate_->OnPeerAssociatedEndpointClosed(event->id, reason);
   }
 
   DVLOG(1) << "Unsupported command in a RunOrClosePipe message pipe control "
diff --git a/mojo/public/cpp/bindings/lib/pipe_control_message_proxy.cc b/mojo/public/cpp/bindings/lib/pipe_control_message_proxy.cc
index 55ee64b..701108e 100644
--- a/mojo/public/cpp/bindings/lib/pipe_control_message_proxy.cc
+++ b/mojo/public/cpp/bindings/lib/pipe_control_message_proxy.cc
@@ -11,33 +11,28 @@
 #include "base/logging.h"
 #include "mojo/public/cpp/bindings/lib/message_builder.h"
 #include "mojo/public/cpp/bindings/lib/serialization.h"
-#include "mojo/public/cpp/bindings/message.h"
 #include "mojo/public/interfaces/bindings/pipe_control_messages.mojom.h"
 
 namespace mojo {
 namespace {
 
-void SendRunOrClosePipeMessage(MessageReceiver* receiver,
-                               pipe_control::RunOrClosePipeInputPtr input,
-                               internal::SerializationContext* context) {
-  pipe_control::RunOrClosePipeMessageParamsPtr params_ptr(
-      pipe_control::RunOrClosePipeMessageParams::New());
-  params_ptr->input = std::move(input);
+Message ConstructRunOrClosePipeMessage(
+    pipe_control::RunOrClosePipeInputPtr input_ptr) {
+  internal::SerializationContext context;
 
-  size_t size =
-      internal::PrepareToSerialize<
-          pipe_control::RunOrClosePipeMessageParamsPtr>(params_ptr, context);
-  internal::MessageBuilder builder(pipe_control::kRunOrClosePipeMessageId,
-                                   size);
+  auto params_ptr = pipe_control::RunOrClosePipeMessageParams::New();
+  params_ptr->input = std::move(input_ptr);
+
+  size_t size = internal::PrepareToSerialize<
+      pipe_control::RunOrClosePipeMessageParamsDataView>(params_ptr, &context);
+  internal::MessageBuilder builder(pipe_control::kRunOrClosePipeMessageId, 0,
+                                   size, 0);
 
   pipe_control::internal::RunOrClosePipeMessageParams_Data* params = nullptr;
-  internal::Serialize<pipe_control::RunOrClosePipeMessageParamsPtr>(
-      params_ptr, builder.buffer(), &params, context);
+  internal::Serialize<pipe_control::RunOrClosePipeMessageParamsDataView>(
+      params_ptr, builder.buffer(), &params, &context);
   builder.message()->set_interface_id(kInvalidInterfaceId);
-  bool ok = receiver->Accept(builder.message());
-  // This return value may be ignored as !ok implies the underlying message pipe
-  // has encountered an error, which will be visible through other means.
-  ALLOW_UNUSED_LOCAL(ok);
+  return std::move(*builder.message());
 }
 
 }  // namespace
@@ -45,30 +40,30 @@ void SendRunOrClosePipeMessage(MessageReceiver* receiver,
 PipeControlMessageProxy::PipeControlMessageProxy(MessageReceiver* receiver)
     : receiver_(receiver) {}
 
-void PipeControlMessageProxy::NotifyPeerEndpointClosed(InterfaceId id) {
-  DCHECK(!IsMasterInterfaceId(id));
-  pipe_control::PeerAssociatedEndpointClosedEventPtr event(
-      pipe_control::PeerAssociatedEndpointClosedEvent::New());
-  event->id = id;
-
-  pipe_control::RunOrClosePipeInputPtr input(
-      pipe_control::RunOrClosePipeInput::New());
-  input->set_peer_associated_endpoint_closed_event(std::move(event));
-
-  SendRunOrClosePipeMessage(receiver_, std::move(input), &context_);
+void PipeControlMessageProxy::NotifyPeerEndpointClosed(
+    InterfaceId id,
+    const base::Optional<DisconnectReason>& reason) {
+  Message message(ConstructPeerEndpointClosedMessage(id, reason));
+  bool ok = receiver_->Accept(&message);
+  ALLOW_UNUSED_LOCAL(ok);
 }
 
-void PipeControlMessageProxy::NotifyEndpointClosedBeforeSent(InterfaceId id) {
-  DCHECK(!IsMasterInterfaceId(id));
-  pipe_control::AssociatedEndpointClosedBeforeSentEventPtr event(
-      pipe_control::AssociatedEndpointClosedBeforeSentEvent::New());
+// static
+Message PipeControlMessageProxy::ConstructPeerEndpointClosedMessage(
+    InterfaceId id,
+    const base::Optional<DisconnectReason>& reason) {
+  auto event = pipe_control::PeerAssociatedEndpointClosedEvent::New();
   event->id = id;
+  if (reason) {
+    event->disconnect_reason = pipe_control::DisconnectReason::New();
+    event->disconnect_reason->custom_reason = reason->custom_reason;
+    event->disconnect_reason->description = reason->description;
+  }
 
-  pipe_control::RunOrClosePipeInputPtr input(
-      pipe_control::RunOrClosePipeInput::New());
-  input->set_associated_endpoint_closed_before_sent_event(std::move(event));
+  auto input = pipe_control::RunOrClosePipeInput::New();
+  input->set_peer_associated_endpoint_closed_event(std::move(event));
 
-  SendRunOrClosePipeMessage(receiver_, std::move(input), &context_);
+  return ConstructRunOrClosePipeMessage(std::move(input));
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/router.cc b/mojo/public/cpp/bindings/lib/router.cc
deleted file mode 100644
index 8c1b77d..0000000
--- a/mojo/public/cpp/bindings/lib/router.cc
+++ /dev/null
@@ -1,323 +0,0 @@
-// Copyright 2014 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "mojo/public/cpp/bindings/lib/router.h"
-
-#include <stdint.h>
-
-#include <utility>
-
-#include "base/bind.h"
-#include "base/location.h"
-#include "base/logging.h"
-#include "base/memory/ptr_util.h"
-#include "base/stl_util.h"
-#include "mojo/public/cpp/bindings/sync_call_restrictions.h"
-
-namespace mojo {
-namespace internal {
-
-// ----------------------------------------------------------------------------
-
-namespace {
-
-void DCheckIfInvalid(const base::WeakPtr<Router>& router,
-                   const std::string& message) {
-  bool is_valid = router && !router->encountered_error() && router->is_valid();
-  DCHECK(!is_valid) << message;
-}
-
-class ResponderThunk : public MessageReceiverWithStatus {
- public:
-  explicit ResponderThunk(const base::WeakPtr<Router>& router,
-                          scoped_refptr<base::SingleThreadTaskRunner> runner)
-      : router_(router),
-        accept_was_invoked_(false),
-        task_runner_(std::move(runner)) {}
-  ~ResponderThunk() override {
-    if (!accept_was_invoked_) {
-      // The Mojo application handled a message that was expecting a response
-      // but did not send a response.
-      // We raise an error to signal the calling application that an error
-      // condition occurred. Without this the calling application would have no
-      // way of knowing it should stop waiting for a response.
-      if (task_runner_->RunsTasksOnCurrentThread()) {
-        // Please note that even if this code is run from a different task
-        // runner on the same thread as |task_runner_|, it is okay to directly
-        // call Router::RaiseError(), because it will raise error from the
-        // correct task runner asynchronously.
-        if (router_)
-          router_->RaiseError();
-      } else {
-        task_runner_->PostTask(FROM_HERE,
-                               base::Bind(&Router::RaiseError, router_));
-      }
-    }
-  }
-
-  // MessageReceiver implementation:
-  bool Accept(Message* message) override {
-    DCHECK(task_runner_->RunsTasksOnCurrentThread());
-    accept_was_invoked_ = true;
-    DCHECK(message->has_flag(Message::kFlagIsResponse));
-
-    bool result = false;
-
-    if (router_)
-      result = router_->Accept(message);
-
-    return result;
-  }
-
-  // MessageReceiverWithStatus implementation:
-  bool IsValid() override {
-    DCHECK(task_runner_->RunsTasksOnCurrentThread());
-    return router_ && !router_->encountered_error() && router_->is_valid();
-  }
-
-  void DCheckInvalid(const std::string& message) override {
-    if (task_runner_->RunsTasksOnCurrentThread()) {
-      DCheckIfInvalid(router_, message);
-    } else {
-      task_runner_->PostTask(FROM_HERE,
-                             base::Bind(&DCheckIfInvalid, router_, message));
-    }
-  }
-
- private:
-  base::WeakPtr<Router> router_;
-  bool accept_was_invoked_;
-  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
-};
-
-}  // namespace
-
-// ----------------------------------------------------------------------------
-
-Router::SyncResponseInfo::SyncResponseInfo(bool* in_response_received)
-    : response_received(in_response_received) {}
-
-Router::SyncResponseInfo::~SyncResponseInfo() {}
-
-// ----------------------------------------------------------------------------
-
-Router::HandleIncomingMessageThunk::HandleIncomingMessageThunk(Router* router)
-    : router_(router) {
-}
-
-Router::HandleIncomingMessageThunk::~HandleIncomingMessageThunk() {
-}
-
-bool Router::HandleIncomingMessageThunk::Accept(Message* message) {
-  return router_->HandleIncomingMessage(message);
-}
-
-// ----------------------------------------------------------------------------
-
-Router::Router(ScopedMessagePipeHandle message_pipe,
-               FilterChain filters,
-               bool expects_sync_requests,
-               scoped_refptr<base::SingleThreadTaskRunner> runner)
-    : thunk_(this),
-      filters_(std::move(filters)),
-      connector_(std::move(message_pipe),
-                 Connector::SINGLE_THREADED_SEND,
-                 std::move(runner)),
-      incoming_receiver_(nullptr),
-      next_request_id_(0),
-      testing_mode_(false),
-      pending_task_for_messages_(false),
-      encountered_error_(false),
-      weak_factory_(this) {
-  filters_.SetSink(&thunk_);
-  if (expects_sync_requests)
-    connector_.AllowWokenUpBySyncWatchOnSameThread();
-  connector_.set_incoming_receiver(filters_.GetHead());
-  connector_.set_connection_error_handler(
-      base::Bind(&Router::OnConnectionError, base::Unretained(this)));
-}
-
-Router::~Router() {}
-
-bool Router::Accept(Message* message) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  DCHECK(!message->has_flag(Message::kFlagExpectsResponse));
-  return connector_.Accept(message);
-}
-
-bool Router::AcceptWithResponder(Message* message, MessageReceiver* responder) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  DCHECK(message->has_flag(Message::kFlagExpectsResponse));
-
-  // Reserve 0 in case we want it to convey special meaning in the future.
-  uint64_t request_id = next_request_id_++;
-  if (request_id == 0)
-    request_id = next_request_id_++;
-
-  bool is_sync = message->has_flag(Message::kFlagIsSync);
-  message->set_request_id(request_id);
-  if (!connector_.Accept(message))
-    return false;
-
-  if (!is_sync) {
-    // We assume ownership of |responder|.
-    async_responders_[request_id] = base::WrapUnique(responder);
-    return true;
-  }
-
-  SyncCallRestrictions::AssertSyncCallAllowed();
-
-  bool response_received = false;
-  std::unique_ptr<MessageReceiver> sync_responder(responder);
-  sync_responses_.insert(std::make_pair(
-      request_id, base::WrapUnique(new SyncResponseInfo(&response_received))));
-
-  base::WeakPtr<Router> weak_self = weak_factory_.GetWeakPtr();
-  connector_.SyncWatch(&response_received);
-  // Make sure that this instance hasn't been destroyed.
-  if (weak_self) {
-    DCHECK(ContainsKey(sync_responses_, request_id));
-    auto iter = sync_responses_.find(request_id);
-    DCHECK_EQ(&response_received, iter->second->response_received);
-    if (response_received) {
-      std::unique_ptr<Message> response = std::move(iter->second->response);
-      ignore_result(sync_responder->Accept(response.get()));
-    }
-    sync_responses_.erase(iter);
-  }
-
-  // Return true means that we take ownership of |responder|.
-  return true;
-}
-
-void Router::EnableTestingMode() {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  testing_mode_ = true;
-  connector_.set_enforce_errors_from_incoming_receiver(false);
-}
-
-bool Router::HandleIncomingMessage(Message* message) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-
-  const bool during_sync_call =
-      connector_.during_sync_handle_watcher_callback();
-  if (!message->has_flag(Message::kFlagIsSync) &&
-      (during_sync_call || !pending_messages_.empty())) {
-    std::unique_ptr<Message> pending_message(new Message);
-    message->MoveTo(pending_message.get());
-    pending_messages_.push(std::move(pending_message));
-
-    if (!pending_task_for_messages_) {
-      pending_task_for_messages_ = true;
-      connector_.task_runner()->PostTask(
-          FROM_HERE, base::Bind(&Router::HandleQueuedMessages,
-                                weak_factory_.GetWeakPtr()));
-    }
-
-    return true;
-  }
-
-  return HandleMessageInternal(message);
-}
-
-void Router::HandleQueuedMessages() {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  DCHECK(pending_task_for_messages_);
-
-  base::WeakPtr<Router> weak_self = weak_factory_.GetWeakPtr();
-  while (!pending_messages_.empty()) {
-    std::unique_ptr<Message> message(std::move(pending_messages_.front()));
-    pending_messages_.pop();
-
-    bool result = HandleMessageInternal(message.get());
-    if (!weak_self)
-      return;
-
-    if (!result && !testing_mode_) {
-      connector_.RaiseError();
-      break;
-    }
-  }
-
-  pending_task_for_messages_ = false;
-
-  // We may have already seen a connection error from the connector, but
-  // haven't notified the user because we want to process all the queued
-  // messages first. We should do it now.
-  if (connector_.encountered_error() && !encountered_error_)
-    OnConnectionError();
-}
-
-bool Router::HandleMessageInternal(Message* message) {
-  if (message->has_flag(Message::kFlagExpectsResponse)) {
-    if (!incoming_receiver_)
-      return false;
-
-    MessageReceiverWithStatus* responder = new ResponderThunk(
-        weak_factory_.GetWeakPtr(), connector_.task_runner());
-    bool ok = incoming_receiver_->AcceptWithResponder(message, responder);
-    if (!ok)
-      delete responder;
-    return ok;
-
-  } else if (message->has_flag(Message::kFlagIsResponse)) {
-    uint64_t request_id = message->request_id();
-
-    if (message->has_flag(Message::kFlagIsSync)) {
-      auto it = sync_responses_.find(request_id);
-      if (it == sync_responses_.end()) {
-        DCHECK(testing_mode_);
-        return false;
-      }
-      it->second->response.reset(new Message());
-      message->MoveTo(it->second->response.get());
-      *it->second->response_received = true;
-      return true;
-    }
-
-    auto it = async_responders_.find(request_id);
-    if (it == async_responders_.end()) {
-      DCHECK(testing_mode_);
-      return false;
-    }
-    std::unique_ptr<MessageReceiver> responder = std::move(it->second);
-    async_responders_.erase(it);
-    return responder->Accept(message);
-  } else {
-    if (!incoming_receiver_)
-      return false;
-
-    return incoming_receiver_->Accept(message);
-  }
-}
-
-void Router::OnConnectionError() {
-  if (encountered_error_)
-    return;
-
-  if (!pending_messages_.empty()) {
-    // After all the pending messages are processed, we will check whether an
-    // error has been encountered and run the user's connection error handler
-    // if necessary.
-    DCHECK(pending_task_for_messages_);
-    return;
-  }
-
-  if (connector_.during_sync_handle_watcher_callback()) {
-    // We don't want the error handler to reenter an ongoing sync call.
-    connector_.task_runner()->PostTask(
-        FROM_HERE,
-        base::Bind(&Router::OnConnectionError, weak_factory_.GetWeakPtr()));
-    return;
-  }
-
-  encountered_error_ = true;
-  if (!error_handler_.is_null())
-    error_handler_.Run();
-}
-
-// ----------------------------------------------------------------------------
-
-}  // namespace internal
-}  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/router.h b/mojo/public/cpp/bindings/lib/router.h
deleted file mode 100644
index 6dbe08d..0000000
--- a/mojo/public/cpp/bindings/lib/router.h
+++ /dev/null
@@ -1,177 +0,0 @@
-// Copyright 2014 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_ROUTER_H_
-#define MOJO_PUBLIC_CPP_BINDINGS_LIB_ROUTER_H_
-
-#include <stdint.h>
-
-#include <map>
-#include <memory>
-#include <queue>
-
-#include "base/callback.h"
-#include "base/macros.h"
-#include "base/memory/ref_counted.h"
-#include "base/memory/weak_ptr.h"
-#include "base/single_thread_task_runner.h"
-#include "base/threading/thread_checker.h"
-#include "mojo/public/cpp/bindings/connector.h"
-#include "mojo/public/cpp/bindings/lib/filter_chain.h"
-
-namespace mojo {
-namespace internal {
-
-// TODO(yzshen): Consider removing this class and use MultiplexRouter in all
-// cases. crbug.com/594244
-class Router : public MessageReceiverWithResponder {
- public:
-  Router(ScopedMessagePipeHandle message_pipe,
-         FilterChain filters,
-         bool expects_sync_requests,
-         scoped_refptr<base::SingleThreadTaskRunner> runner);
-  ~Router() override;
-
-  // Sets the receiver to handle messages read from the message pipe that do
-  // not have the Message::kFlagIsResponse flag set.
-  void set_incoming_receiver(MessageReceiverWithResponderStatus* receiver) {
-    incoming_receiver_ = receiver;
-  }
-
-  // Sets the error handler to receive notifications when an error is
-  // encountered while reading from the pipe or waiting to read from the pipe.
-  void set_connection_error_handler(const base::Closure& error_handler) {
-    error_handler_ = error_handler;
-  }
-
-  // Returns true if an error was encountered while reading from the pipe or
-  // waiting to read from the pipe.
-  bool encountered_error() const {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    return encountered_error_;
-  }
-
-  // Is the router bound to a MessagePipe handle?
-  bool is_valid() const {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    return connector_.is_valid();
-  }
-
-  // Please note that this method shouldn't be called unless it results from an
-  // explicit request of the user of bindings (e.g., the user sets an
-  // InterfacePtr to null or closes a Binding).
-  void CloseMessagePipe() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.CloseMessagePipe();
-  }
-
-  ScopedMessagePipeHandle PassMessagePipe() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    return connector_.PassMessagePipe();
-  }
-
-  void RaiseError() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.RaiseError();
-  }
-
-  // MessageReceiver implementation:
-  bool Accept(Message* message) override;
-  bool AcceptWithResponder(Message* message,
-                           MessageReceiver* responder) override;
-
-  // Blocks the current thread until the first incoming method call, i.e.,
-  // either a call to a client method or a callback method, or |deadline|.
-  bool WaitForIncomingMessage(MojoDeadline deadline) {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    return connector_.WaitForIncomingMessage(deadline);
-  }
-
-  // See Binding for details of pause/resume.
-  void PauseIncomingMethodCallProcessing() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.PauseIncomingMethodCallProcessing();
-  }
-  void ResumeIncomingMethodCallProcessing() {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.ResumeIncomingMethodCallProcessing();
-  }
-
-  // Sets this object to testing mode.
-  // In testing mode:
-  // - the object is more tolerant of unrecognized response messages;
-  // - the connector continues working after seeing errors from its incoming
-  //   receiver.
-  void EnableTestingMode();
-
-  MessagePipeHandle handle() const { return connector_.handle(); }
-
-  // Returns true if this Router has any pending callbacks.
-  bool has_pending_responders() const {
-    DCHECK(thread_checker_.CalledOnValidThread());
-    return !async_responders_.empty() || !sync_responses_.empty();
-  }
-
- private:
-  // Maps from the id of a response to the MessageReceiver that handles the
-  // response.
-  using AsyncResponderMap =
-      std::map<uint64_t, std::unique_ptr<MessageReceiver>>;
-
-  struct SyncResponseInfo {
-   public:
-    explicit SyncResponseInfo(bool* in_response_received);
-    ~SyncResponseInfo();
-
-    std::unique_ptr<Message> response;
-
-    // Points to a stack-allocated variable.
-    bool* response_received;
-
-   private:
-    DISALLOW_COPY_AND_ASSIGN(SyncResponseInfo);
-  };
-
-  using SyncResponseMap = std::map<uint64_t, std::unique_ptr<SyncResponseInfo>>;
-
-  class HandleIncomingMessageThunk : public MessageReceiver {
-   public:
-    HandleIncomingMessageThunk(Router* router);
-    ~HandleIncomingMessageThunk() override;
-
-    // MessageReceiver implementation:
-    bool Accept(Message* message) override;
-
-   private:
-    Router* router_;
-  };
-
-  bool HandleIncomingMessage(Message* message);
-  void HandleQueuedMessages();
-  bool HandleMessageInternal(Message* message);
-
-  void OnConnectionError();
-
-  HandleIncomingMessageThunk thunk_;
-  FilterChain filters_;
-  Connector connector_;
-  MessageReceiverWithResponderStatus* incoming_receiver_;
-  AsyncResponderMap async_responders_;
-  SyncResponseMap sync_responses_;
-  uint64_t next_request_id_;
-  bool testing_mode_;
-  std::queue<std::unique_ptr<Message>> pending_messages_;
-  // Whether a task has been posted to trigger processing of
-  // |pending_messages_|.
-  bool pending_task_for_messages_;
-  bool encountered_error_;
-  base::Closure error_handler_;
-  base::ThreadChecker thread_checker_;
-  base::WeakPtrFactory<Router> weak_factory_;
-};
-
-}  // namespace internal
-}  // namespace mojo
-
-#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_ROUTER_H_
diff --git a/mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.cc b/mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.cc
index f54c3f7..c134507 100644
--- a/mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.cc
+++ b/mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.cc
@@ -4,69 +4,379 @@
 
 #include "mojo/public/cpp/bindings/scoped_interface_endpoint_handle.h"
 
+#include "base/bind.h"
 #include "base/logging.h"
+#include "base/synchronization/lock.h"
 #include "mojo/public/cpp/bindings/associated_group_controller.h"
+#include "mojo/public/cpp/bindings/lib/may_auto_lock.h"
 
 namespace mojo {
 
+// ScopedInterfaceEndpointHandle::State ----------------------------------------
+
+// State could be called from multiple threads.
+class ScopedInterfaceEndpointHandle::State
+    : public base::RefCountedThreadSafe<State> {
+ public:
+  State() = default;
+
+  State(InterfaceId id,
+        scoped_refptr<AssociatedGroupController> group_controller)
+      : id_(id), group_controller_(group_controller) {}
+
+  void InitPendingState(scoped_refptr<State> peer) {
+    DCHECK(!lock_);
+    DCHECK(!pending_association_);
+
+    lock_.emplace();
+    pending_association_ = true;
+    peer_state_ = std::move(peer);
+  }
+
+  void Close(const base::Optional<DisconnectReason>& reason) {
+    scoped_refptr<AssociatedGroupController> cached_group_controller;
+    InterfaceId cached_id = kInvalidInterfaceId;
+    scoped_refptr<State> cached_peer_state;
+
+    {
+      internal::MayAutoLock locker(&lock_);
+
+      if (!association_event_handler_.is_null()) {
+        association_event_handler_.Reset();
+        runner_ = nullptr;
+      }
+
+      if (!pending_association_) {
+        if (IsValidInterfaceId(id_)) {
+          // Intentionally keep |group_controller_| unchanged.
+          // That is because the callback created by
+          // CreateGroupControllerGetter() could still be used after this point,
+          // potentially from another thread. We would like it to continue
+          // returning the same group controller.
+          //
+          // Imagine there is a ThreadSafeForwarder A:
+          // (1) On the IO thread, A's underlying associated interface pointer
+          //     is closed.
+          // (2) On the proxy thread, the user makes a call on A to pass an
+          //     associated request B_asso_req. The callback returned by
+          //     CreateGroupControllerGetter() is used to associate B_asso_req.
+          // (3) On the proxy thread, the user immediately binds B_asso_ptr_info
+          //     to B_asso_ptr and makes calls on it.
+          //
+          // If we reset |group_controller_| in step (1), step (2) won't be able
+          // to associate B_asso_req. Therefore, in step (3) B_asso_ptr won't be
+          // able to serialize associated endpoints or send message because it
+          // is still in "pending_association" state and doesn't have a group
+          // controller.
+          //
+          // We could "address" this issue by ignoring messages if there isn't a
+          // group controller. But the side effect is that we cannot detect
+          // programming errors of "using associated interface pointer before
+          // sending associated request".
+
+          cached_group_controller = group_controller_;
+          cached_id = id_;
+          id_ = kInvalidInterfaceId;
+        }
+      } else {
+        pending_association_ = false;
+        cached_peer_state = std::move(peer_state_);
+      }
+    }
+
+    if (cached_group_controller) {
+      cached_group_controller->CloseEndpointHandle(cached_id, reason);
+    } else if (cached_peer_state) {
+      cached_peer_state->OnPeerClosedBeforeAssociation(reason);
+    }
+  }
+
+  void SetAssociationEventHandler(AssociationEventCallback handler) {
+    internal::MayAutoLock locker(&lock_);
+
+    if (!pending_association_ && !IsValidInterfaceId(id_))
+      return;
+
+    association_event_handler_ = std::move(handler);
+    if (association_event_handler_.is_null()) {
+      runner_ = nullptr;
+      return;
+    }
+
+    runner_ = base::ThreadTaskRunnerHandle::Get();
+    if (!pending_association_) {
+      runner_->PostTask(
+          FROM_HERE,
+          base::Bind(
+              &ScopedInterfaceEndpointHandle::State::RunAssociationEventHandler,
+              this, runner_, ASSOCIATED));
+    } else if (!peer_state_) {
+      runner_->PostTask(
+          FROM_HERE,
+          base::Bind(
+              &ScopedInterfaceEndpointHandle::State::RunAssociationEventHandler,
+              this, runner_, PEER_CLOSED_BEFORE_ASSOCIATION));
+    }
+  }
+
+  bool NotifyAssociation(
+      InterfaceId id,
+      scoped_refptr<AssociatedGroupController> peer_group_controller) {
+    scoped_refptr<State> cached_peer_state;
+    {
+      internal::MayAutoLock locker(&lock_);
+
+      DCHECK(pending_association_);
+      pending_association_ = false;
+      cached_peer_state = std::move(peer_state_);
+    }
+
+    if (cached_peer_state) {
+      cached_peer_state->OnAssociated(id, std::move(peer_group_controller));
+      return true;
+    }
+    return false;
+  }
+
+  bool is_valid() const {
+    internal::MayAutoLock locker(&lock_);
+    return pending_association_ || IsValidInterfaceId(id_);
+  }
+
+  bool pending_association() const {
+    internal::MayAutoLock locker(&lock_);
+    return pending_association_;
+  }
+
+  InterfaceId id() const {
+    internal::MayAutoLock locker(&lock_);
+    return id_;
+  }
+
+  AssociatedGroupController* group_controller() const {
+    internal::MayAutoLock locker(&lock_);
+    return group_controller_.get();
+  }
+
+  const base::Optional<DisconnectReason>& disconnect_reason() const {
+    internal::MayAutoLock locker(&lock_);
+    return disconnect_reason_;
+  }
+
+ private:
+  friend class base::RefCountedThreadSafe<State>;
+
+  ~State() {
+    DCHECK(!pending_association_);
+    DCHECK(!IsValidInterfaceId(id_));
+  }
+
+  // Called by the peer, maybe from a different thread.
+  void OnAssociated(InterfaceId id,
+                    scoped_refptr<AssociatedGroupController> group_controller) {
+    AssociationEventCallback handler;
+    {
+      internal::MayAutoLock locker(&lock_);
+
+      // There may be race between Close() of endpoint A and
+      // NotifyPeerAssociation() of endpoint A_peer on different threads.
+      // Therefore, it is possible that endpoint A has been closed but it
+      // still gets OnAssociated() call from its peer.
+      if (!pending_association_)
+        return;
+
+      pending_association_ = false;
+      peer_state_ = nullptr;
+      id_ = id;
+      group_controller_ = std::move(group_controller);
+
+      if (!association_event_handler_.is_null()) {
+        if (runner_->BelongsToCurrentThread()) {
+          handler = std::move(association_event_handler_);
+          runner_ = nullptr;
+        } else {
+          runner_->PostTask(FROM_HERE,
+                            base::Bind(&ScopedInterfaceEndpointHandle::State::
+                                           RunAssociationEventHandler,
+                                       this, runner_, ASSOCIATED));
+        }
+      }
+    }
+
+    if (!handler.is_null())
+      std::move(handler).Run(ASSOCIATED);
+  }
+
+  // Called by the peer, maybe from a different thread.
+  void OnPeerClosedBeforeAssociation(
+      const base::Optional<DisconnectReason>& reason) {
+    AssociationEventCallback handler;
+    {
+      internal::MayAutoLock locker(&lock_);
+
+      // There may be race between Close()/NotifyPeerAssociation() of endpoint
+      // A and Close() of endpoint A_peer on different threads.
+      // Therefore, it is possible that endpoint A is not in pending association
+      // state but still gets OnPeerClosedBeforeAssociation() call from its
+      // peer.
+      if (!pending_association_)
+        return;
+
+      disconnect_reason_ = reason;
+      // NOTE: This handle itself is still pending.
+      peer_state_ = nullptr;
+
+      if (!association_event_handler_.is_null()) {
+        if (runner_->BelongsToCurrentThread()) {
+          handler = std::move(association_event_handler_);
+          runner_ = nullptr;
+        } else {
+          runner_->PostTask(
+              FROM_HERE,
+              base::Bind(&ScopedInterfaceEndpointHandle::State::
+                             RunAssociationEventHandler,
+                         this, runner_, PEER_CLOSED_BEFORE_ASSOCIATION));
+        }
+      }
+    }
+
+    if (!handler.is_null())
+      std::move(handler).Run(PEER_CLOSED_BEFORE_ASSOCIATION);
+  }
+
+  void RunAssociationEventHandler(
+      scoped_refptr<base::SingleThreadTaskRunner> posted_to_runner,
+      AssociationEvent event) {
+    AssociationEventCallback handler;
+
+    {
+      internal::MayAutoLock locker(&lock_);
+      if (posted_to_runner == runner_) {
+        runner_ = nullptr;
+        handler = std::move(association_event_handler_);
+      }
+    }
+
+    if (!handler.is_null())
+      std::move(handler).Run(event);
+  }
+
+  // Protects the following members if the handle is initially set to pending
+  // association.
+  mutable base::Optional<base::Lock> lock_;
+
+  bool pending_association_ = false;
+  base::Optional<DisconnectReason> disconnect_reason_;
+
+  scoped_refptr<State> peer_state_;
+
+  AssociationEventCallback association_event_handler_;
+  scoped_refptr<base::SingleThreadTaskRunner> runner_;
+
+  InterfaceId id_ = kInvalidInterfaceId;
+  scoped_refptr<AssociatedGroupController> group_controller_;
+
+  DISALLOW_COPY_AND_ASSIGN(State);
+};
+
+// ScopedInterfaceEndpointHandle -----------------------------------------------
+
+// static
+void ScopedInterfaceEndpointHandle::CreatePairPendingAssociation(
+    ScopedInterfaceEndpointHandle* handle0,
+    ScopedInterfaceEndpointHandle* handle1) {
+  ScopedInterfaceEndpointHandle result0;
+  ScopedInterfaceEndpointHandle result1;
+  result0.state_->InitPendingState(result1.state_);
+  result1.state_->InitPendingState(result0.state_);
+
+  *handle0 = std::move(result0);
+  *handle1 = std::move(result1);
+}
+
 ScopedInterfaceEndpointHandle::ScopedInterfaceEndpointHandle()
-    : ScopedInterfaceEndpointHandle(kInvalidInterfaceId, true, nullptr) {}
+    : state_(new State) {}
 
 ScopedInterfaceEndpointHandle::ScopedInterfaceEndpointHandle(
     ScopedInterfaceEndpointHandle&& other)
-    : id_(other.id_), is_local_(other.is_local_) {
-  group_controller_.swap(other.group_controller_);
-  other.id_ = kInvalidInterfaceId;
+    : state_(new State) {
+  state_.swap(other.state_);
 }
 
 ScopedInterfaceEndpointHandle::~ScopedInterfaceEndpointHandle() {
-  reset();
+  state_->Close(base::nullopt);
 }
 
 ScopedInterfaceEndpointHandle& ScopedInterfaceEndpointHandle::operator=(
     ScopedInterfaceEndpointHandle&& other) {
   reset();
-  swap(other);
-
+  state_.swap(other.state_);
   return *this;
 }
 
-void ScopedInterfaceEndpointHandle::reset() {
-  if (!IsValidInterfaceId(id_))
-    return;
+bool ScopedInterfaceEndpointHandle::is_valid() const {
+  return state_->is_valid();
+}
+
+bool ScopedInterfaceEndpointHandle::pending_association() const {
+  return state_->pending_association();
+}
 
-  group_controller_->CloseEndpointHandle(id_, is_local_);
+InterfaceId ScopedInterfaceEndpointHandle::id() const {
+  return state_->id();
+}
 
-  id_ = kInvalidInterfaceId;
-  is_local_ = true;
-  group_controller_ = nullptr;
+AssociatedGroupController* ScopedInterfaceEndpointHandle::group_controller()
+    const {
+  return state_->group_controller();
 }
 
-void ScopedInterfaceEndpointHandle::swap(ScopedInterfaceEndpointHandle& other) {
-  using std::swap;
-  swap(other.id_, id_);
-  swap(other.is_local_, is_local_);
-  swap(other.group_controller_, group_controller_);
+const base::Optional<DisconnectReason>&
+ScopedInterfaceEndpointHandle::disconnect_reason() const {
+  return state_->disconnect_reason();
 }
 
-InterfaceId ScopedInterfaceEndpointHandle::release() {
-  InterfaceId result = id_;
+void ScopedInterfaceEndpointHandle::SetAssociationEventHandler(
+    AssociationEventCallback handler) {
+  state_->SetAssociationEventHandler(std::move(handler));
+}
 
-  id_ = kInvalidInterfaceId;
-  is_local_ = true;
-  group_controller_ = nullptr;
+void ScopedInterfaceEndpointHandle::reset() {
+  ResetInternal(base::nullopt);
+}
 
-  return result;
+void ScopedInterfaceEndpointHandle::ResetWithReason(
+    uint32_t custom_reason,
+    const std::string& description) {
+  ResetInternal(DisconnectReason(custom_reason, description));
 }
 
 ScopedInterfaceEndpointHandle::ScopedInterfaceEndpointHandle(
     InterfaceId id,
-    bool is_local,
     scoped_refptr<AssociatedGroupController> group_controller)
-    : id_(id),
-      is_local_(is_local),
-      group_controller_(std::move(group_controller)) {
-  DCHECK(!IsValidInterfaceId(id) || group_controller_);
+    : state_(new State(id, std::move(group_controller))) {
+  DCHECK(!IsValidInterfaceId(state_->id()) || state_->group_controller());
+}
+
+bool ScopedInterfaceEndpointHandle::NotifyAssociation(
+    InterfaceId id,
+    scoped_refptr<AssociatedGroupController> peer_group_controller) {
+  return state_->NotifyAssociation(id, peer_group_controller);
+}
+
+void ScopedInterfaceEndpointHandle::ResetInternal(
+    const base::Optional<DisconnectReason>& reason) {
+  scoped_refptr<State> new_state(new State);
+  state_->Close(reason);
+  state_.swap(new_state);
+}
+
+base::Callback<AssociatedGroupController*()>
+ScopedInterfaceEndpointHandle::CreateGroupControllerGetter() const {
+  // We allow this callback to be run on any thread. If this handle is created
+  // in non-pending state, we don't have a lock but it should still be safe
+  // because the group controller never changes.
+  return base::Bind(&State::group_controller, state_);
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/serialization.h b/mojo/public/cpp/bindings/lib/serialization.h
index 6d7dd8e..359b02b 100644
--- a/mojo/public/cpp/bindings/lib/serialization.h
+++ b/mojo/public/cpp/bindings/lib/serialization.h
@@ -8,19 +8,16 @@
 #include <string.h>
 
 #include "mojo/public/cpp/bindings/array_traits_carray.h"
-#include "mojo/public/cpp/bindings/array_traits_standard.h"
 #include "mojo/public/cpp/bindings/array_traits_stl.h"
 #include "mojo/public/cpp/bindings/lib/array_serialization.h"
-#include "mojo/public/cpp/bindings/lib/fixed_buffer.h"
+#include "mojo/public/cpp/bindings/lib/buffer.h"
 #include "mojo/public/cpp/bindings/lib/handle_interface_serialization.h"
 #include "mojo/public/cpp/bindings/lib/map_serialization.h"
 #include "mojo/public/cpp/bindings/lib/native_enum_serialization.h"
 #include "mojo/public/cpp/bindings/lib/native_struct_serialization.h"
 #include "mojo/public/cpp/bindings/lib/string_serialization.h"
 #include "mojo/public/cpp/bindings/lib/template_util.h"
-#include "mojo/public/cpp/bindings/map_traits_standard.h"
 #include "mojo/public/cpp/bindings/map_traits_stl.h"
-#include "mojo/public/cpp/bindings/string_traits_standard.h"
 #include "mojo/public/cpp/bindings/string_traits_stl.h"
 #include "mojo/public/cpp/bindings/string_traits_string16.h"
 #include "mojo/public/cpp/bindings/string_traits_string_piece.h"
@@ -44,7 +41,7 @@ DataArrayType StructSerializeImpl(UserType* input) {
   void* result_buffer = &result.front();
   // The serialization logic requires that the buffer is 8-byte aligned. If the
   // result buffer is not properly aligned, we have to do an extra copy. In
-  // practice, this should never happen for mojo::Array (backed by std::vector).
+  // practice, this should never happen for std::vector.
   bool need_copy = !IsAligned(result_buffer);
 
   if (need_copy) {
@@ -53,9 +50,9 @@ DataArrayType StructSerializeImpl(UserType* input) {
     DCHECK(IsAligned(result_buffer));
   }
 
-  FixedBuffer buffer;
+  Buffer buffer;
   buffer.Initialize(result_buffer, size);
-  typename MojomType::Struct::Data_* data = nullptr;
+  typename MojomTypeTraits<MojomType>::Data* data = nullptr;
   Serialize<MojomType>(*input, &buffer, &data, &context);
 
   if (need_copy) {
@@ -70,13 +67,11 @@ template <typename MojomType, typename DataArrayType, typename UserType>
 bool StructDeserializeImpl(const DataArrayType& input, UserType* output) {
   static_assert(BelongsTo<MojomType, MojomTypeCategory::STRUCT>::value,
                 "Unexpected type.");
-  using DataType = typename MojomType::Struct::Data_;
-
-  if (input.is_null())
-    return false;
+  using DataType = typename MojomTypeTraits<MojomType>::Data;
 
+  // TODO(sammc): Use DataArrayType::empty() once WTF::Vector::empty() exists.
   void* input_buffer =
-      input.empty()
+      input.size() == 0
           ? nullptr
           : const_cast<void*>(reinterpret_cast<const void*>(&input.front()));
 
@@ -89,7 +84,7 @@ bool StructDeserializeImpl(const DataArrayType& input, UserType* output) {
     memcpy(input_buffer, &input.front(), input.size());
   }
 
-  ValidationContext validation_context(input_buffer, input.size(), 0);
+  ValidationContext validation_context(input_buffer, input.size(), 0, 0);
   bool result = false;
   if (DataType::Validate(input_buffer, &validation_context)) {
     auto data = reinterpret_cast<DataType*>(input_buffer);
diff --git a/mojo/public/cpp/bindings/lib/serialization_context.cc b/mojo/public/cpp/bindings/lib/serialization_context.cc
index 7fd80be..e2fd5c6 100644
--- a/mojo/public/cpp/bindings/lib/serialization_context.cc
+++ b/mojo/public/cpp/bindings/lib/serialization_context.cc
@@ -7,7 +7,6 @@
 #include <limits>
 
 #include "base/logging.h"
-#include "mojo/public/cpp/bindings/associated_group_controller.h"
 #include "mojo/public/cpp/system/core.h"
 
 namespace mojo {
@@ -50,10 +49,6 @@ void SerializedHandleVector::Swap(std::vector<mojo::Handle>* other) {
 
 SerializationContext::SerializationContext() {}
 
-SerializationContext::SerializationContext(
-    scoped_refptr<AssociatedGroupController> in_group_controller)
-    : group_controller(std::move(in_group_controller)) {}
-
 SerializationContext::~SerializationContext() {
   DCHECK(!custom_contexts || custom_contexts->empty());
 }
diff --git a/mojo/public/cpp/bindings/lib/serialization_context.h b/mojo/public/cpp/bindings/lib/serialization_context.h
index 64d2a1a..a34fe3d 100644
--- a/mojo/public/cpp/bindings/lib/serialization_context.h
+++ b/mojo/public/cpp/bindings/lib/serialization_context.h
@@ -12,18 +12,16 @@
 #include <vector>
 
 #include "base/macros.h"
-#include "base/memory/ref_counted.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
+#include "mojo/public/cpp/bindings/scoped_interface_endpoint_handle.h"
 #include "mojo/public/cpp/system/handle.h"
 
 namespace mojo {
-
-class AssociatedGroupController;
-
 namespace internal {
 
 // A container for handles during serialization/deserialization.
-class SerializedHandleVector {
+class MOJO_CPP_BINDINGS_EXPORT SerializedHandleVector {
  public:
   SerializedHandleVector();
   ~SerializedHandleVector();
@@ -54,21 +52,23 @@ class SerializedHandleVector {
 };
 
 // Context information for serialization/deserialization routines.
-struct SerializationContext {
+struct MOJO_CPP_BINDINGS_EXPORT SerializationContext {
   SerializationContext();
-  explicit SerializationContext(
-      scoped_refptr<AssociatedGroupController> in_group_controller);
 
   ~SerializationContext();
 
-  // Used to serialize/deserialize associated interface pointers and requests.
-  scoped_refptr<AssociatedGroupController> group_controller;
-
   // Opaque context pointers returned by StringTraits::SetUpContext().
   std::unique_ptr<std::queue<void*>> custom_contexts;
 
   // Stashes handles encoded in a message by index.
   SerializedHandleVector handles;
+
+  // The number of ScopedInterfaceEndpointHandles that need to be serialized.
+  // It is calculated by PrepareToSerialize().
+  uint32_t associated_endpoint_count = 0;
+
+  // Stashes ScopedInterfaceEndpointHandles encoded in a message by index.
+  std::vector<ScopedInterfaceEndpointHandle> associated_endpoint_handles;
 };
 
 }  // namespace internal
diff --git a/mojo/public/cpp/bindings/lib/serialization_forward.h b/mojo/public/cpp/bindings/lib/serialization_forward.h
index 5bed126..55c9982 100644
--- a/mojo/public/cpp/bindings/lib/serialization_forward.h
+++ b/mojo/public/cpp/bindings/lib/serialization_forward.h
@@ -12,6 +12,7 @@
 #include "mojo/public/cpp/bindings/map_traits.h"
 #include "mojo/public/cpp/bindings/string_traits.h"
 #include "mojo/public/cpp/bindings/struct_traits.h"
+#include "mojo/public/cpp/bindings/union_traits.h"
 
 // This file is included by serialization implementation files to avoid circular
 // includes.
diff --git a/mojo/public/cpp/bindings/lib/string_serialization.h b/mojo/public/cpp/bindings/lib/string_serialization.h
index 5e65891..6e0c758 100644
--- a/mojo/public/cpp/bindings/lib/string_serialization.h
+++ b/mojo/public/cpp/bindings/lib/string_serialization.h
@@ -11,14 +11,14 @@
 #include "mojo/public/cpp/bindings/lib/array_internal.h"
 #include "mojo/public/cpp/bindings/lib/serialization_forward.h"
 #include "mojo/public/cpp/bindings/lib/serialization_util.h"
-#include "mojo/public/cpp/bindings/string.h"
+#include "mojo/public/cpp/bindings/string_data_view.h"
 #include "mojo/public/cpp/bindings/string_traits.h"
 
 namespace mojo {
 namespace internal {
 
 template <typename MaybeConstUserType>
-struct Serializer<String, MaybeConstUserType> {
+struct Serializer<StringDataView, MaybeConstUserType> {
   using UserType = typename std::remove_const<MaybeConstUserType>::type;
   using Traits = StringTraits<UserType>;
 
@@ -60,7 +60,7 @@ struct Serializer<String, MaybeConstUserType> {
                           SerializationContext* context) {
     if (!input)
       return CallSetToNullIfExists<Traits>(output);
-    return Traits::Read(StringDataView(input), output);
+    return Traits::Read(StringDataView(input, context), output);
   }
 };
 
diff --git a/mojo/public/cpp/bindings/lib/string_traits_wtf.cc b/mojo/public/cpp/bindings/lib/string_traits_wtf.cc
index 19fa907..203f6f5 100644
--- a/mojo/public/cpp/bindings/lib/string_traits_wtf.cc
+++ b/mojo/public/cpp/bindings/lib/string_traits_wtf.cc
@@ -16,7 +16,7 @@ namespace {
 struct UTF8AdaptorInfo {
   explicit UTF8AdaptorInfo(const WTF::String& input) : utf8_adaptor(input) {
 #if DCHECK_IS_ON()
-    original_size_in_bytes = static_cast<size_t>(input.sizeInBytes());
+    original_size_in_bytes = input.charactersSizeInBytes();
 #endif
   }
 
@@ -34,8 +34,7 @@ UTF8AdaptorInfo* ToAdaptor(const WTF::String& input, void* context) {
   UTF8AdaptorInfo* adaptor = static_cast<UTF8AdaptorInfo*>(context);
 
 #if DCHECK_IS_ON()
-  DCHECK_EQ(adaptor->original_size_in_bytes,
-            static_cast<size_t>(input.sizeInBytes()));
+  DCHECK_EQ(adaptor->original_size_in_bytes, input.charactersSizeInBytes());
 #endif
   return adaptor;
 }
diff --git a/mojo/public/cpp/bindings/lib/sync_call_restrictions.cc b/mojo/public/cpp/bindings/lib/sync_call_restrictions.cc
index 3d864af..585a8f0 100644
--- a/mojo/public/cpp/bindings/lib/sync_call_restrictions.cc
+++ b/mojo/public/cpp/bindings/lib/sync_call_restrictions.cc
@@ -6,6 +6,7 @@
 
 #if ENABLE_SYNC_CALL_RESTRICTIONS
 
+#include "base/debug/leak_annotations.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/threading/thread_local.h"
@@ -37,7 +38,7 @@ class SyncCallSettings {
   size_t scoped_allow_count_ = 0;
 };
 
-base::LazyInstance<base::ThreadLocalPointer<SyncCallSettings>>
+base::LazyInstance<base::ThreadLocalPointer<SyncCallSettings>>::DestructorAtExit
     g_sync_call_settings = LAZY_INSTANCE_INITIALIZER;
 
 // static
@@ -45,6 +46,7 @@ SyncCallSettings* SyncCallSettings::current() {
   SyncCallSettings* result = g_sync_call_settings.Pointer()->Get();
   if (!result) {
     result = new SyncCallSettings();
+    ANNOTATE_LEAKING_OBJECT_PTR(result);
     DCHECK_EQ(result, g_sync_call_settings.Pointer()->Get());
   }
   return result;
diff --git a/mojo/public/cpp/bindings/lib/sync_handle_registry.cc b/mojo/public/cpp/bindings/lib/sync_handle_registry.cc
index f6372d9..5ae763b 100644
--- a/mojo/public/cpp/bindings/lib/sync_handle_registry.cc
+++ b/mojo/public/cpp/bindings/lib/sync_handle_registry.cc
@@ -13,7 +13,7 @@
 namespace mojo {
 namespace {
 
-base::LazyInstance<base::ThreadLocalPointer<SyncHandleRegistry>>
+base::LazyInstance<base::ThreadLocalPointer<SyncHandleRegistry>>::Leaky
     g_current_sync_handle_watcher = LAZY_INSTANCE_INITIALIZER;
 
 }  // namespace
@@ -34,7 +34,7 @@ bool SyncHandleRegistry::RegisterHandle(const Handle& handle,
                                         const HandleCallback& callback) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  if (ContainsKey(handles_, handle))
+  if (base::ContainsKey(handles_, handle))
     return false;
 
   MojoResult result = MojoAddHandle(wait_set_handle_.get().value(),
@@ -48,7 +48,7 @@ bool SyncHandleRegistry::RegisterHandle(const Handle& handle,
 
 void SyncHandleRegistry::UnregisterHandle(const Handle& handle) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  if (!ContainsKey(handles_, handle))
+  if (!base::ContainsKey(handles_, handle))
     return;
 
   MojoResult result =
@@ -107,6 +107,19 @@ SyncHandleRegistry::SyncHandleRegistry() {
 
 SyncHandleRegistry::~SyncHandleRegistry() {
   DCHECK(thread_checker_.CalledOnValidThread());
+
+  // This object may be destructed after the thread local storage slot used by
+  // |g_current_sync_handle_watcher| is reset during thread shutdown.
+  // For example, another slot in the thread local storage holds a referrence to
+  // this object, and that slot is cleaned up after
+  // |g_current_sync_handle_watcher|.
+  if (!g_current_sync_handle_watcher.Pointer()->Get())
+    return;
+
+  // If this breaks, it is likely that the global variable is bulit into and
+  // accessed from multiple modules.
+  DCHECK_EQ(this, g_current_sync_handle_watcher.Pointer()->Get());
+
   g_current_sync_handle_watcher.Pointer()->Set(nullptr);
 }
 
diff --git a/mojo/public/cpp/bindings/lib/sync_handle_registry.h b/mojo/public/cpp/bindings/lib/sync_handle_registry.h
deleted file mode 100644
index d6b8c38..0000000
--- a/mojo/public/cpp/bindings/lib/sync_handle_registry.h
+++ /dev/null
@@ -1,67 +0,0 @@
-// Copyright 2016 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_SYNC_HANDLE_REGISTRY_H_
-#define MOJO_PUBLIC_CPP_BINDINGS_LIB_SYNC_HANDLE_REGISTRY_H_
-
-#include <unordered_map>
-
-#include "base/callback.h"
-#include "base/macros.h"
-#include "base/memory/ref_counted.h"
-#include "base/threading/thread_checker.h"
-#include "mojo/public/cpp/system/core.h"
-
-namespace mojo {
-namespace internal {
-
-// SyncHandleRegistry is a thread-local storage to register handles that want to
-// be watched together.
-//
-// This class is not thread safe.
-class SyncHandleRegistry : public base::RefCounted<SyncHandleRegistry> {
- public:
-  // Returns a thread-local object.
-  static scoped_refptr<SyncHandleRegistry> current();
-
-  using HandleCallback = base::Callback<void(MojoResult)>;
-  bool RegisterHandle(const Handle& handle,
-                      MojoHandleSignals handle_signals,
-                      const HandleCallback& callback);
-
-  void UnregisterHandle(const Handle& handle);
-
-  // Waits on all the registered handles and runs callbacks synchronously for
-  // those ready handles.
-  // The method:
-  //   - returns true when any element of |should_stop| is set to true;
-  //   - returns false when any error occurs.
-  bool WatchAllHandles(const bool* should_stop[], size_t count);
-
- private:
-  friend class base::RefCounted<SyncHandleRegistry>;
-
-  struct HandleHasher {
-    size_t operator()(const Handle& handle) const {
-      return std::hash<uint32_t>()(static_cast<uint32_t>(handle.value()));
-    }
-  };
-  using HandleMap = std::unordered_map<Handle, HandleCallback, HandleHasher>;
-
-  SyncHandleRegistry();
-  ~SyncHandleRegistry();
-
-  HandleMap handles_;
-
-  ScopedHandle wait_set_handle_;
-
-  base::ThreadChecker thread_checker_;
-
-  DISALLOW_COPY_AND_ASSIGN(SyncHandleRegistry);
-};
-
-}  // namespace internal
-}  // namespace mojo
-
-#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_SYNC_HANDLE_REGISTRY_H_
diff --git a/mojo/public/cpp/bindings/lib/validation_context.cc b/mojo/public/cpp/bindings/lib/validation_context.cc
index a95e07e..ad0a364 100644
--- a/mojo/public/cpp/bindings/lib/validation_context.cc
+++ b/mojo/public/cpp/bindings/lib/validation_context.cc
@@ -4,13 +4,7 @@
 
 #include "mojo/public/cpp/bindings/lib/validation_context.h"
 
-#include <stddef.h>
-#include <stdint.h>
-
 #include "base/logging.h"
-#include "mojo/public/cpp/bindings/lib/serialization_util.h"
-#include "mojo/public/cpp/bindings/message.h"
-#include "mojo/public/cpp/system/handle.h"
 
 namespace mojo {
 namespace internal {
@@ -18,69 +12,39 @@ namespace internal {
 ValidationContext::ValidationContext(const void* data,
                                      size_t data_num_bytes,
                                      size_t num_handles,
+                                     size_t num_associated_endpoint_handles,
                                      Message* message,
-                                     const base::StringPiece& description)
+                                     const base::StringPiece& description,
+                                     int stack_depth)
     : message_(message),
       description_(description),
       data_begin_(reinterpret_cast<uintptr_t>(data)),
       data_end_(data_begin_ + data_num_bytes),
       handle_begin_(0),
-      handle_end_(static_cast<uint32_t>(num_handles)) {
+      handle_end_(static_cast<uint32_t>(num_handles)),
+      associated_endpoint_handle_begin_(0),
+      associated_endpoint_handle_end_(
+          static_cast<uint32_t>(num_associated_endpoint_handles)),
+      stack_depth_(stack_depth) {
+  // Check whether the calculation of |data_end_| or static_cast from size_t to
+  // uint32_t causes overflow.
+  // They shouldn't happen but they do, set the corresponding range to empty.
   if (data_end_ < data_begin_) {
-    // The calculation of |data_end_| overflowed.
-    // It shouldn't happen but if it does, set the range to empty so
-    // IsValidRange() and ClaimMemory() always fail.
     NOTREACHED();
     data_end_ = data_begin_;
   }
   if (handle_end_ < num_handles) {
-    // Assigning |num_handles| to |handle_end_| overflowed.
-    // It shouldn't happen but if it does, set the handle index range to empty.
     NOTREACHED();
     handle_end_ = 0;
   }
+  if (associated_endpoint_handle_end_ < num_associated_endpoint_handles) {
+    NOTREACHED();
+    associated_endpoint_handle_end_ = 0;
+  }
 }
 
 ValidationContext::~ValidationContext() {
 }
 
-bool ValidationContext::ClaimMemory(const void* position, uint32_t num_bytes) {
-  uintptr_t begin = reinterpret_cast<uintptr_t>(position);
-  uintptr_t end = begin + num_bytes;
-
-  if (!InternalIsValidRange(begin, end))
-    return false;
-
-  data_begin_ = end;
-  return true;
-}
-
-bool ValidationContext::ClaimHandle(const Handle_Data& encoded_handle) {
-  uint32_t index = encoded_handle.value;
-  if (index == kEncodedInvalidHandleValue)
-    return true;
-
-  if (index < handle_begin_ || index >= handle_end_)
-    return false;
-
-  // |index| + 1 shouldn't overflow, because |index| is not the max value of
-  // uint32_t (it is less than |handle_end_|).
-  handle_begin_ = index + 1;
-  return true;
-}
-
-bool ValidationContext::IsValidRange(const void* position,
-                                     uint32_t num_bytes) const {
-  uintptr_t begin = reinterpret_cast<uintptr_t>(position);
-  uintptr_t end = begin + num_bytes;
-
-  return InternalIsValidRange(begin, end);
-}
-
-bool ValidationContext::InternalIsValidRange(uintptr_t begin,
-                                             uintptr_t end) const {
-  return end > begin && begin >= data_begin_ && end <= data_end_;
-}
-
 }  // namespace internal
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/validation_context.h b/mojo/public/cpp/bindings/lib/validation_context.h
index 6045ca8..ed6c654 100644
--- a/mojo/public/cpp/bindings/lib/validation_context.h
+++ b/mojo/public/cpp/bindings/lib/validation_context.h
@@ -8,23 +8,28 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include "base/compiler_specific.h"
 #include "base/macros.h"
 #include "base/strings/string_piece.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
 
+static const int kMaxRecursionDepth = 100;
+
 namespace mojo {
 
-class Handle;
 class Message;
 
 namespace internal {
 
 // ValidationContext is used when validating object sizes, pointers and handle
 // indices in the payload of incoming messages.
-class ValidationContext {
+class MOJO_CPP_BINDINGS_EXPORT ValidationContext {
  public:
   // [data, data + data_num_bytes) specifies the initial valid memory range.
   // [0, num_handles) specifies the initial valid range of handle indices.
+  // [0, num_associated_endpoint_handles) specifies the initial valid range of
+  // associated endpoint handle indices.
   //
   // If provided, |message| and |description| provide additional information
   // to use when reporting validation errors. In addition if |message| is
@@ -33,8 +38,10 @@ class ValidationContext {
   ValidationContext(const void* data,
                     size_t data_num_bytes,
                     size_t num_handles,
+                    size_t num_associated_endpoint_handles,
                     Message* message = nullptr,
-                    const base::StringPiece& description = "");
+                    const base::StringPiece& description = "",
+                    int stack_depth = 0);
 
   ~ValidationContext();
 
@@ -43,24 +50,97 @@ class ValidationContext {
   // the comments for IsValidRange().)
   // On success, the valid memory range is shrinked to begin right after the end
   // of the claimed range.
-  bool ClaimMemory(const void* position, uint32_t num_bytes);
+  bool ClaimMemory(const void* position, uint32_t num_bytes) {
+    uintptr_t begin = reinterpret_cast<uintptr_t>(position);
+    uintptr_t end = begin + num_bytes;
+
+    if (!InternalIsValidRange(begin, end))
+      return false;
+
+    data_begin_ = end;
+    return true;
+  }
 
   // Claims the specified encoded handle (which is basically a handle index).
   // The method succeeds if:
   // - |encoded_handle|'s value is |kEncodedInvalidHandleValue|.
   // - the handle is contained inside the valid range of handle indices. In this
   // case, the valid range is shinked to begin right after the claimed handle.
-  bool ClaimHandle(const Handle_Data& encoded_handle);
+  bool ClaimHandle(const Handle_Data& encoded_handle) {
+    uint32_t index = encoded_handle.value;
+    if (index == kEncodedInvalidHandleValue)
+      return true;
+
+    if (index < handle_begin_ || index >= handle_end_)
+      return false;
+
+    // |index| + 1 shouldn't overflow, because |index| is not the max value of
+    // uint32_t (it is less than |handle_end_|).
+    handle_begin_ = index + 1;
+    return true;
+  }
+
+  // Claims the specified encoded associated endpoint handle.
+  // The method succeeds if:
+  // - |encoded_handle|'s value is |kEncodedInvalidHandleValue|.
+  // - the handle is contained inside the valid range of associated endpoint
+  //   handle indices. In this case, the valid range is shinked to begin right
+  //   after the claimed handle.
+  bool ClaimAssociatedEndpointHandle(
+      const AssociatedEndpointHandle_Data& encoded_handle) {
+    uint32_t index = encoded_handle.value;
+    if (index == kEncodedInvalidHandleValue)
+      return true;
+
+    if (index < associated_endpoint_handle_begin_ ||
+        index >= associated_endpoint_handle_end_)
+      return false;
+
+    // |index| + 1 shouldn't overflow, because |index| is not the max value of
+    // uint32_t (it is less than |associated_endpoint_handle_end_|).
+    associated_endpoint_handle_begin_ = index + 1;
+    return true;
+  }
 
   // Returns true if the specified range is not empty, and the range is
   // contained inside the valid memory range.
-  bool IsValidRange(const void* position, uint32_t num_bytes) const;
+  bool IsValidRange(const void* position, uint32_t num_bytes) const {
+    uintptr_t begin = reinterpret_cast<uintptr_t>(position);
+    uintptr_t end = begin + num_bytes;
+
+    return InternalIsValidRange(begin, end);
+  }
+
+  // This object should be created on the stack once every time we recurse down
+  // into a subfield during validation to make sure we don't recurse too deep
+  // and blow the stack.
+  class ScopedDepthTracker {
+   public:
+    // |ctx| must outlive this object.
+    explicit ScopedDepthTracker(ValidationContext* ctx) : ctx_(ctx) {
+      ++ctx_->stack_depth_;
+    }
+
+    ~ScopedDepthTracker() { --ctx_->stack_depth_; }
+
+   private:
+    ValidationContext* ctx_;
+
+    DISALLOW_COPY_AND_ASSIGN(ScopedDepthTracker);
+  };
+
+  // Returns true if the recursion depth limit has been reached.
+  bool ExceedsMaxDepth() WARN_UNUSED_RESULT {
+    return stack_depth_ > kMaxRecursionDepth;
+  }
 
   Message* message() const { return message_; }
   const base::StringPiece& description() const { return description_; }
 
  private:
-  bool InternalIsValidRange(uintptr_t begin, uintptr_t end) const;
+  bool InternalIsValidRange(uintptr_t begin, uintptr_t end) const {
+    return end > begin && begin >= data_begin_ && end <= data_end_;
+  }
 
   Message* const message_;
   const base::StringPiece description_;
@@ -73,6 +153,13 @@ class ValidationContext {
   uint32_t handle_begin_;
   uint32_t handle_end_;
 
+  // [associated_endpoint_handle_begin_, associated_endpoint_handle_end_) is the
+  // valid associated endpoint handle index range.
+  uint32_t associated_endpoint_handle_begin_;
+  uint32_t associated_endpoint_handle_end_;
+
+  int stack_depth_;
+
   DISALLOW_COPY_AND_ASSIGN(ValidationContext);
 };
 
diff --git a/mojo/public/cpp/bindings/lib/validation_errors.cc b/mojo/public/cpp/bindings/lib/validation_errors.cc
index 90652de..904f5e4 100644
--- a/mojo/public/cpp/bindings/lib/validation_errors.cc
+++ b/mojo/public/cpp/bindings/lib/validation_errors.cc
@@ -14,6 +14,7 @@ namespace {
 ValidationErrorObserverForTesting* g_validation_error_observer = nullptr;
 SerializationWarningObserverForTesting* g_serialization_warning_observer =
     nullptr;
+bool g_suppress_logging = false;
 
 }  // namespace
 
@@ -55,6 +56,8 @@ const char* ValidationErrorToString(ValidationError error) {
       return "VALIDATION_ERROR_UNKNOWN_ENUM_VALUE";
     case VALIDATION_ERROR_DESERIALIZATION_FAILED:
       return "VALIDATION_ERROR_DESERIALIZATION_FAILED";
+    case VALIDATION_ERROR_MAX_RECURSION_DEPTH:
+      return "VALIDATION_ERROR_MAX_RECURSION_DEPTH";
   }
 
   return "Unknown error";
@@ -69,8 +72,10 @@ void ReportValidationError(ValidationContext* context,
   }
 
   if (description) {
-    LOG(ERROR) << "Invalid message: " << ValidationErrorToString(error) << " ("
-               << description << ")";
+    if (!g_suppress_logging) {
+      LOG(ERROR) << "Invalid message: " << ValidationErrorToString(error)
+                 << " (" << description << ")";
+    }
     if (context->message()) {
       context->message()->NotifyBadMessage(
           base::StringPrintf("Validation failed for %s [%s (%s)]",
@@ -78,7 +83,8 @@ void ReportValidationError(ValidationContext* context,
                              ValidationErrorToString(error), description));
     }
   } else {
-    LOG(ERROR) << "Invalid message: " << ValidationErrorToString(error);
+    if (!g_suppress_logging)
+      LOG(ERROR) << "Invalid message: " << ValidationErrorToString(error);
     if (context->message()) {
       context->message()->NotifyBadMessage(
           base::StringPrintf("Validation failed for %s [%s]",
@@ -88,6 +94,25 @@ void ReportValidationError(ValidationContext* context,
   }
 }
 
+void ReportValidationErrorForMessage(
+    mojo::Message* message,
+    ValidationError error,
+    const char* description) {
+  ValidationContext validation_context(nullptr, 0, 0, 0, message, description);
+  ReportValidationError(&validation_context, error);
+}
+
+ScopedSuppressValidationErrorLoggingForTests
+    ::ScopedSuppressValidationErrorLoggingForTests()
+    : was_suppressed_(g_suppress_logging) {
+  g_suppress_logging = true;
+}
+
+ScopedSuppressValidationErrorLoggingForTests
+    ::~ScopedSuppressValidationErrorLoggingForTests() {
+  g_suppress_logging = was_suppressed_;
+}
+
 ValidationErrorObserverForTesting::ValidationErrorObserverForTesting(
     const base::Closure& callback)
     : last_error_(VALIDATION_ERROR_NONE), callback_(callback) {
diff --git a/mojo/public/cpp/bindings/lib/validation_errors.h b/mojo/public/cpp/bindings/lib/validation_errors.h
index ec0aa27..122418d 100644
--- a/mojo/public/cpp/bindings/lib/validation_errors.h
+++ b/mojo/public/cpp/bindings/lib/validation_errors.h
@@ -8,9 +8,13 @@
 #include "base/callback.h"
 #include "base/logging.h"
 #include "base/macros.h"
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/validation_context.h"
 
 namespace mojo {
+
+class Message;
+
 namespace internal {
 
 enum ValidationError {
@@ -67,17 +71,41 @@ enum ValidationError {
   // Message deserialization failure, for example due to rejection by custom
   // validation logic.
   VALIDATION_ERROR_DESERIALIZATION_FAILED,
+  // The message contains a too deeply nested value, for example a recursively
+  // defined field which runtime value is too large.
+  VALIDATION_ERROR_MAX_RECURSION_DEPTH,
 };
 
-const char* ValidationErrorToString(ValidationError error);
+MOJO_CPP_BINDINGS_EXPORT const char* ValidationErrorToString(
+    ValidationError error);
+
+MOJO_CPP_BINDINGS_EXPORT void ReportValidationError(
+    ValidationContext* context,
+    ValidationError error,
+    const char* description = nullptr);
 
-void ReportValidationError(ValidationContext* context,
-                           ValidationError error,
-                           const char* description = nullptr);
+MOJO_CPP_BINDINGS_EXPORT void ReportValidationErrorForMessage(
+    mojo::Message* message,
+    ValidationError error,
+    const char* description = nullptr);
+
+// This class may be used by tests to suppress validation error logging. This is
+// not thread-safe and must only be instantiated on the main thread with no
+// other threads using Mojo bindings at the time of construction or destruction.
+class MOJO_CPP_BINDINGS_EXPORT ScopedSuppressValidationErrorLoggingForTests {
+ public:
+  ScopedSuppressValidationErrorLoggingForTests();
+  ~ScopedSuppressValidationErrorLoggingForTests();
+
+ private:
+  const bool was_suppressed_;
+
+  DISALLOW_COPY_AND_ASSIGN(ScopedSuppressValidationErrorLoggingForTests);
+};
 
 // Only used by validation tests and when there is only one thread doing message
 // validation.
-class ValidationErrorObserverForTesting {
+class MOJO_CPP_BINDINGS_EXPORT ValidationErrorObserverForTesting {
  public:
   explicit ValidationErrorObserverForTesting(const base::Closure& callback);
   ~ValidationErrorObserverForTesting();
@@ -99,11 +127,11 @@ class ValidationErrorObserverForTesting {
 //
 // The function returns true if the error is recorded (by a
 // SerializationWarningObserverForTesting object), false otherwise.
-bool ReportSerializationWarning(ValidationError error);
+MOJO_CPP_BINDINGS_EXPORT bool ReportSerializationWarning(ValidationError error);
 
 // Only used by serialization tests and when there is only one thread doing
 // message serialization.
-class SerializationWarningObserverForTesting {
+class MOJO_CPP_BINDINGS_EXPORT SerializationWarningObserverForTesting {
  public:
   SerializationWarningObserverForTesting();
   ~SerializationWarningObserverForTesting();
diff --git a/mojo/public/cpp/bindings/lib/validation_util.cc b/mojo/public/cpp/bindings/lib/validation_util.cc
index 9e63521..7614df5 100644
--- a/mojo/public/cpp/bindings/lib/validation_util.cc
+++ b/mojo/public/cpp/bindings/lib/validation_util.cc
@@ -16,16 +16,6 @@
 namespace mojo {
 namespace internal {
 
-bool ValidateEncodedPointer(const uint64_t* offset) {
-  // - Make sure |*offset| is no more than 32-bits.
-  // - Cast |offset| to uintptr_t so overflow behavior is well defined across
-  //   32-bit and 64-bit systems.
-  return *offset <= std::numeric_limits<uint32_t>::max() &&
-         (reinterpret_cast<uintptr_t>(offset) +
-              static_cast<uint32_t>(*offset) >=
-          reinterpret_cast<uintptr_t>(offset));
-}
-
 bool ValidateStructHeaderAndClaimMemory(const void* data,
                                         ValidationContext* validation_context) {
   if (!IsAligned(data)) {
@@ -56,20 +46,17 @@ bool ValidateStructHeaderAndClaimMemory(const void* data,
   return true;
 }
 
-bool ValidateUnionHeaderAndClaimMemory(const void* data,
-                                       bool inlined,
-                                       ValidationContext* validation_context) {
+bool ValidateNonInlinedUnionHeaderAndClaimMemory(
+    const void* data,
+    ValidationContext* validation_context) {
   if (!IsAligned(data)) {
     ReportValidationError(validation_context,
                           VALIDATION_ERROR_MISALIGNED_OBJECT);
     return false;
   }
 
-  // If the union is inlined in another structure its memory was already
-  // claimed.
-  // This ONLY applies to the union itself, NOT anything which the union points
-  // to.
-  if (!inlined && !validation_context->ClaimMemory(data, kUnionDataSize)) {
+  if (!validation_context->ClaimMemory(data, kUnionDataSize) ||
+      *static_cast<const uint32_t*>(data) != kUnionDataSize) {
     ReportValidationError(validation_context,
                           VALIDATION_ERROR_ILLEGAL_MEMORY_RANGE);
     return false;
@@ -113,41 +100,12 @@ bool ValidateMessageIsResponse(const Message* message,
   return true;
 }
 
-bool ValidateControlRequest(const Message* message,
-                            ValidationContext* validation_context) {
-  switch (message->header()->name) {
-    case kRunMessageId:
-      return ValidateMessageIsRequestExpectingResponse(message,
-                                                       validation_context) &&
-             ValidateMessagePayload<RunMessageParams_Data>(message,
-                                                           validation_context);
-    case kRunOrClosePipeMessageId:
-      return ValidateMessageIsRequestWithoutResponse(message,
-                                                     validation_context) &&
-          ValidateMessagePayload<RunOrClosePipeMessageParams_Data>(
-              message, validation_context);
-  }
-  return false;
-}
-
-bool ValidateControlResponse(const Message* message,
-                             ValidationContext* validation_context) {
-  if (!ValidateMessageIsResponse(message, validation_context))
-    return false;
-  switch (message->header()->name) {
-    case kRunMessageId:
-      return ValidateMessagePayload<RunResponseMessageParams_Data>(
-          message, validation_context);
-  }
-  return false;
-}
-
 bool IsHandleOrInterfaceValid(const AssociatedInterface_Data& input) {
-  return IsValidInterfaceId(input.interface_id);
+  return input.handle.is_valid();
 }
 
-bool IsHandleOrInterfaceValid(const AssociatedInterfaceRequest_Data& input) {
-  return IsValidInterfaceId(input.interface_id);
+bool IsHandleOrInterfaceValid(const AssociatedEndpointHandle_Data& input) {
+  return input.is_valid();
 }
 
 bool IsHandleOrInterfaceValid(const Interface_Data& input) {
@@ -172,7 +130,7 @@ bool ValidateHandleOrInterfaceNonNullable(
 }
 
 bool ValidateHandleOrInterfaceNonNullable(
-    const AssociatedInterfaceRequest_Data& input,
+    const AssociatedEndpointHandle_Data& input,
     const char* error_message,
     ValidationContext* validation_context) {
   if (IsHandleOrInterfaceValid(input))
@@ -212,7 +170,7 @@ bool ValidateHandleOrInterfaceNonNullable(
 
 bool ValidateHandleOrInterface(const AssociatedInterface_Data& input,
                                ValidationContext* validation_context) {
-  if (!IsMasterInterfaceId(input.interface_id))
+  if (validation_context->ClaimAssociatedEndpointHandle(input.handle))
     return true;
 
   ReportValidationError(validation_context,
@@ -220,9 +178,9 @@ bool ValidateHandleOrInterface(const AssociatedInterface_Data& input,
   return false;
 }
 
-bool ValidateHandleOrInterface(const AssociatedInterfaceRequest_Data& input,
+bool ValidateHandleOrInterface(const AssociatedEndpointHandle_Data& input,
                                ValidationContext* validation_context) {
-  if (!IsMasterInterfaceId(input.interface_id))
+  if (validation_context->ClaimAssociatedEndpointHandle(input))
     return true;
 
   ReportValidationError(validation_context,
diff --git a/mojo/public/cpp/bindings/lib/validation_util.h b/mojo/public/cpp/bindings/lib/validation_util.h
index c883392..ea5a991 100644
--- a/mojo/public/cpp/bindings/lib/validation_util.h
+++ b/mojo/public/cpp/bindings/lib/validation_util.h
@@ -7,6 +7,7 @@
 
 #include <stdint.h>
 
+#include "mojo/public/cpp/bindings/bindings_export.h"
 #include "mojo/public/cpp/bindings/lib/bindings_internal.h"
 #include "mojo/public/cpp/bindings/lib/serialization_util.h"
 #include "mojo/public/cpp/bindings/lib/validate_params.h"
@@ -19,7 +20,15 @@ namespace internal {
 
 // Checks whether decoding the pointer will overflow and produce a pointer
 // smaller than |offset|.
-bool ValidateEncodedPointer(const uint64_t* offset);
+inline bool ValidateEncodedPointer(const uint64_t* offset) {
+  // - Make sure |*offset| is no more than 32-bits.
+  // - Cast |offset| to uintptr_t so overflow behavior is well defined across
+  //   32-bit and 64-bit systems.
+  return *offset <= std::numeric_limits<uint32_t>::max() &&
+         (reinterpret_cast<uintptr_t>(offset) +
+              static_cast<uint32_t>(*offset) >=
+          reinterpret_cast<uintptr_t>(offset));
+}
 
 template <typename T>
 bool ValidatePointer(const Pointer<T>& input,
@@ -38,30 +47,32 @@ bool ValidatePointer(const Pointer<T>& input,
 // |validation_context|. On success, the memory range is marked as occupied.
 // Note: Does not verify |version| or that |num_bytes| is correct for the
 // claimed version.
-bool ValidateStructHeaderAndClaimMemory(const void* data,
-                                        ValidationContext* validation_context);
+MOJO_CPP_BINDINGS_EXPORT bool ValidateStructHeaderAndClaimMemory(
+    const void* data,
+    ValidationContext* validation_context);
 
 // Validates that |data| contains a valid union header, in terms of alignment
-// and size. If not inlined, it checks that the memory range
-// [data, data + num_bytes) is not marked as occupied by other objects in
-// |validation_context|. On success, the memory range is marked as occupied.
-bool ValidateUnionHeaderAndClaimMemory(const void* data,
-                                       bool inlined,
-                                       ValidationContext* validation_context);
+// and size. It checks that the memory range [data, data + kUnionDataSize) is
+// not marked as occupied by other objects in |validation_context|. On success,
+// the memory range is marked as occupied.
+MOJO_CPP_BINDINGS_EXPORT bool ValidateNonInlinedUnionHeaderAndClaimMemory(
+    const void* data,
+    ValidationContext* validation_context);
 
 // Validates that the message is a request which doesn't expect a response.
-bool ValidateMessageIsRequestWithoutResponse(
+MOJO_CPP_BINDINGS_EXPORT bool ValidateMessageIsRequestWithoutResponse(
     const Message* message,
     ValidationContext* validation_context);
 
 // Validates that the message is a request expecting a response.
-bool ValidateMessageIsRequestExpectingResponse(
+MOJO_CPP_BINDINGS_EXPORT bool ValidateMessageIsRequestExpectingResponse(
     const Message* message,
     ValidationContext* validation_context);
 
 // Validates that the message is a response.
-bool ValidateMessageIsResponse(const Message* message,
-                               ValidationContext* validation_context);
+MOJO_CPP_BINDINGS_EXPORT bool ValidateMessageIsResponse(
+    const Message* message,
+    ValidationContext* validation_context);
 
 // Validates that the message payload is a valid struct of type ParamsType.
 template <typename ParamsType>
@@ -70,13 +81,6 @@ bool ValidateMessagePayload(const Message* message,
   return ParamsType::Validate(message->payload(), validation_context);
 }
 
-// The following methods validate control messages defined in
-// interface_control_messages.mojom.
-bool ValidateControlRequest(const Message* message,
-                            ValidationContext* validation_context);
-bool ValidateControlResponse(const Message* message,
-                             ValidationContext* validation_context);
-
 // The following Validate.*NonNullable() functions validate that the given
 // |input| is not null/invalid.
 template <typename T>
@@ -105,24 +109,28 @@ bool ValidateInlinedUnionNonNullable(const T& input,
   return false;
 }
 
-bool IsHandleOrInterfaceValid(const AssociatedInterface_Data& input);
-bool IsHandleOrInterfaceValid(const AssociatedInterfaceRequest_Data& input);
-bool IsHandleOrInterfaceValid(const Interface_Data& input);
-bool IsHandleOrInterfaceValid(const Handle_Data& input);
+MOJO_CPP_BINDINGS_EXPORT bool IsHandleOrInterfaceValid(
+    const AssociatedInterface_Data& input);
+MOJO_CPP_BINDINGS_EXPORT bool IsHandleOrInterfaceValid(
+    const AssociatedEndpointHandle_Data& input);
+MOJO_CPP_BINDINGS_EXPORT bool IsHandleOrInterfaceValid(
+    const Interface_Data& input);
+MOJO_CPP_BINDINGS_EXPORT bool IsHandleOrInterfaceValid(
+    const Handle_Data& input);
 
-bool ValidateHandleOrInterfaceNonNullable(
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterfaceNonNullable(
     const AssociatedInterface_Data& input,
     const char* error_message,
     ValidationContext* validation_context);
-bool ValidateHandleOrInterfaceNonNullable(
-    const AssociatedInterfaceRequest_Data& input,
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterfaceNonNullable(
+    const AssociatedEndpointHandle_Data& input,
     const char* error_message,
     ValidationContext* validation_context);
-bool ValidateHandleOrInterfaceNonNullable(
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterfaceNonNullable(
     const Interface_Data& input,
     const char* error_message,
     ValidationContext* validation_context);
-bool ValidateHandleOrInterfaceNonNullable(
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterfaceNonNullable(
     const Handle_Data& input,
     const char* error_message,
     ValidationContext* validation_context);
@@ -131,6 +139,12 @@ template <typename T>
 bool ValidateContainer(const Pointer<T>& input,
                        ValidationContext* validation_context,
                        const ContainerValidateParams* validate_params) {
+  ValidationContext::ScopedDepthTracker depth_tracker(validation_context);
+  if (validation_context->ExceedsMaxDepth()) {
+    ReportValidationError(validation_context,
+                          VALIDATION_ERROR_MAX_RECURSION_DEPTH);
+    return false;
+  }
   return ValidatePointer(input, validation_context) &&
          T::Validate(input.Get(), validation_context, validate_params);
 }
@@ -138,6 +152,12 @@ bool ValidateContainer(const Pointer<T>& input,
 template <typename T>
 bool ValidateStruct(const Pointer<T>& input,
                     ValidationContext* validation_context) {
+  ValidationContext::ScopedDepthTracker depth_tracker(validation_context);
+  if (validation_context->ExceedsMaxDepth()) {
+    ReportValidationError(validation_context,
+                          VALIDATION_ERROR_MAX_RECURSION_DEPTH);
+    return false;
+  }
   return ValidatePointer(input, validation_context) &&
          T::Validate(input.Get(), validation_context);
 }
@@ -145,24 +165,40 @@ bool ValidateStruct(const Pointer<T>& input,
 template <typename T>
 bool ValidateInlinedUnion(const T& input,
                           ValidationContext* validation_context) {
+  ValidationContext::ScopedDepthTracker depth_tracker(validation_context);
+  if (validation_context->ExceedsMaxDepth()) {
+    ReportValidationError(validation_context,
+                          VALIDATION_ERROR_MAX_RECURSION_DEPTH);
+    return false;
+  }
   return T::Validate(&input, validation_context, true);
 }
 
 template <typename T>
 bool ValidateNonInlinedUnion(const Pointer<T>& input,
                              ValidationContext* validation_context) {
+  ValidationContext::ScopedDepthTracker depth_tracker(validation_context);
+  if (validation_context->ExceedsMaxDepth()) {
+    ReportValidationError(validation_context,
+                          VALIDATION_ERROR_MAX_RECURSION_DEPTH);
+    return false;
+  }
   return ValidatePointer(input, validation_context) &&
          T::Validate(input.Get(), validation_context, false);
 }
 
-bool ValidateHandleOrInterface(const AssociatedInterface_Data& input,
-                               ValidationContext* validation_context);
-bool ValidateHandleOrInterface(const AssociatedInterfaceRequest_Data& input,
-                               ValidationContext* validation_context);
-bool ValidateHandleOrInterface(const Interface_Data& input,
-                               ValidationContext* validation_context);
-bool ValidateHandleOrInterface(const Handle_Data& input,
-                               ValidationContext* validation_context);
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterface(
+    const AssociatedInterface_Data& input,
+    ValidationContext* validation_context);
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterface(
+    const AssociatedEndpointHandle_Data& input,
+    ValidationContext* validation_context);
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterface(
+    const Interface_Data& input,
+    ValidationContext* validation_context);
+MOJO_CPP_BINDINGS_EXPORT bool ValidateHandleOrInterface(
+    const Handle_Data& input,
+    ValidationContext* validation_context);
 
 }  // namespace internal
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/lib/wtf_clone_equals_util.h b/mojo/public/cpp/bindings/lib/wtf_clone_equals_util.h
index edbf27b..cb24bc4 100644
--- a/mojo/public/cpp/bindings/lib/wtf_clone_equals_util.h
+++ b/mojo/public/cpp/bindings/lib/wtf_clone_equals_util.h
@@ -7,14 +7,14 @@
 
 #include <type_traits>
 
-#include "mojo/public/cpp/bindings/lib/clone_equals_util.h"
+#include "mojo/public/cpp/bindings/clone_traits.h"
+#include "mojo/public/cpp/bindings/lib/equals_traits.h"
 #include "third_party/WebKit/Source/wtf/HashMap.h"
 #include "third_party/WebKit/Source/wtf/Optional.h"
 #include "third_party/WebKit/Source/wtf/Vector.h"
 #include "third_party/WebKit/Source/wtf/text/WTFString.h"
 
 namespace mojo {
-namespace internal {
 
 template <typename T>
 struct CloneTraits<WTF::Vector<T>, false> {
@@ -22,7 +22,7 @@ struct CloneTraits<WTF::Vector<T>, false> {
     WTF::Vector<T> result;
     result.reserveCapacity(input.size());
     for (const auto& element : input)
-      result.append(internal::Clone(element));
+      result.push_back(mojo::Clone(element));
 
     return result;
   }
@@ -34,11 +34,13 @@ struct CloneTraits<WTF::HashMap<K, V>, false> {
     WTF::HashMap<K, V> result;
     auto input_end = input.end();
     for (auto it = input.begin(); it != input_end; ++it)
-      result.add(internal::Clone(it->key), internal::Clone(it->value));
+      result.add(mojo::Clone(it->key), mojo::Clone(it->value));
     return result;
   }
 };
 
+namespace internal {
+
 template <typename T>
 struct EqualsTraits<WTF::Vector<T>, false> {
   static bool Equals(const WTF::Vector<T>& a, const WTF::Vector<T>& b) {
diff --git a/mojo/public/cpp/bindings/lib/wtf_serialization.h b/mojo/public/cpp/bindings/lib/wtf_serialization.h
index 132e19c..0f112b9 100644
--- a/mojo/public/cpp/bindings/lib/wtf_serialization.h
+++ b/mojo/public/cpp/bindings/lib/wtf_serialization.h
@@ -5,9 +5,7 @@
 #ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_WTF_SERIALIZATION_H_
 #define MOJO_PUBLIC_CPP_BINDINGS_LIB_WTF_SERIALIZATION_H_
 
-#include "mojo/public/cpp/bindings/array_traits_wtf.h"
 #include "mojo/public/cpp/bindings/array_traits_wtf_vector.h"
-#include "mojo/public/cpp/bindings/map_traits_wtf.h"
 #include "mojo/public/cpp/bindings/map_traits_wtf_hash_map.h"
 #include "mojo/public/cpp/bindings/string_traits_wtf.h"