OLD | NEW |
---|---|
1 // Copyright 2014 The Chromium Authors. All rights reserved. | 1 // Copyright 2014 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #include "mojo/spy/spy.h" | 5 #include "mojo/spy/spy.h" |
6 | 6 |
7 #include "base/bind.h" | 7 #include "base/bind.h" |
8 #include "base/compiler_specific.h" | 8 #include "base/compiler_specific.h" |
9 #include "base/location.h" | 9 #include "base/location.h" |
10 #include "base/logging.h" | |
10 #include "base/memory/ref_counted.h" | 11 #include "base/memory/ref_counted.h" |
12 #include "base/time/time.h" | |
11 #include "base/strings/string_number_conversions.h" | 13 #include "base/strings/string_number_conversions.h" |
12 #include "base/strings/string_split.h" | 14 #include "base/strings/string_split.h" |
13 #include "base/threading/thread.h" | 15 #include "base/threading/thread.h" |
14 #include "base/threading/worker_pool.h" | 16 #include "base/threading/worker_pool.h" |
15 | 17 |
16 #include "mojo/public/cpp/system/core.h" | 18 #include "mojo/public/cpp/system/core.h" |
17 #include "mojo/service_manager/service_manager.h" | 19 #include "mojo/service_manager/service_manager.h" |
20 #include "mojo/spy/common.h" | |
18 #include "mojo/spy/websocket_server.h" | 21 #include "mojo/spy/websocket_server.h" |
19 | 22 |
20 namespace { | 23 namespace { |
21 | 24 |
25 spy::WebSocketServer* ws_server = NULL; | |
26 | |
22 const size_t kMessageBufSize = 2 * 1024; | 27 const size_t kMessageBufSize = 2 * 1024; |
23 const size_t kHandleBufSize = 64; | 28 const size_t kHandleBufSize = 64; |
24 const int kDefaultWebSocketPort = 42424; | 29 const int kDefaultWebSocketPort = 42424; |
25 | 30 |
26 void CloseHandles(MojoHandle* handles, size_t count) { | 31 void CloseHandles(MojoHandle* handles, size_t count) { |
27 for (size_t ix = 0; ix != count; ++count) | 32 for (size_t ix = 0; ix != count; ++count) |
28 MojoClose(handles[ix]); | 33 MojoClose(handles[ix]); |
29 } | 34 } |
30 | 35 |
31 // In charge of processing messages that flow over a | 36 // In charge of processing messages that flow over a |
32 // single message pipe. | 37 // single message pipe. |
33 class MessageProcessor : | 38 class MessageProcessor : |
34 public base::RefCountedThreadSafe<MessageProcessor> { | 39 public base::RefCountedThreadSafe<MessageProcessor> { |
35 public: | 40 public: |
36 | 41 |
37 MessageProcessor() | 42 MessageProcessor(base::MessageLoopProxy* control_loop_proxy) |
38 : last_result_(MOJO_RESULT_OK), | 43 : last_result_(MOJO_RESULT_OK), |
39 bytes_transfered_(0) { | 44 bytes_transfered_(0), |
45 control_loop_proxy_(control_loop_proxy) { | |
40 | 46 |
41 message_count_[0] = 0; | 47 message_count_[0] = 0; |
42 message_count_[1] = 0; | 48 message_count_[1] = 0; |
43 handle_count_[0] = 0; | 49 handle_count_[0] = 0; |
44 handle_count_[1] = 0; | 50 handle_count_[1] = 0; |
45 } | 51 } |
46 | 52 |
47 void Start(mojo::ScopedMessagePipeHandle client, | 53 void Start(mojo::ScopedMessagePipeHandle client, |
48 mojo::ScopedMessagePipeHandle interceptor) { | 54 mojo::ScopedMessagePipeHandle interceptor, |
55 const GURL& url) { | |
49 std::vector<mojo::MessagePipeHandle> pipes; | 56 std::vector<mojo::MessagePipeHandle> pipes; |
50 pipes.push_back(client.get()); | 57 pipes.push_back(client.get()); |
51 pipes.push_back(interceptor.get()); | 58 pipes.push_back(interceptor.get()); |
52 std::vector<MojoHandleSignals> handle_signals; | 59 std::vector<MojoHandleSignals> handle_signals; |
53 handle_signals.push_back(MOJO_HANDLE_SIGNAL_READABLE); | 60 handle_signals.push_back(MOJO_HANDLE_SIGNAL_READABLE); |
54 handle_signals.push_back(MOJO_HANDLE_SIGNAL_READABLE); | 61 handle_signals.push_back(MOJO_HANDLE_SIGNAL_READABLE); |
55 | 62 |
56 scoped_ptr<char[]> mbuf(new char[kMessageBufSize]); | 63 scoped_ptr<char[]> mbuf(new char[kMessageBufSize]); |
57 scoped_ptr<MojoHandle[]> hbuf(new MojoHandle[kHandleBufSize]); | 64 scoped_ptr<MojoHandle[]> hbuf(new MojoHandle[kHandleBufSize]); |
58 | 65 |
(...skipping 22 matching lines...) Expand all Loading... | |
81 | 88 |
82 if (!bytes_read && !handles_read) | 89 if (!bytes_read && !handles_read) |
83 continue; | 90 continue; |
84 | 91 |
85 if (handles_read) | 92 if (handles_read) |
86 handle_count_[r] += handles_read; | 93 handle_count_[r] += handles_read; |
87 | 94 |
88 ++message_count_[r]; | 95 ++message_count_[r]; |
89 bytes_transfered_ += bytes_read; | 96 bytes_transfered_ += bytes_read; |
90 | 97 |
98 LogMessageInfo(mbuf.get(), url); | |
99 | |
91 mojo::MessagePipeHandle write_handle = (r == 0) ? pipes[1] : pipes[0]; | 100 mojo::MessagePipeHandle write_handle = (r == 0) ? pipes[1] : pipes[0]; |
92 if (!CheckResult(Wait(write_handle, | 101 if (!CheckResult(Wait(write_handle, |
93 MOJO_HANDLE_SIGNAL_WRITABLE, | 102 MOJO_HANDLE_SIGNAL_WRITABLE, |
94 MOJO_DEADLINE_INDEFINITE))) | 103 MOJO_DEADLINE_INDEFINITE))) |
95 break; | 104 break; |
96 | 105 |
97 if (!CheckResult(WriteMessageRaw(write_handle, | 106 if (!CheckResult(WriteMessageRaw(write_handle, |
98 mbuf.get(), bytes_read, | 107 mbuf.get(), bytes_read, |
99 hbuf.get(), handles_read, | 108 hbuf.get(), handles_read, |
100 MOJO_WRITE_MESSAGE_FLAG_NONE))) { | 109 MOJO_WRITE_MESSAGE_FLAG_NONE))) { |
101 // On failure we own the handles. For now just close them. | 110 // On failure we own the handles. For now just close them. |
102 if (handles_read) | 111 if (handles_read) |
103 CloseHandles(hbuf.get(), handles_read); | 112 CloseHandles(hbuf.get(), handles_read); |
104 break; | 113 break; |
105 } | 114 } |
106 } | 115 } |
107 } | 116 } |
108 | 117 |
109 private: | 118 private: |
110 friend class base::RefCountedThreadSafe<MessageProcessor>; | 119 friend class base::RefCountedThreadSafe<MessageProcessor>; |
111 virtual ~MessageProcessor() {} | 120 virtual ~MessageProcessor() {} |
112 | 121 |
113 bool CheckResult(MojoResult mr) { | 122 bool CheckResult(MojoResult mr) { |
114 if (mr == MOJO_RESULT_OK) | 123 if (mr == MOJO_RESULT_OK) |
115 return true; | 124 return true; |
116 last_result_ = mr; | 125 last_result_ = mr; |
117 return false; | 126 return false; |
118 } | 127 } |
119 | 128 |
129 void LogMessageInfo(void* data, const GURL& url) { | |
130 mojo::MojoMessageData* message_data = | |
131 reinterpret_cast<mojo::MojoMessageData*>(data); | |
viettrungluu
2014/06/30 19:16:54
static_cast
More importantly, before you do so, y
ananta
2014/07/03 00:57:23
Added a function to validate the message as per th
| |
132 control_loop_proxy_->PostTask( | |
133 FROM_HERE, | |
134 base::Bind(&spy::WebSocketServer::LogMessageInfo, | |
135 base::Unretained(ws_server), | |
136 message_data->header, url, base::Time::Now())); | |
137 } | |
138 | |
120 MojoResult last_result_; | 139 MojoResult last_result_; |
121 uint32_t bytes_transfered_; | 140 uint32_t bytes_transfered_; |
122 uint32_t message_count_[2]; | 141 uint32_t message_count_[2]; |
123 uint32_t handle_count_[2]; | 142 uint32_t handle_count_[2]; |
143 scoped_refptr<base::MessageLoopProxy> control_loop_proxy_; | |
124 }; | 144 }; |
125 | 145 |
126 // In charge of intercepting access to the service manager. | 146 // In charge of intercepting access to the service manager. |
127 class SpyInterceptor : public mojo::ServiceManager::Interceptor { | 147 class SpyInterceptor : public mojo::ServiceManager::Interceptor { |
148 public: | |
149 SpyInterceptor(base::MessageLoopProxy* control_loop_proxy) | |
150 : control_loop_proxy_(control_loop_proxy) {} | |
151 | |
128 private: | 152 private: |
129 virtual mojo::ServiceProviderPtr OnConnectToClient( | 153 virtual mojo::ServiceProviderPtr OnConnectToClient( |
130 const GURL& url, mojo::ServiceProviderPtr real_client) OVERRIDE { | 154 const GURL& url, mojo::ServiceProviderPtr real_client) OVERRIDE { |
131 if (!MustIntercept(url)) | 155 if (!MustIntercept(url)) |
132 return real_client.Pass(); | 156 return real_client.Pass(); |
133 | 157 |
134 // You can get an invalid handle if the app (or service) is | 158 // You can get an invalid handle if the app (or service) is |
135 // created by unconventional means, for example the command line. | 159 // created by unconventional means, for example the command line. |
136 if (!real_client.get()) | 160 if (!real_client.get()) |
137 return real_client.Pass(); | 161 return real_client.Pass(); |
138 | 162 |
139 mojo::ScopedMessagePipeHandle faux_client; | 163 mojo::ScopedMessagePipeHandle faux_client; |
140 mojo::ScopedMessagePipeHandle interceptor; | 164 mojo::ScopedMessagePipeHandle interceptor; |
141 CreateMessagePipe(NULL, &faux_client, &interceptor); | 165 CreateMessagePipe(NULL, &faux_client, &interceptor); |
142 | 166 |
143 scoped_refptr<MessageProcessor> processor = new MessageProcessor(); | 167 scoped_refptr<MessageProcessor> processor = new MessageProcessor( |
168 control_loop_proxy_); | |
144 mojo::ScopedMessagePipeHandle real_handle = real_client.PassMessagePipe(); | 169 mojo::ScopedMessagePipeHandle real_handle = real_client.PassMessagePipe(); |
145 base::WorkerPool::PostTask( | 170 base::WorkerPool::PostTask( |
146 FROM_HERE, | 171 FROM_HERE, |
147 base::Bind(&MessageProcessor::Start, | 172 base::Bind(&MessageProcessor::Start, |
148 processor, | 173 processor, |
149 base::Passed(&real_handle), base::Passed(&interceptor)), | 174 base::Passed(&real_handle), base::Passed(&interceptor), |
175 url), | |
150 true); | 176 true); |
151 | 177 |
152 mojo::ServiceProviderPtr faux_provider; | 178 mojo::ServiceProviderPtr faux_provider; |
153 faux_provider.Bind(faux_client.Pass()); | 179 faux_provider.Bind(faux_client.Pass()); |
154 return faux_provider.Pass(); | 180 return faux_provider.Pass(); |
155 } | 181 } |
156 | 182 |
157 bool MustIntercept(const GURL& url) { | 183 bool MustIntercept(const GURL& url) { |
158 // TODO(cpu): manage who and when to intercept. | 184 // TODO(cpu): manage who and when to intercept. |
159 return true; | 185 return true; |
160 } | 186 } |
187 | |
188 scoped_refptr<base::MessageLoopProxy> control_loop_proxy_; | |
161 }; | 189 }; |
162 | 190 |
163 spy::WebSocketServer* ws_server = NULL; | |
164 | |
165 void StartServer(int port) { | 191 void StartServer(int port) { |
166 // TODO(cpu) figure out lifetime of the server. See Spy() dtor. | 192 // TODO(cpu) figure out lifetime of the server. See Spy() dtor. |
167 ws_server = new spy::WebSocketServer(port); | 193 ws_server = new spy::WebSocketServer(port); |
168 ws_server->Start(); | 194 ws_server->Start(); |
169 } | 195 } |
170 | 196 |
171 struct SpyOptions { | 197 struct SpyOptions { |
172 int websocket_port; | 198 int websocket_port; |
173 | 199 |
174 SpyOptions() | 200 SpyOptions() |
(...skipping 25 matching lines...) Expand all Loading... | |
200 Spy::Spy(mojo::ServiceManager* service_manager, const std::string& options) { | 226 Spy::Spy(mojo::ServiceManager* service_manager, const std::string& options) { |
201 SpyOptions spy_options = ProcessOptions(options); | 227 SpyOptions spy_options = ProcessOptions(options); |
202 // Start the tread what will accept commands from the frontend. | 228 // Start the tread what will accept commands from the frontend. |
203 control_thread_.reset(new base::Thread("mojo_spy_control_thread")); | 229 control_thread_.reset(new base::Thread("mojo_spy_control_thread")); |
204 base::Thread::Options thread_options(base::MessageLoop::TYPE_IO, 0); | 230 base::Thread::Options thread_options(base::MessageLoop::TYPE_IO, 0); |
205 control_thread_->StartWithOptions(thread_options); | 231 control_thread_->StartWithOptions(thread_options); |
206 control_thread_->message_loop_proxy()->PostTask( | 232 control_thread_->message_loop_proxy()->PostTask( |
207 FROM_HERE, base::Bind(&StartServer, spy_options.websocket_port)); | 233 FROM_HERE, base::Bind(&StartServer, spy_options.websocket_port)); |
208 | 234 |
209 // Start intercepting mojo services. | 235 // Start intercepting mojo services. |
210 service_manager->SetInterceptor(new SpyInterceptor()); | 236 service_manager->SetInterceptor(new SpyInterceptor( |
237 control_thread_->message_loop_proxy())); | |
211 } | 238 } |
212 | 239 |
213 Spy::~Spy(){ | 240 Spy::~Spy(){ |
214 // TODO(cpu): Do not leak the interceptor. Lifetime between the | 241 // TODO(cpu): Do not leak the interceptor. Lifetime between the |
215 // service_manager and the spy is still unclear hence the leak. | 242 // service_manager and the spy is still unclear hence the leak. |
216 } | 243 } |
217 | 244 |
218 } // namespace mojo | 245 } // namespace mojo |
OLD | NEW |