blob: 30707b2deb48e7d54787f9b9ac34c975d6580355 [file] [log] [blame]
Alex Perryb3b50792020-01-18 16:13:45 -08001#include "aos/network/web_proxy.h"
Alex Perry5f474f22020-02-01 12:14:24 -08002
3#include "aos/flatbuffer_merge.h"
4#include "aos/network/connect_generated.h"
Alex Perryb3b50792020-01-18 16:13:45 -08005#include "aos/network/web_proxy_generated.h"
Alex Perry5f474f22020-02-01 12:14:24 -08006#include "aos/network/web_proxy_utils.h"
James Kuszmaul71a81932020-12-15 21:08:01 -08007#include "aos/seasocks/seasocks_logger.h"
Alex Perryb3b50792020-01-18 16:13:45 -08008#include "glog/logging.h"
James Kuszmaul48671362020-12-24 13:54:16 -08009#include "internal/Embedded.h"
Alex Perryb3b50792020-01-18 16:13:45 -080010
Austin Schuh52e5e3a2021-04-24 22:30:02 -070011extern "C" {
12#include <rawrtc.h>
13
14#define DEBUG_LEVEL 7
15#define DEBUG_MODULE "web-proxy"
16#include <re_dbg.h>
17struct list *tmrl_get(void);
18}
19
Austin Schuhbbc3df32021-10-29 23:06:39 -070020DEFINE_int32(proxy_port, 1180, "Port to use for the web proxy server.");
James Kuszmaula5822682021-12-23 18:39:28 -080021DEFINE_int32(pre_send_messages, 10000,
22 "Number of messages / queue to send to a client before waiting on "
23 "confirmation that the initial message was received. If set to "
24 "-1, will not throttle messages at all. This prevents a situation "
25 "where, when run on localhost, the large number of WebRTC packets "
26 "can overwhelm the browser and crash the webpage.");
James Kuszmaul1e95bed2021-01-09 21:02:49 -080027
Alex Perryb3b50792020-01-18 16:13:45 -080028namespace aos {
29namespace web_proxy {
James Kuszmaul7ad91522020-09-01 19:15:35 -070030WebsocketHandler::WebsocketHandler(::seasocks::Server *server,
James Kuszmaul1a29c082022-02-03 14:02:47 -080031 aos::EventLoop *event_loop,
32 StoreHistory store_history,
33 int per_channel_buffer_size_bytes)
James Kuszmaul7ad91522020-09-01 19:15:35 -070034 : server_(server),
James Kuszmaul71a81932020-12-15 21:08:01 -080035 config_(aos::CopyFlatBuffer(event_loop->configuration())),
36 event_loop_(event_loop) {
Austin Schuh52e5e3a2021-04-24 22:30:02 -070037 if (VLOG_IS_ON(2)) {
38 dbg_init(DBG_DEBUG, DBG_ALL);
39 }
40 CHECK_RAWRTC(rawrtc_init(true));
41
James Kuszmaul48671362020-12-24 13:54:16 -080042 // We need to reference findEmbeddedContent() to make the linker happy...
43 findEmbeddedContent("");
Austin Schuh52e5e3a2021-04-24 22:30:02 -070044 const aos::Node *self = event_loop_->node();
James Kuszmaul7ad91522020-09-01 19:15:35 -070045
Austin Schuh52e5e3a2021-04-24 22:30:02 -070046 subscribers_.reserve(event_loop_->configuration()->channels()->size());
47 for (uint i = 0; i < event_loop_->configuration()->channels()->size(); ++i) {
48 auto channel = event_loop_->configuration()->channels()->Get(i);
James Kuszmaul7ad91522020-09-01 19:15:35 -070049 if (aos::configuration::ChannelIsReadableOnNode(channel, self)) {
Austin Schuh52e5e3a2021-04-24 22:30:02 -070050 auto fetcher = event_loop_->MakeRawFetcher(channel);
James Kuszmaul71a81932020-12-15 21:08:01 -080051 subscribers_.emplace_back(std::make_unique<aos::web_proxy::Subscriber>(
James Kuszmaul1a29c082022-02-03 14:02:47 -080052 std::move(fetcher), i, store_history,
53 per_channel_buffer_size_bytes < 0
54 ? -1
55 : per_channel_buffer_size_bytes / channel->max_size()));
Austin Schuh52e5e3a2021-04-24 22:30:02 -070056 } else {
57 subscribers_.emplace_back(nullptr);
James Kuszmaul7ad91522020-09-01 19:15:35 -070058 }
59 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -070060 TimerHandler *const timer = event_loop_->AddTimer([this]() {
James Kuszmaul7ad91522020-09-01 19:15:35 -070061 for (auto &subscriber : subscribers_) {
Austin Schuh52e5e3a2021-04-24 22:30:02 -070062 if (subscriber) subscriber->RunIteration();
James Kuszmaul7ad91522020-09-01 19:15:35 -070063 }
64 });
65
Austin Schuh52e5e3a2021-04-24 22:30:02 -070066 event_loop_->OnRun([this, timer]() {
67 timer->Setup(event_loop_->monotonic_now(), std::chrono::milliseconds(100));
James Kuszmaul7ad91522020-09-01 19:15:35 -070068 });
69}
Alex Perryb3b50792020-01-18 16:13:45 -080070
71void WebsocketHandler::onConnect(::seasocks::WebSocket *sock) {
Austin Schuh52e5e3a2021-04-24 22:30:02 -070072 std::unique_ptr<ApplicationConnection> connection =
73 std::make_unique<ApplicationConnection>(server_, sock, subscribers_,
74 config_, event_loop_);
75
76 connections_.insert({sock, std::move(connection)});
Alex Perryb3b50792020-01-18 16:13:45 -080077}
78
79void WebsocketHandler::onData(::seasocks::WebSocket *sock, const uint8_t *data,
80 size_t size) {
Austin Schuh52e5e3a2021-04-24 22:30:02 -070081 const FlatbufferSpan<WebSocketMessage> message({data, size});
82 if (!message.Verify()) {
83 LOG(ERROR) << "Invalid WebsocketMessage received from browser.";
84 return;
85 }
86 VLOG(1) << "Got msg " << aos::FlatbufferToJson(message);
87 switch (message.message().payload_type()) {
88 case Payload::WebSocketSdp: {
89 const WebSocketSdp *offer = message.message().payload_as_WebSocketSdp();
90 if (offer->type() != SdpType::OFFER) {
91 LOG(WARNING) << "Got the wrong sdp type from client";
92 break;
93 }
94 const flatbuffers::String *sdp = offer->payload();
95 connections_[sock]->OnSdp(sdp->c_str());
96 break;
97 }
98 case Payload::WebSocketIce: {
99 const WebSocketIce *ice = message.message().payload_as_WebSocketIce();
100 connections_[sock]->OnIce(ice);
101 break;
102 }
Brian Silverman225c5072021-11-17 19:56:31 -0800103 default: {
104 break;
105 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700106 }
Alex Perryb3b50792020-01-18 16:13:45 -0800107}
108
109void WebsocketHandler::onDisconnect(::seasocks::WebSocket *sock) {
110 connections_.erase(sock);
111}
112
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700113// Global epoll pointer
114static aos::internal::EPoll *global_epoll = nullptr;
115
116static int ReFdListen(int fd, int flags, fd_h *fh, void *arg) {
117 if (flags & 0x1) {
118 global_epoll->OnReadable(fd, [fh, arg]() { (*fh)(0x1, arg); });
119 }
120 if (flags & 0x2) {
121 global_epoll->OnWriteable(fd, [fh, arg]() { (*fh)(0x2, arg); });
122 }
123 if (flags & 0x4) {
124 global_epoll->OnError(fd, [fh, arg]() { (*fh)(0x4, arg); });
125 }
126 return 0;
127}
128
129static void ReFdClose(int fd) {
130 CHECK(global_epoll != nullptr);
131 global_epoll->DeleteFd(fd);
132}
133
James Kuszmaul1a29c082022-02-03 14:02:47 -0800134WebProxy::WebProxy(aos::EventLoop *event_loop, StoreHistory store_history,
135 int per_channel_buffer_size_bytes)
136 : WebProxy(event_loop, &internal_epoll_, store_history,
137 per_channel_buffer_size_bytes) {}
James Kuszmaul71a81932020-12-15 21:08:01 -0800138
James Kuszmaul1a29c082022-02-03 14:02:47 -0800139WebProxy::WebProxy(aos::ShmEventLoop *event_loop, StoreHistory store_history,
140 int per_channel_buffer_size_bytes)
141 : WebProxy(event_loop, event_loop->epoll(), store_history,
142 per_channel_buffer_size_bytes) {}
James Kuszmaul71a81932020-12-15 21:08:01 -0800143
144WebProxy::WebProxy(aos::EventLoop *event_loop, aos::internal::EPoll *epoll,
James Kuszmaul1a29c082022-02-03 14:02:47 -0800145 StoreHistory store_history,
146 int per_channel_buffer_size_bytes)
James Kuszmaul71a81932020-12-15 21:08:01 -0800147 : epoll_(epoll),
148 server_(std::make_shared<aos::seasocks::SeasocksLogger>(
149 ::seasocks::Logger::Level::Info)),
James Kuszmaul1a29c082022-02-03 14:02:47 -0800150 websocket_handler_(new WebsocketHandler(
151 &server_, event_loop, store_history, per_channel_buffer_size_bytes)) {
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700152 CHECK(!global_epoll);
153 global_epoll = epoll;
154
155 re_fd_set_listen_callback(&ReFdListen);
156 re_fd_set_close_callback(&ReFdClose);
157
158 epoll->BeforeWait([]() {
159 const uint64_t to = tmr_next_timeout(tmrl_get());
160 if (to != 0) {
Austin Schuh0c8dd362021-10-30 10:23:25 -0700161 VLOG(3) << "Next timeout " << to;
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700162 }
163 // Note: this only works because we are spinning on it...
164 // TODO(austin): If we choose to actually sleep, use a timerfd reserved just
165 // for handling tmr.
166 tmr_poll(tmrl_get());
167 });
168
James Kuszmaul71a81932020-12-15 21:08:01 -0800169 server_.addWebSocketHandler("/ws", websocket_handler_);
James Kuszmaul1e95bed2021-01-09 21:02:49 -0800170 CHECK(server_.startListening(FLAGS_proxy_port));
James Kuszmaul71a81932020-12-15 21:08:01 -0800171
172 epoll->OnReadable(server_.fd(), [this]() {
173 CHECK(::seasocks::Server::PollResult::Continue == server_.poll(0));
174 });
175
176 if (&internal_epoll_ == epoll) {
177 TimerHandler *const timer = event_loop->AddTimer([this]() {
178 // Run the epoll poller until there are no more events (if we are being
179 // backed by a shm event loop, there won't be anything registered to
180 // internal_epoll_ and this will just return false).
181 // We just deal with clearing all the epoll events using a simulated
182 // timer. This does mean that we will spin rather than actually sleeping
183 // in any coherent manner, which will be particularly noticeable when past
184 // the end of processing other events.
185 while (internal_epoll_.Poll(false)) {
186 continue;
187 }
188 });
189
190 event_loop->OnRun([timer, event_loop]() {
191 timer->Setup(event_loop->monotonic_now(), std::chrono::milliseconds(10));
192 });
193 }
194}
195
196WebProxy::~WebProxy() {
197 epoll_->DeleteFd(server_.fd());
198 server_.terminate();
199 CHECK(::seasocks::Server::PollResult::Terminated == server_.poll(0));
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700200 CHECK(global_epoll == epoll_);
201 global_epoll = nullptr;
James Kuszmaul71a81932020-12-15 21:08:01 -0800202}
203
Alex Perry5f474f22020-02-01 12:14:24 -0800204void Subscriber::RunIteration() {
James Kuszmaul1a29c082022-02-03 14:02:47 -0800205 if (channels_.empty() && (buffer_size_ == 0 || !store_history_)) {
206 fetcher_->Fetch();
207 message_buffer_.clear();
Alex Perry5f474f22020-02-01 12:14:24 -0800208 return;
209 }
210
James Kuszmaul1a29c082022-02-03 14:02:47 -0800211
James Kuszmaul71a81932020-12-15 21:08:01 -0800212 while (fetcher_->FetchNext()) {
213 // If we aren't building up a buffer, short-circuit the FetchNext().
214 if (buffer_size_ == 0) {
215 fetcher_->Fetch();
216 }
217 Message message;
218 message.index = fetcher_->context().queue_index;
219 VLOG(2) << "Packing a message with " << GetPacketCount(fetcher_->context())
220 << "packets";
221 for (int packet_index = 0;
222 packet_index < GetPacketCount(fetcher_->context()); ++packet_index) {
Austin Schuhd16ef442021-04-25 14:44:42 -0700223 // Pack directly into the mbuffer. This is admittedly a bit painful.
224 const size_t packet_size =
225 PackedMessageSize(fetcher_->context(), packet_index);
226 struct mbuf *mbuffer = mbuf_alloc(packet_size);
Alex Perry5f474f22020-02-01 12:14:24 -0800227
Austin Schuhd16ef442021-04-25 14:44:42 -0700228 {
229 // Wrap a pre-allocated builder around the mbuffer.
230 PreallocatedAllocator allocator(mbuf_buf(mbuffer), packet_size);
231 flatbuffers::FlatBufferBuilder fbb(packet_size, &allocator);
232 flatbuffers::Offset<MessageHeader> message_offset = PackMessage(
233 &fbb, fetcher_->context(), channel_index_, packet_index);
234 fbb.Finish(message_offset);
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700235
Austin Schuhd16ef442021-04-25 14:44:42 -0700236 // Now, the flatbuffer is built from the back to the front. So any
237 // extra memory will be at the front. Setup the end and start pointers
238 // on the mbuf.
239 mbuf_set_end(mbuffer, packet_size);
240 mbuf_set_pos(mbuffer, packet_size - fbb.GetSize());
241 }
Alex Perry5f474f22020-02-01 12:14:24 -0800242
James Kuszmaul71a81932020-12-15 21:08:01 -0800243 message.data.emplace_back(
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700244 std::shared_ptr<struct mbuf>(mbuffer, mem_deref));
James Kuszmaul71a81932020-12-15 21:08:01 -0800245 }
246 message_buffer_.push_back(std::move(message));
James Kuszmaul45139e62021-09-11 11:41:03 -0700247 // If we aren't keeping a buffer, then we should only do one iteration of
248 // the while loop--otherwise, if additional messages arrive between the
249 // first FetchNext() and the second iteration then we can end up behaving
250 // poorly (since we do a Fetch() when buffer_size_ == 0).
251 if (buffer_size_ == 0) {
252 break;
253 }
James Kuszmaul71a81932020-12-15 21:08:01 -0800254 }
255 for (auto &conn : channels_) {
James Kuszmaula5822682021-12-23 18:39:28 -0800256 std::shared_ptr<ScopedDataChannel> rtc_channel = conn.first.lock();
257 CHECK(rtc_channel) << "data_channel was destroyed too early.";
James Kuszmaul71a81932020-12-15 21:08:01 -0800258 ChannelInformation *channel_data = &conn.second;
259 if (channel_data->transfer_method == TransferMethod::SUBSAMPLE) {
260 SkipToLastMessage(channel_data);
261 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700262 std::shared_ptr<struct mbuf> buffer = NextBuffer(channel_data);
263 while (buffer) {
264 // TODO(austin): This is a nop so we just buffer forever. Fix this when
265 // we care.
James Kuszmaul71a81932020-12-15 21:08:01 -0800266 if (rtc_channel->buffered_amount() > 14000000) {
Alex Perry3dfcb812020-03-04 19:32:17 -0800267 VLOG(1) << "skipping a send because buffered amount is too high";
James Kuszmaul71a81932020-12-15 21:08:01 -0800268 break;
Alex Perry3dfcb812020-03-04 19:32:17 -0800269 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700270
271 rtc_channel->Send(buffer.get());
James Kuszmaul71a81932020-12-15 21:08:01 -0800272 buffer = NextBuffer(channel_data);
273 }
274 }
275 if (buffer_size_ >= 0) {
276 while (message_buffer_.size() > static_cast<size_t>(buffer_size_)) {
277 message_buffer_.pop_front();
Alex Perry5f474f22020-02-01 12:14:24 -0800278 }
279 }
280}
281
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700282void Subscriber::AddListener(std::shared_ptr<ScopedDataChannel> data_channel,
283 TransferMethod transfer_method) {
James Kuszmaul71a81932020-12-15 21:08:01 -0800284 ChannelInformation info;
285 info.transfer_method = transfer_method;
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700286
James Kuszmaula5822682021-12-23 18:39:28 -0800287 channels_.emplace_back(std::make_pair(data_channel, info));
288
289 data_channel->set_on_message(
290 [this, index = channels_.size() - 1](
291 struct mbuf *const buffer,
292 const enum rawrtc_data_channel_message_flag /*flags*/) {
293 FlatbufferSpan<ChannelState> message(
294 {mbuf_buf(buffer), mbuf_get_left(buffer)});
295 if (!message.Verify()) {
296 LOG(ERROR) << "Invalid flatbuffer received from browser client.";
297 return;
298 }
299
300 channels_[index].second.reported_queue_index =
301 message.message().queue_index();
302 channels_[index].second.reported_packet_index =
303 message.message().packet_index();
304 });
James Kuszmaul71a81932020-12-15 21:08:01 -0800305}
306
James Kuszmaula5822682021-12-23 18:39:28 -0800307void Subscriber::RemoveListener(std::shared_ptr<ScopedDataChannel> data_channel) {
308 channels_.erase(
309 std::remove_if(
310 channels_.begin(), channels_.end(),
311 [data_channel](const std::pair<std::weak_ptr<ScopedDataChannel>,
312 ChannelInformation> &channel) {
313 return channel.first.lock().get() == data_channel.get();
314 }),
315 channels_.end());
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700316}
317
318std::shared_ptr<struct mbuf> Subscriber::NextBuffer(
319 ChannelInformation *channel) {
James Kuszmaul71a81932020-12-15 21:08:01 -0800320 CHECK_NOTNULL(channel);
321 if (message_buffer_.empty()) {
322 return nullptr;
323 }
324 const uint32_t earliest_index = message_buffer_.front().index;
325 const uint32_t latest_index = message_buffer_.back().index;
326 const bool fell_behind = channel->current_queue_index < earliest_index;
327 if (fell_behind) {
328 channel->current_queue_index = earliest_index;
329 channel->next_packet_number = 0;
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700330 return message_buffer_.front().data.at(0);
James Kuszmaul71a81932020-12-15 21:08:01 -0800331 }
James Kuszmaula5822682021-12-23 18:39:28 -0800332 // TODO(james): Handle queue index wrapping when >2^32 messages are sent on a
333 // channel.
James Kuszmaul71a81932020-12-15 21:08:01 -0800334 if (channel->current_queue_index > latest_index) {
335 // We are still waiting on the next message to appear; return.
336 return nullptr;
337 }
James Kuszmaula5822682021-12-23 18:39:28 -0800338 if (FLAGS_pre_send_messages > 0) {
339 // Don't buffer up an excessive number of messages to the client.
340 // This currently ignores the packet index (and really, any concept of
341 // message size), but the main goal is just to avoid locking up the client
342 // browser, not to be ultra precise about anything. It's also not clear that
343 // message *size* is necessarily even the determining factor in causing
344 // issues.
345 if (channel->reported_queue_index + FLAGS_pre_send_messages <
346 channel->current_queue_index) {
347 return nullptr;
348 }
349 }
James Kuszmaul71a81932020-12-15 21:08:01 -0800350 CHECK_EQ(latest_index - earliest_index + 1, message_buffer_.size())
351 << "Inconsistent queue indices.";
352 const size_t packets_in_message =
353 message_buffer_[channel->current_queue_index - earliest_index]
354 .data.size();
355 CHECK_LT(0u, packets_in_message);
356 CHECK_LT(channel->next_packet_number, packets_in_message);
357
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700358 std::shared_ptr<struct mbuf> original_data =
359 message_buffer_[channel->current_queue_index - earliest_index].data.at(
James Kuszmaul71a81932020-12-15 21:08:01 -0800360 channel->next_packet_number);
361
362 ++channel->next_packet_number;
363 if (channel->next_packet_number == packets_in_message) {
364 ++channel->current_queue_index;
365 channel->next_packet_number = 0;
366 }
367
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700368 // Trigger a copy of the mbuf without copying the data.
369 return std::shared_ptr<struct mbuf>(mbuf_alloc_ref(original_data.get()),
370 mem_deref);
James Kuszmaul71a81932020-12-15 21:08:01 -0800371}
372
373void Subscriber::SkipToLastMessage(ChannelInformation *channel) {
374 CHECK_NOTNULL(channel);
375 if (message_buffer_.empty() ||
376 channel->current_queue_index == message_buffer_.back().index) {
377 return;
378 }
379 channel->current_queue_index = message_buffer_.back().index;
James Kuszmaul87200a42022-03-26 18:09:18 -0700380 channel->reported_queue_index = message_buffer_.back().index;
James Kuszmaul71a81932020-12-15 21:08:01 -0800381 channel->next_packet_number = 0;
382}
383
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700384ApplicationConnection::ApplicationConnection(
385 ::seasocks::Server *server, ::seasocks::WebSocket *sock,
Alex Perry5f474f22020-02-01 12:14:24 -0800386 const std::vector<std::unique_ptr<Subscriber>> &subscribers,
James Kuszmaul71a81932020-12-15 21:08:01 -0800387 const aos::FlatbufferDetachedBuffer<aos::Configuration> &config,
388 const EventLoop *event_loop)
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700389 : server_(server),
390 sock_(sock),
Alex Perry5f474f22020-02-01 12:14:24 -0800391 subscribers_(subscribers),
James Kuszmaul71a81932020-12-15 21:08:01 -0800392 config_headers_(PackBuffer(config.span())),
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700393 event_loop_(event_loop) {
394 connection_.set_on_negotiation_needed([]() {
395 VLOG(1) << "Negotiation needed, not offering so not creating offer.";
396 });
Alex Perryb3b50792020-01-18 16:13:45 -0800397
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700398 connection_.set_on_local_candidate(
399 [this](struct rawrtc_peer_connection_ice_candidate *const candidate,
400 char const *const url) { LocalCandidate(candidate, url); });
401
402 connection_.set_on_data_channel(
403 [this](std::shared_ptr<ScopedDataChannel> channel) {
404 OnDataChannel(channel);
405 });
406
407 connection_.Open();
408}
409
410ApplicationConnection::~ApplicationConnection() {
411 for (auto &it : channels_) {
412 it.second.data_channel->Close();
413 it.second.data_channel = nullptr;
414 }
415
416 // Eh, we are done, tell the channel to shut down. If we didn't, it would
417 // just hang around until the connection closes, which is rather shortly
418 // after.
419 if (channel_) {
420 channel_->Close();
421 }
422}
423
424void ApplicationConnection::OnSdp(const char *sdp) {
425 struct rawrtc_peer_connection_description *remote_description = NULL;
426
427 auto error = rawrtc_peer_connection_description_create(
428 &remote_description, RAWRTC_SDP_TYPE_OFFER, sdp);
429 if (error) {
430 LOG(WARNING) << "Cannot parse remote description: "
431 << rawrtc_code_to_str(error);
James Kuszmaul48671362020-12-24 13:54:16 -0800432 return;
433 }
Alex Perryb3b50792020-01-18 16:13:45 -0800434
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700435 CHECK_RAWRTC(rawrtc_peer_connection_set_remote_description(
436 connection_.connection(), remote_description));
Alex Perryb3b50792020-01-18 16:13:45 -0800437
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700438 struct rawrtc_peer_connection_description *local_description;
439 CHECK_RAWRTC(rawrtc_peer_connection_create_answer(&local_description,
440 connection_.connection()));
441 CHECK_RAWRTC(rawrtc_peer_connection_set_local_description(
442 connection_.connection(), local_description));
Alex Perryb3b50792020-01-18 16:13:45 -0800443
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700444 enum rawrtc_sdp_type type;
445 char *local_sdp = nullptr;
446 // Get SDP type & the SDP itself
447 CHECK_RAWRTC(rawrtc_peer_connection_description_get_sdp_type(
448 &type, local_description));
449 CHECK_RAWRTC(rawrtc_peer_connection_description_get_sdp(&local_sdp,
450 local_description));
James Kuszmaul8d928d02020-12-25 17:47:49 -0800451
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700452 flatbuffers::FlatBufferBuilder fbb;
Alex Perryb3b50792020-01-18 16:13:45 -0800453 flatbuffers::Offset<WebSocketSdp> sdp_fb =
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700454 CreateWebSocketSdpDirect(fbb, SdpType::ANSWER, local_sdp);
Alex Perryb3b50792020-01-18 16:13:45 -0800455 flatbuffers::Offset<WebSocketMessage> answer_message =
456 CreateWebSocketMessage(fbb, Payload::WebSocketSdp, sdp_fb.Union());
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700457
458 VLOG(1) << aos::FlatbufferToJson(
459 flatbuffers::GetTemporaryPointer(fbb, answer_message));
Alex Perryb3b50792020-01-18 16:13:45 -0800460 fbb.Finish(answer_message);
461
462 server_->execute(std::make_shared<UpdateData>(sock_, fbb.Release()));
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700463 mem_deref(local_sdp);
Alex Perryb3b50792020-01-18 16:13:45 -0800464}
465
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700466void ApplicationConnection::OnIce(const WebSocketIce *ice) {
467 if (!ice->has_candidate()) {
468 return;
469 }
Brian Silverman225c5072021-11-17 19:56:31 -0800470 uint8_t sdp_m_line_index = ice->sdp_m_line_index();
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700471
472 struct rawrtc_peer_connection_ice_candidate *ice_candidate = nullptr;
473 CHECK_RAWRTC(rawrtc_peer_connection_ice_candidate_create(
Brian Silverman225c5072021-11-17 19:56:31 -0800474 &ice_candidate, ice->candidate()->c_str(), ice->sdp_mid()->c_str(),
475 &sdp_m_line_index, nullptr));
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700476
477 rawrtc_peer_connection_add_ice_candidate(connection_.connection(),
478 ice_candidate);
479
480 mem_deref(ice_candidate);
481}
482
483void ApplicationConnection::LocalCandidate(
484 struct rawrtc_peer_connection_ice_candidate *const candidate,
485 char const *const url) {
486 struct rawrtc_ice_candidate *ortc_candidate = nullptr;
487 if (candidate) {
488 CHECK_RAWRTC(rawrtc_peer_connection_ice_candidate_get_ortc_candidate(
489 &ortc_candidate, candidate));
490
491 flatbuffers::FlatBufferBuilder fbb;
492 char *sdpp = nullptr;
493 CHECK_RAWRTC(
494 rawrtc_peer_connection_ice_candidate_get_sdp(&sdpp, candidate));
495 char *midp = nullptr;
496 CHECK_RAWRTC(
497 rawrtc_peer_connection_ice_candidate_get_sdp_mid(&midp, candidate));
498
499 uint8_t media_line_index;
500 enum rawrtc_code error =
501 rawrtc_peer_connection_ice_candidate_get_sdp_media_line_index(
502 &media_line_index, candidate);
503
504 flatbuffers::Offset<flatbuffers::String> sdpp_offset =
505 fbb.CreateString(sdpp);
506 flatbuffers::Offset<flatbuffers::String> sdp_mid_offset =
507 fbb.CreateString(midp);
508
509 WebSocketIce::Builder web_socket_ice_builder(fbb);
510
511 web_socket_ice_builder.add_candidate(sdpp_offset);
Brian Silverman225c5072021-11-17 19:56:31 -0800512 web_socket_ice_builder.add_sdp_mid(sdp_mid_offset);
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700513
514 if (error == RAWRTC_CODE_SUCCESS) {
Brian Silverman225c5072021-11-17 19:56:31 -0800515 web_socket_ice_builder.add_sdp_m_line_index(media_line_index);
James Kuszmaul1ec74432020-07-30 20:26:45 -0700516 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700517 flatbuffers::Offset<WebSocketIce> ice_offset =
518 web_socket_ice_builder.Finish();
519
520 flatbuffers::Offset<WebSocketMessage> ice_message =
521 CreateWebSocketMessage(fbb, Payload::WebSocketIce, ice_offset.Union());
522 VLOG(1) << url << ": "
523 << aos::FlatbufferToJson(
524 flatbuffers::GetTemporaryPointer(fbb, ice_message));
525 fbb.Finish(ice_message);
526
527 server_->execute(std::make_shared<UpdateData>(sock_, fbb.Release()));
528
529 mem_deref(sdpp);
530 mem_deref(midp);
Alex Perry5f474f22020-02-01 12:14:24 -0800531 }
532}
533
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700534void ApplicationConnection::OnDataChannel(
535 std::shared_ptr<ScopedDataChannel> channel) {
536 if (channel->label() == std::string_view("signalling")) {
537 CHECK(!channel_);
538 channel_ = channel;
539
540 channel_->set_on_message(
541 [this](struct mbuf *const buffer,
542 const enum rawrtc_data_channel_message_flag flags) {
543 HandleSignallingData(buffer, flags);
544 });
545
546 channel_->set_on_open([this]() {
547 for (const auto &header : config_headers_) {
548 channel_->Send(header.buffer());
549 }
550 });
551
552 channel_->set_on_error([this]() { LOG(ERROR) << "Error on " << this; });
553
554 // Register an on_close callback which does nothing but keeps channel alive
555 // until it is done. This keeps the memory around until rawrtc can finish
556 // calling the close callback.
557 channel_->set_on_close([channel]() {});
558 } else {
559 channel_->set_on_close([channel]() {});
560 channel->Close();
561 }
562}
563
564void ApplicationConnection::HandleSignallingData(
565 struct mbuf *const
566 buffer, // nullable (in case partial delivery has been requested)
567 const enum rawrtc_data_channel_message_flag /*flags*/) {
James Kuszmaul71a81932020-12-15 21:08:01 -0800568 FlatbufferSpan<SubscriberRequest> message(
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700569 {mbuf_buf(buffer), mbuf_get_left(buffer)});
James Kuszmaul71a81932020-12-15 21:08:01 -0800570 if (!message.Verify()) {
571 LOG(ERROR) << "Invalid flatbuffer received from browser client.";
572 return;
573 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700574 VLOG(1) << "Got a subscription message "
James Kuszmaul71a81932020-12-15 21:08:01 -0800575 << aos::FlatbufferToJson(&message.message());
576 if (!message.message().has_channels_to_transfer()) {
577 LOG(ERROR) << "No channels requested for transfer.";
578 return;
579 }
Alex Perry5f474f22020-02-01 12:14:24 -0800580
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700581 // The client each time sends a full list of everything it wants to be
582 // subscribed to. It is our responsibility to remove channels which aren't
583 // in that list and add ones which need to be.
584 //
585 // Start by clearing a tracking bit on each channel. This takes O(number of
586 // open channels), which should be small.
587 //
588 // Then open any new channels. For any we visit which are already open,
589 // don't update those.
590 //
591 // Finally, iterate over the channel list and purge anything which we didn't
592 // touch.
593 for (auto &it : channels_) {
594 it.second.requested = false;
595 }
596 for (auto channel_request : *message.message().channels_to_transfer()) {
597 const Channel *channel = channel_request->channel();
598 if (channel == nullptr) {
599 LOG(ERROR) << "Got unpopulated channel.";
600 continue;
Alex Perry5f474f22020-02-01 12:14:24 -0800601 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700602 const TransferMethod transfer_method = channel_request->method();
603 // Call GetChannel() before comparing the channel name/type to each
604 // subscriber. This allows us to resolve any node or application
605 // specific mappings.
606 const Channel *comparison_channel =
607 configuration::GetChannel(event_loop_->configuration(), channel,
608 event_loop_->name(), event_loop_->node());
609 if (comparison_channel == nullptr) {
James Kuszmaul5e6aa252021-08-28 22:19:29 -0700610 LOG(ERROR) << "Channel does not exist: "
611 << configuration::StrippedChannelToString(channel);
612 continue;
613 }
614 if (!configuration::ChannelIsReadableOnNode(comparison_channel,
615 event_loop_->node())) {
616 LOG(ERROR) << "Channel not available on node "
617 << event_loop_->node()->name()->string_view() << ": "
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700618 << configuration::StrippedChannelToString(channel);
619 continue;
620 }
621
622 size_t channel_index = configuration::ChannelIndex(
623 event_loop_->configuration(), comparison_channel);
624
625 auto it = channels_.find(channel_index);
626 if (it == channels_.end()) {
627 std::shared_ptr<ScopedDataChannel> data_channel =
James Kuszmaula5822682021-12-23 18:39:28 -0800628 ScopedDataChannel::MakeDataChannel();
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700629
630 std::weak_ptr<ScopedDataChannel> data_channel_weak_ptr = data_channel;
631
632 data_channel->set_on_open([this, data_channel_weak_ptr, transfer_method,
633 channel_index]() {
634 std::shared_ptr<ScopedDataChannel> data_channel =
635 data_channel_weak_ptr.lock();
636 CHECK(data_channel) << ": Subscriber got destroyed before we started.";
James Kuszmaula5822682021-12-23 18:39:28 -0800637 // Raw pointer inside the subscriber so we don't have a circular
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700638 // reference. AddListener will close it.
James Kuszmaula5822682021-12-23 18:39:28 -0800639 subscribers_[channel_index]->AddListener(data_channel,
640 transfer_method);
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700641 });
642
643 Subscriber *subscriber = subscribers_[channel_index].get();
644 data_channel->set_on_close([subscriber, data_channel_weak_ptr]() {
645 std::shared_ptr<ScopedDataChannel> data_channel =
646 data_channel_weak_ptr.lock();
647 CHECK(data_channel) << ": Subscriber got destroyed before we finished.";
648 subscriber->RemoveListener(data_channel);
649 });
650
651 data_channel->Open(
652 connection_.connection(),
653 absl::StrCat(channel->name()->str(), "/", channel->type()->str()));
654
655 auto pair = channels_.insert({channel_index, {data_channel, true}});
656 it = pair.first;
657 }
658
659 it->second.requested = true;
660
661 VLOG(1) << "Subscribe to: " << channel->type()->str();
662 }
663
664 for (auto &it : channels_) {
665 if (!it.second.requested) {
666 it.second.data_channel->Close();
667 it.second.data_channel = nullptr;
Alex Perry5f474f22020-02-01 12:14:24 -0800668 }
669 }
Alex Perryb3b50792020-01-18 16:13:45 -0800670}
671
672} // namespace web_proxy
673} // namespace aos