blob: 27510e757ed5524b9d4bf3851699753e70bc43be (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
// 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"
namespace mojo {
namespace internal {
// ----------------------------------------------------------------------------
class ResponderThunk : public MessageReceiver {
public:
explicit ResponderThunk(const SharedData<Router*>& router)
: router_(router) {
}
virtual ~ResponderThunk() {
}
// MessageReceiver implementation:
virtual bool Accept(Message* message) MOJO_OVERRIDE {
assert(message->has_flag(kMessageIsResponse));
bool result = false;
Router* router = router_.value();
if (router)
result = router->Accept(message);
return result;
}
private:
SharedData<Router*> router_;
};
// ----------------------------------------------------------------------------
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,
const MojoAsyncWaiter* waiter)
: thunk_(this),
filters_(filters.Pass()),
connector_(message_pipe.Pass(), waiter),
weak_self_(this),
incoming_receiver_(NULL),
next_request_id_(0) {
filters_.SetSink(&thunk_);
connector_.set_incoming_receiver(filters_.GetHead());
}
Router::~Router() {
weak_self_.set_value(NULL);
for (ResponderMap::const_iterator i = responders_.begin();
i != responders_.end(); ++i) {
delete i->second;
}
}
bool Router::Accept(Message* message) {
assert(!message->has_flag(kMessageExpectsResponse));
return connector_.Accept(message);
}
bool Router::AcceptWithResponder(Message* message,
MessageReceiver* responder) {
assert(message->has_flag(kMessageExpectsResponse));
// 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_++;
message->set_request_id(request_id);
if (!connector_.Accept(message))
return false;
// We assume ownership of |responder|.
responders_[request_id] = responder;
return true;
}
bool Router::HandleIncomingMessage(Message* message) {
if (message->has_flag(kMessageExpectsResponse)) {
if (incoming_receiver_) {
MessageReceiver* responder = new ResponderThunk(weak_self_);
bool ok = incoming_receiver_->AcceptWithResponder(message, responder);
if (!ok)
delete responder;
return ok;
}
// If we receive a request expecting a response when the client is not
// listening, then we have no choice but to tear down the pipe.
connector_.CloseMessagePipe();
} else if (message->has_flag(kMessageIsResponse)) {
uint64_t request_id = message->request_id();
ResponderMap::iterator it = responders_.find(request_id);
if (it == responders_.end()) {
assert(false);
return false;
}
MessageReceiver* responder = it->second;
responders_.erase(it);
bool ok = responder->Accept(message);
delete responder;
return ok;
} else {
if (incoming_receiver_)
return incoming_receiver_->Accept(message);
// OK to drop message on the floor.
}
return false;
}
// ----------------------------------------------------------------------------
} // namespace internal
} // namespace mojo
|
