Add support for logging information about mojo messages retrieved by the mojo debugger

from the message pipes.

This includes the following:-
1. Number of bytes in the message.
2. Number of fields in the message.
3. The message id.
4. The request id of the message.
5. Whether the message expects a response message.
6. Whether the message is a response message.
7. The URL of the service.

We also intercept the first message pipe which is used for
vending services in the debugger.

Support for tracing additional information will be added as needed.

BUG=360188
TBR=viettrungluu

Review URL: https://codereview.chromium.org/354043003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@282417 0039d316-1c4b-4281-b951-d872f2087c98

mojo/mojo.gyp

@@ -514,6 +514,7 @@
       'includes': [ 'public/tools/bindings/mojom_bindings_generator.gypi' ],
       'sources': [
         'spy/public/spy.mojom',
+        'spy/common.h',
         'spy/spy.cc',
         'spy/spy.h',
         'spy/spy_server_impl.h',

mojo/spy/common.h

+// 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_SPY_COMMON_H_
+#define MOJO_SPY_COMMON_H_
+
+#include <stdint.h>
+
+namespace mojo {
+
+#pragma pack(push, 1)
+
+// Mojo message header structures. These are based off the Mojo spec.
+
+enum {
+  kMessageExpectsResponse = 1 << 0,
+  kMessageIsResponse      = 1 << 1
+};
+
+struct MojoCommonHeader {
+  uint32_t num_bytes;
+  uint32_t num_fields;
+};
+
+struct MojoMessageHeader : public MojoCommonHeader {
+  uint32_t name;
+  uint32_t flags;
+};
+
+struct MojoRequestHeader : public MojoMessageHeader {
+  uint64_t request_id;
+};
+
+struct MojoMessageData  {
+  MojoRequestHeader header;
+};
+
+#pragma pack(pop)
+
+}  // namespace mojo
+
+#endif  // MOJO_SPY_COMMON_H_

mojo/spy/spy.cc

@@ -9,15 +9,17 @@
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
+#include "base/logging.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop/message_loop_proxy.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_split.h"
 #include "base/threading/thread.h"
 #include "base/threading/worker_pool.h"
-
+#include "base/time/time.h"
 #include "mojo/public/cpp/system/core.h"
 #include "mojo/service_manager/service_manager.h"
+#include "mojo/spy/common.h"
 #include "mojo/spy/public/spy.mojom.h"
 #include "mojo/spy/spy_server_impl.h"
 #include "mojo/spy/websocket_server.h"
@@ -25,6 +27,8 @@
 
 namespace {
 
+mojo::WebSocketServer* ws_server = NULL;
+
 const size_t kMessageBufSize = 2 * 1024;
 const size_t kHandleBufSize = 64;
 const int kDefaultWebSocketPort = 42424;
@@ -39,9 +43,11 @@ void CloseHandles(MojoHandle* handles, size_t count) {
 class MessageProcessor :
     public base::RefCountedThreadSafe<MessageProcessor> {
  public:
-  MessageProcessor()
+  MessageProcessor(base::MessageLoopProxy* control_loop_proxy)
       : last_result_(MOJO_RESULT_OK),
-        bytes_transfered_(0) {
+        bytes_transfered_(0),
+        control_loop_proxy_(control_loop_proxy),
+        service_vendor_message_pipe_received_(false) {
     message_count_[0] = 0;
     message_count_[1] = 0;
     handle_count_[0] = 0;
@@ -49,7 +55,8 @@ class MessageProcessor :
   }
 
   void Start(mojo::ScopedMessagePipeHandle client,
-             mojo::ScopedMessagePipeHandle interceptor) {
+             mojo::ScopedMessagePipeHandle interceptor,
+             const GURL& url) {
     std::vector<mojo::MessagePipeHandle> pipes;
     pipes.push_back(client.get());
     pipes.push_back(interceptor.get());
@@ -86,12 +93,42 @@ class MessageProcessor :
       if (!bytes_read && !handles_read)
         continue;
 
-      if (handles_read)
+      if (handles_read) {
         handle_count_[r] += handles_read;
 
+        // Intercept the first set of handles to message pipes with the
+        // assumption that these would be used for vending mojo services.
+        // TODO(ananta)
+        // The above approach is hacky and could cause us to miss other message
+        // pipes which could be exchanged between the client and the server.
+        // Look into a cleaner way of identifying message pipe handles.
+        if (!service_vendor_message_pipe_received_) {
+          service_vendor_message_pipe_received_ = true;
+          for (uint32_t i = 0; i < handles_read; i++) {
+            mojo::ScopedMessagePipeHandle message_pipe_handle;
+            message_pipe_handle.reset(mojo::MessagePipeHandle(hbuf[i]));
+
+            mojo::ScopedMessagePipeHandle faux_client;
+            mojo::ScopedMessagePipeHandle interceptor;
+            CreateMessagePipe(NULL, &faux_client, &interceptor);
+
+            base::WorkerPool::PostTask(
+                FROM_HERE,
+                base::Bind(&MessageProcessor::Start,
+                            this,
+                            base::Passed(&message_pipe_handle),
+                              base::Passed(&interceptor),
+                            url),
+                true);
+            hbuf.get()[i] = faux_client.release().value();
+          }
+        }
+      }
       ++message_count_[r];
       bytes_transfered_ += bytes_read;
 
+      LogMessageInfo(mbuf.get(), url);
+
       mojo::MessagePipeHandle write_handle = (r == 0) ? pipes[1] : pipes[0];
       if (!CheckResult(Wait(write_handle,
                             MOJO_HANDLE_SIGNAL_WRITABLE,
@@ -121,18 +158,80 @@ class MessageProcessor :
     return false;
   }
 
+  void LogInvalidMessage(const mojo::MojoMessageHeader& header) {
+    LOG(ERROR) << "Invalid message: Number of Fields: "
+               << header.num_fields
+               << " Number of bytes: "
+               << header.num_bytes
+               << " Flags: "
+               << header.flags;
+  }
+
+  // Validates the message as per the mojo spec.
+  bool IsValidMessage(const mojo::MojoMessageHeader& header) {
+    if (header.num_fields == 2) {
+      if (header.num_bytes != sizeof(mojo::MojoMessageHeader)) {
+        LogInvalidMessage(header);
+        return false;
+      }
+    } else if (header.num_fields == 3) {
+      if (header.num_bytes != sizeof(mojo::MojoRequestHeader)) {
+        LogInvalidMessage(header);
+      }
+    } else if (header.num_fields > 3) {
+      if (header.num_bytes < sizeof(mojo::MojoRequestHeader)) {
+        LogInvalidMessage(header);
+        return false;
+      }
+    }
+    // These flags should be specified in request or response messages.
+    if (header.num_fields < 3 &&
+          ((header.flags & mojo::kMessageExpectsResponse) ||
+           (header.flags & mojo::kMessageIsResponse))) {
+      LOG(ERROR) << "Invalid request message.";
+      LogInvalidMessage(header);
+      return false;
+    }
+    // These flags are mutually exclusive.
+    if ((header.flags & mojo::kMessageExpectsResponse) &&
+        (header.flags & mojo::kMessageIsResponse)) {
+      LOG(ERROR) << "Invalid flags combination in request message.";
+      LogInvalidMessage(header);
+      return false;
+    }
+    return true;
+  }
+
+  void LogMessageInfo(void* data, const GURL& url) {
+    mojo::MojoMessageData* message_data =
+        reinterpret_cast<mojo::MojoMessageData*>(data);
+    if (IsValidMessage(message_data->header)) {
+      control_loop_proxy_->PostTask(
+          FROM_HERE,
+          base::Bind(&mojo::WebSocketServer::LogMessageInfo,
+                      base::Unretained(ws_server),
+                      message_data->header, url, base::Time::Now()));
+    }
+  }
+
   MojoResult last_result_;
   uint32_t bytes_transfered_;
   uint32_t message_count_[2];
   uint32_t handle_count_[2];
+  scoped_refptr<base::MessageLoopProxy> control_loop_proxy_;
+  // This flag helps us intercept the first message pipe exchanged between
+  // the client and the service vendor.
+  bool service_vendor_message_pipe_received_;
 };
 
 // In charge of intercepting access to the service manager.
 class SpyInterceptor : public mojo::ServiceManager::Interceptor {
  public:
-  explicit SpyInterceptor(scoped_refptr<mojo::SpyServerImpl> spy_server)
+  explicit SpyInterceptor(scoped_refptr<mojo::SpyServerImpl> spy_server,
+                          base::MessageLoopProxy* control_loop_proxy)
       : spy_server_(spy_server),
-        proxy_(base::MessageLoopProxy::current()) {
+        proxy_(base::MessageLoopProxy::current()),
+        control_loop_proxy_(control_loop_proxy){
   }
 
  private:
@@ -150,13 +249,15 @@ class SpyInterceptor : public mojo::ServiceManager::Interceptor {
       mojo::ScopedMessagePipeHandle interceptor;
       CreateMessagePipe(NULL, &faux_client, &interceptor);
 
-      scoped_refptr<MessageProcessor> processor = new MessageProcessor();
+      scoped_refptr<MessageProcessor> processor = new MessageProcessor(
+          control_loop_proxy_);
       mojo::ScopedMessagePipeHandle real_handle = real_client.PassMessagePipe();
       base::WorkerPool::PostTask(
           FROM_HERE,
           base::Bind(&MessageProcessor::Start,
                      processor,
-                     base::Passed(&real_handle), base::Passed(&interceptor)),
+                     base::Passed(&real_handle), base::Passed(&interceptor),
+                     url),
           true);
 
       mojo::ServiceProviderPtr faux_provider;
@@ -174,10 +275,9 @@ class SpyInterceptor : public mojo::ServiceManager::Interceptor {
 
   scoped_refptr<mojo::SpyServerImpl> spy_server_;
   scoped_refptr<base::MessageLoopProxy> proxy_;
+  scoped_refptr<base::MessageLoopProxy> control_loop_proxy_;
 };
 
-mojo::WebSocketServer* ws_server = NULL;
-
 void StartWebServer(int port, mojo::ScopedMessagePipeHandle pipe) {
   // TODO(cpu) figure out lifetime of the server. See Spy() dtor.
   ws_server = new mojo::WebSocketServer(port, pipe.Pass());
@@ -228,7 +328,8 @@ Spy::Spy(mojo::ServiceManager* service_manager, const std::string& options) {
                             base::Passed(spy_server_->ServerPipe())));
 
   // Start intercepting mojo services.
-  service_manager->SetInterceptor(new SpyInterceptor(spy_server_));
+  service_manager->SetInterceptor(new SpyInterceptor(
+      spy_server_, control_thread_->message_loop_proxy()));
 }
 
 Spy::~Spy() {

mojo/spy/test/spy_repl_test.html

-<!DOCTYPE html>
+<!DOCTYPE html>
 
 <html lang="en">
 <head>
@@ -36,7 +36,7 @@
         };
         conn.onmessage = function (event) {
           var message = event.data;
-          log.innerHTML = message;
+          log.innerHTML += "<br/>" + message;
         };
         conn.onclose = function (event) {
           state.innerHTML = 'connection closed';

mojo/spy/websocket_server.cc

@@ -7,20 +7,29 @@
 #include <string>
 
 #include "base/bind.h"
+#include "base/logging.h"
 #include "base/strings/stringprintf.h"
-
 #include "mojo/public/cpp/bindings/message.h"
-
 #include "net/base/ip_endpoint.h"
 #include "net/base/net_errors.h"
 #include "net/server/http_server_request_info.h"
 #include "net/server/http_server_response_info.h"
 #include "net/socket/tcp_listen_socket.h"
+#include "url/gurl.h"
 
 namespace mojo {
 
 const int kNotConnected = -1;
 
+#define MOJO_DEBUGGER_MESSAGE_FORMAT "\"url: %s\n," \
+                                     "time: %02d:%02d:%02d\n,"\
+                                     "bytes: %u\n,"\
+                                     "fields: %u\n,"\
+                                     "name: %u\n,"\
+                                     "requestId: %llu\n,"\
+                                     "type: %s\n,"\
+                                     "end:\n\""
+
 WebSocketServer::WebSocketServer(int port,
                                  mojo::ScopedMessagePipeHandle server_pipe)
     : port_(port),
@@ -41,6 +50,36 @@ bool WebSocketServer::Start() {
   return (error == net::OK);
 }
 
+void WebSocketServer::LogMessageInfo(
+    const mojo::MojoRequestHeader& message_header,
+    const GURL& url,
+    const base::Time& message_time) {
+  base::Time::Exploded exploded;
+  message_time.LocalExplode(&exploded);
+
+  std::string output_message = base::StringPrintf(
+      MOJO_DEBUGGER_MESSAGE_FORMAT,
+      url.spec().c_str(),
+      exploded.hour,
+      exploded.minute,
+      exploded.second,
+      static_cast<unsigned>(message_header.num_bytes),
+      static_cast<unsigned>(message_header.num_fields),
+      static_cast<unsigned>(message_header.name),
+      static_cast<unsigned long long>(
+          message_header.num_fields == 3 ? message_header.request_id
+              : 0),
+      message_header.flags != 0 ?
+        (message_header.flags & mojo::kMessageExpectsResponse ?
+            "Expects response" : "Response message")
+        : "Not a request or response message");
+  if (Connected()) {
+    web_server_->SendOverWebSocket(connection_id_, output_message.c_str());
+  } else {
+    DVLOG(1) << output_message;
+  }
+}
+
 void WebSocketServer::OnHttpRequest(
     int connection_id,
     const net::HttpServerRequestInfo& info) {
@@ -113,4 +152,9 @@ void WebSocketServer::OnStartSession(spy_api::Result, mojo::String) {
   web_server_->SendOverWebSocket(connection_id_, "\"ok start\"");
 }
 
+bool WebSocketServer::Connected() const {
+  return connection_id_ !=  kNotConnected;
+}
+
 }  // namespace mojo
+

mojo/spy/websocket_server.h

@@ -7,9 +7,16 @@
 
 #include <string>
 
+#include "mojo/spy/common.h"
 #include "mojo/spy/public/spy.mojom.h"
 #include "net/server/http_server.h"
 
+namespace base {
+class Time;
+};
+
+class GURL;
+
 namespace mojo {
 
 class WebSocketServer : public net::HttpServer::Delegate,
@@ -23,6 +30,12 @@ class WebSocketServer : public net::HttpServer::Delegate,
   // Returns the listening port, useful if 0 was passed to the contructor.
   int port() const { return port_; }
 
+  // Maintains a log of the message passed in.
+  void LogMessageInfo(
+      const mojo::MojoRequestHeader& message_header,
+      const GURL& url,
+      const base::Time& message_time);
+
  protected:
   // Overridden from net::HttpServer::Delegate.
   virtual void OnHttpRequest(
@@ -48,6 +61,8 @@ class WebSocketServer : public net::HttpServer::Delegate,
   // Callbacks from calling spy_api::SpyServer.
   void OnStartSession(spy_api::Result, mojo::String);
 
+  bool Connected() const;
+
  private:
   int port_;
   int connection_id_;