OLD | NEW |
1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2012 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 "remoting/protocol/channel_multiplexer.h" | 5 #include "remoting/protocol/channel_multiplexer.h" |
6 | 6 |
7 #include <stddef.h> | 7 #include <stddef.h> |
8 #include <string.h> | 8 #include <string.h> |
9 | 9 |
10 #include <utility> | 10 #include <utility> |
(...skipping 12 matching lines...) Expand all Loading... |
23 | 23 |
24 namespace remoting { | 24 namespace remoting { |
25 namespace protocol { | 25 namespace protocol { |
26 | 26 |
27 namespace { | 27 namespace { |
28 const int kChannelIdUnknown = -1; | 28 const int kChannelIdUnknown = -1; |
29 const int kMaxPacketSize = 1024; | 29 const int kMaxPacketSize = 1024; |
30 | 30 |
31 class PendingPacket { | 31 class PendingPacket { |
32 public: | 32 public: |
33 PendingPacket(scoped_ptr<MultiplexPacket> packet, | 33 PendingPacket(scoped_ptr<MultiplexPacket> packet) |
34 const base::Closure& done_task) | 34 : packet(std::move(packet)) {} |
35 : packet(std::move(packet)), | 35 ~PendingPacket() {} |
36 done_task(done_task), | |
37 pos(0U) { | |
38 } | |
39 ~PendingPacket() { | |
40 done_task.Run(); | |
41 } | |
42 | 36 |
43 bool is_empty() { return pos >= packet->data().size(); } | 37 bool is_empty() { return pos >= packet->data().size(); } |
44 | 38 |
45 int Read(char* buffer, size_t size) { | 39 int Read(char* buffer, size_t size) { |
46 size = std::min(size, packet->data().size() - pos); | 40 size = std::min(size, packet->data().size() - pos); |
47 memcpy(buffer, packet->data().data() + pos, size); | 41 memcpy(buffer, packet->data().data() + pos, size); |
48 pos += size; | 42 pos += size; |
49 return size; | 43 return size; |
50 } | 44 } |
51 | 45 |
52 private: | 46 private: |
53 scoped_ptr<MultiplexPacket> packet; | 47 scoped_ptr<MultiplexPacket> packet; |
54 base::Closure done_task; | 48 size_t pos = 0U; |
55 size_t pos; | |
56 | 49 |
57 DISALLOW_COPY_AND_ASSIGN(PendingPacket); | 50 DISALLOW_COPY_AND_ASSIGN(PendingPacket); |
58 }; | 51 }; |
59 | 52 |
60 } // namespace | 53 } // namespace |
61 | 54 |
62 const char ChannelMultiplexer::kMuxChannelName[] = "mux"; | 55 const char ChannelMultiplexer::kMuxChannelName[] = "mux"; |
63 | 56 |
64 struct ChannelMultiplexer::PendingChannel { | 57 struct ChannelMultiplexer::PendingChannel { |
65 PendingChannel(const std::string& name, | 58 PendingChannel(const std::string& name, |
66 const ChannelCreatedCallback& callback) | 59 const ChannelCreatedCallback& callback) |
67 : name(name), callback(callback) { | 60 : name(name), callback(callback) { |
68 } | 61 } |
69 std::string name; | 62 std::string name; |
70 ChannelCreatedCallback callback; | 63 ChannelCreatedCallback callback; |
71 }; | 64 }; |
72 | 65 |
73 class ChannelMultiplexer::MuxChannel { | 66 class ChannelMultiplexer::MuxChannel { |
74 public: | 67 public: |
75 MuxChannel(ChannelMultiplexer* multiplexer, const std::string& name, | 68 MuxChannel(ChannelMultiplexer* multiplexer, const std::string& name, |
76 int send_id); | 69 int send_id); |
77 ~MuxChannel(); | 70 ~MuxChannel(); |
78 | 71 |
79 const std::string& name() { return name_; } | 72 const std::string& name() { return name_; } |
80 int receive_id() { return receive_id_; } | 73 int receive_id() { return receive_id_; } |
81 void set_receive_id(int id) { receive_id_ = id; } | 74 void set_receive_id(int id) { receive_id_ = id; } |
82 | 75 |
83 // Called by ChannelMultiplexer. | 76 // Called by ChannelMultiplexer. |
84 scoped_ptr<P2PStreamSocket> CreateSocket(); | 77 scoped_ptr<P2PStreamSocket> CreateSocket(); |
85 void OnIncomingPacket(scoped_ptr<MultiplexPacket> packet, | 78 void OnIncomingPacket(scoped_ptr<MultiplexPacket> packet); |
86 const base::Closure& done_task); | |
87 void OnBaseChannelError(int error); | 79 void OnBaseChannelError(int error); |
88 | 80 |
89 // Called by MuxSocket. | 81 // Called by MuxSocket. |
90 void OnSocketDestroyed(); | 82 void OnSocketDestroyed(); |
91 void DoWrite(scoped_ptr<MultiplexPacket> packet, | 83 void DoWrite(scoped_ptr<MultiplexPacket> packet, |
92 const base::Closure& done_task); | 84 const base::Closure& done_task); |
93 int DoRead(const scoped_refptr<net::IOBuffer>& buffer, int buffer_len); | 85 int DoRead(const scoped_refptr<net::IOBuffer>& buffer, int buffer_len); |
94 | 86 |
95 private: | 87 private: |
96 ChannelMultiplexer* multiplexer_; | 88 ChannelMultiplexer* multiplexer_; |
(...skipping 60 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
157 } | 149 } |
158 | 150 |
159 scoped_ptr<P2PStreamSocket> ChannelMultiplexer::MuxChannel::CreateSocket() { | 151 scoped_ptr<P2PStreamSocket> ChannelMultiplexer::MuxChannel::CreateSocket() { |
160 DCHECK(!socket_); // Can't create more than one socket per channel. | 152 DCHECK(!socket_); // Can't create more than one socket per channel. |
161 scoped_ptr<MuxSocket> result(new MuxSocket(this)); | 153 scoped_ptr<MuxSocket> result(new MuxSocket(this)); |
162 socket_ = result.get(); | 154 socket_ = result.get(); |
163 return std::move(result); | 155 return std::move(result); |
164 } | 156 } |
165 | 157 |
166 void ChannelMultiplexer::MuxChannel::OnIncomingPacket( | 158 void ChannelMultiplexer::MuxChannel::OnIncomingPacket( |
167 scoped_ptr<MultiplexPacket> packet, | 159 scoped_ptr<MultiplexPacket> packet) { |
168 const base::Closure& done_task) { | |
169 DCHECK_EQ(packet->channel_id(), receive_id_); | 160 DCHECK_EQ(packet->channel_id(), receive_id_); |
170 if (packet->data().size() > 0) { | 161 if (packet->data().size() > 0) { |
171 pending_packets_.push_back(new PendingPacket(std::move(packet), done_task)); | 162 pending_packets_.push_back(new PendingPacket(std::move(packet))); |
172 if (socket_) { | 163 if (socket_) { |
173 // Notify the socket that we have more data. | 164 // Notify the socket that we have more data. |
174 socket_->OnPacketReceived(); | 165 socket_->OnPacketReceived(); |
175 } | 166 } |
176 } | 167 } |
177 } | 168 } |
178 | 169 |
179 void ChannelMultiplexer::MuxChannel::OnBaseChannelError(int error) { | 170 void ChannelMultiplexer::MuxChannel::OnBaseChannelError(int error) { |
180 if (socket_) | 171 if (socket_) |
181 socket_->OnBaseChannelError(error); | 172 socket_->OnBaseChannelError(error); |
(...skipping 244 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
426 } | 417 } |
427 } | 418 } |
428 | 419 |
429 void ChannelMultiplexer::NotifyBaseChannelError(const std::string& name, | 420 void ChannelMultiplexer::NotifyBaseChannelError(const std::string& name, |
430 int error) { | 421 int error) { |
431 std::map<std::string, MuxChannel*>::iterator it = channels_.find(name); | 422 std::map<std::string, MuxChannel*>::iterator it = channels_.find(name); |
432 if (it != channels_.end()) | 423 if (it != channels_.end()) |
433 it->second->OnBaseChannelError(error); | 424 it->second->OnBaseChannelError(error); |
434 } | 425 } |
435 | 426 |
436 void ChannelMultiplexer::OnIncomingPacket(scoped_ptr<MultiplexPacket> packet, | 427 void ChannelMultiplexer::OnIncomingPacket(scoped_ptr<MultiplexPacket> packet) { |
437 const base::Closure& done_task) { | |
438 DCHECK(packet->has_channel_id()); | 428 DCHECK(packet->has_channel_id()); |
439 if (!packet->has_channel_id()) { | 429 if (!packet->has_channel_id()) { |
440 LOG(ERROR) << "Received packet without channel_id."; | 430 LOG(ERROR) << "Received packet without channel_id."; |
441 done_task.Run(); | |
442 return; | 431 return; |
443 } | 432 } |
444 | 433 |
445 int receive_id = packet->channel_id(); | 434 int receive_id = packet->channel_id(); |
446 MuxChannel* channel = nullptr; | 435 MuxChannel* channel = nullptr; |
447 std::map<int, MuxChannel*>::iterator it = | 436 std::map<int, MuxChannel*>::iterator it = |
448 channels_by_receive_id_.find(receive_id); | 437 channels_by_receive_id_.find(receive_id); |
449 if (it != channels_by_receive_id_.end()) { | 438 if (it != channels_by_receive_id_.end()) { |
450 channel = it->second; | 439 channel = it->second; |
451 } else { | 440 } else { |
452 // This is a new |channel_id| we haven't seen before. Look it up by name. | 441 // This is a new |channel_id| we haven't seen before. Look it up by name. |
453 if (!packet->has_channel_name()) { | 442 if (!packet->has_channel_name()) { |
454 LOG(ERROR) << "Received packet with unknown channel_id and " | 443 LOG(ERROR) << "Received packet with unknown channel_id and " |
455 "without channel_name."; | 444 "without channel_name."; |
456 done_task.Run(); | |
457 return; | 445 return; |
458 } | 446 } |
459 channel = GetOrCreateChannel(packet->channel_name()); | 447 channel = GetOrCreateChannel(packet->channel_name()); |
460 channel->set_receive_id(receive_id); | 448 channel->set_receive_id(receive_id); |
461 channels_by_receive_id_[receive_id] = channel; | 449 channels_by_receive_id_[receive_id] = channel; |
462 } | 450 } |
463 | 451 |
464 channel->OnIncomingPacket(std::move(packet), done_task); | 452 channel->OnIncomingPacket(std::move(packet)); |
465 } | 453 } |
466 | 454 |
467 void ChannelMultiplexer::DoWrite(scoped_ptr<MultiplexPacket> packet, | 455 void ChannelMultiplexer::DoWrite(scoped_ptr<MultiplexPacket> packet, |
468 const base::Closure& done_task) { | 456 const base::Closure& done_task) { |
469 writer_.Write(SerializeAndFrameMessage(*packet), done_task); | 457 writer_.Write(SerializeAndFrameMessage(*packet), done_task); |
470 } | 458 } |
471 | 459 |
472 } // namespace protocol | 460 } // namespace protocol |
473 } // namespace remoting | 461 } // namespace remoting |
OLD | NEW |