blob: 7161ee327fc5efadfc92da245f2240631ca1a1e4 [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 Kuszmaul71a81932020-12-15 21:08:01 -0800211 while (fetcher_->FetchNext()) {
212 // If we aren't building up a buffer, short-circuit the FetchNext().
213 if (buffer_size_ == 0) {
214 fetcher_->Fetch();
215 }
216 Message message;
217 message.index = fetcher_->context().queue_index;
218 VLOG(2) << "Packing a message with " << GetPacketCount(fetcher_->context())
219 << "packets";
220 for (int packet_index = 0;
221 packet_index < GetPacketCount(fetcher_->context()); ++packet_index) {
Austin Schuhd16ef442021-04-25 14:44:42 -0700222 // Pack directly into the mbuffer. This is admittedly a bit painful.
223 const size_t packet_size =
224 PackedMessageSize(fetcher_->context(), packet_index);
225 struct mbuf *mbuffer = mbuf_alloc(packet_size);
Alex Perry5f474f22020-02-01 12:14:24 -0800226
Austin Schuhd16ef442021-04-25 14:44:42 -0700227 {
228 // Wrap a pre-allocated builder around the mbuffer.
229 PreallocatedAllocator allocator(mbuf_buf(mbuffer), packet_size);
230 flatbuffers::FlatBufferBuilder fbb(packet_size, &allocator);
231 flatbuffers::Offset<MessageHeader> message_offset = PackMessage(
232 &fbb, fetcher_->context(), channel_index_, packet_index);
233 fbb.Finish(message_offset);
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700234
Austin Schuhd16ef442021-04-25 14:44:42 -0700235 // Now, the flatbuffer is built from the back to the front. So any
236 // extra memory will be at the front. Setup the end and start pointers
237 // on the mbuf.
238 mbuf_set_end(mbuffer, packet_size);
239 mbuf_set_pos(mbuffer, packet_size - fbb.GetSize());
240 }
Alex Perry5f474f22020-02-01 12:14:24 -0800241
James Kuszmaul71a81932020-12-15 21:08:01 -0800242 message.data.emplace_back(
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700243 std::shared_ptr<struct mbuf>(mbuffer, mem_deref));
James Kuszmaul71a81932020-12-15 21:08:01 -0800244 }
245 message_buffer_.push_back(std::move(message));
James Kuszmaul45139e62021-09-11 11:41:03 -0700246 // If we aren't keeping a buffer, then we should only do one iteration of
247 // the while loop--otherwise, if additional messages arrive between the
248 // first FetchNext() and the second iteration then we can end up behaving
249 // poorly (since we do a Fetch() when buffer_size_ == 0).
250 if (buffer_size_ == 0) {
251 break;
252 }
James Kuszmaul71a81932020-12-15 21:08:01 -0800253 }
254 for (auto &conn : channels_) {
James Kuszmaula5822682021-12-23 18:39:28 -0800255 std::shared_ptr<ScopedDataChannel> rtc_channel = conn.first.lock();
256 CHECK(rtc_channel) << "data_channel was destroyed too early.";
James Kuszmaul71a81932020-12-15 21:08:01 -0800257 ChannelInformation *channel_data = &conn.second;
258 if (channel_data->transfer_method == TransferMethod::SUBSAMPLE) {
259 SkipToLastMessage(channel_data);
260 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700261 std::shared_ptr<struct mbuf> buffer = NextBuffer(channel_data);
262 while (buffer) {
263 // TODO(austin): This is a nop so we just buffer forever. Fix this when
264 // we care.
James Kuszmaul71a81932020-12-15 21:08:01 -0800265 if (rtc_channel->buffered_amount() > 14000000) {
Alex Perry3dfcb812020-03-04 19:32:17 -0800266 VLOG(1) << "skipping a send because buffered amount is too high";
James Kuszmaul71a81932020-12-15 21:08:01 -0800267 break;
Alex Perry3dfcb812020-03-04 19:32:17 -0800268 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700269
270 rtc_channel->Send(buffer.get());
James Kuszmaul71a81932020-12-15 21:08:01 -0800271 buffer = NextBuffer(channel_data);
272 }
273 }
274 if (buffer_size_ >= 0) {
275 while (message_buffer_.size() > static_cast<size_t>(buffer_size_)) {
276 message_buffer_.pop_front();
Alex Perry5f474f22020-02-01 12:14:24 -0800277 }
278 }
279}
280
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700281void Subscriber::AddListener(std::shared_ptr<ScopedDataChannel> data_channel,
282 TransferMethod transfer_method) {
James Kuszmaul71a81932020-12-15 21:08:01 -0800283 ChannelInformation info;
284 info.transfer_method = transfer_method;
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700285
James Kuszmaula5822682021-12-23 18:39:28 -0800286 channels_.emplace_back(std::make_pair(data_channel, info));
287
288 data_channel->set_on_message(
289 [this, index = channels_.size() - 1](
290 struct mbuf *const buffer,
291 const enum rawrtc_data_channel_message_flag /*flags*/) {
292 FlatbufferSpan<ChannelState> message(
293 {mbuf_buf(buffer), mbuf_get_left(buffer)});
294 if (!message.Verify()) {
295 LOG(ERROR) << "Invalid flatbuffer received from browser client.";
296 return;
297 }
298
299 channels_[index].second.reported_queue_index =
300 message.message().queue_index();
301 channels_[index].second.reported_packet_index =
302 message.message().packet_index();
303 });
James Kuszmaul71a81932020-12-15 21:08:01 -0800304}
305
Austin Schuh60e77942022-05-16 17:48:24 -0700306void Subscriber::RemoveListener(
307 std::shared_ptr<ScopedDataChannel> data_channel) {
James Kuszmaula5822682021-12-23 18:39:28 -0800308 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.
Austin Schuh60e77942022-05-16 17:48:24 -0700639 subscribers_[channel_index]->AddListener(data_channel, transfer_method);
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700640 });
641
642 Subscriber *subscriber = subscribers_[channel_index].get();
643 data_channel->set_on_close([subscriber, data_channel_weak_ptr]() {
644 std::shared_ptr<ScopedDataChannel> data_channel =
645 data_channel_weak_ptr.lock();
646 CHECK(data_channel) << ": Subscriber got destroyed before we finished.";
647 subscriber->RemoveListener(data_channel);
648 });
649
650 data_channel->Open(
651 connection_.connection(),
652 absl::StrCat(channel->name()->str(), "/", channel->type()->str()));
653
654 auto pair = channels_.insert({channel_index, {data_channel, true}});
655 it = pair.first;
656 }
657
658 it->second.requested = true;
659
660 VLOG(1) << "Subscribe to: " << channel->type()->str();
661 }
662
663 for (auto &it : channels_) {
664 if (!it.second.requested) {
665 it.second.data_channel->Close();
666 it.second.data_channel = nullptr;
Alex Perry5f474f22020-02-01 12:14:24 -0800667 }
668 }
Alex Perryb3b50792020-01-18 16:13:45 -0800669}
670
671} // namespace web_proxy
672} // namespace aos