blob: 6d4f23fe2569a7f98679d30bcad947fe49604a20 [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;
380 channel->next_packet_number = 0;
381}
382
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700383ApplicationConnection::ApplicationConnection(
384 ::seasocks::Server *server, ::seasocks::WebSocket *sock,
Alex Perry5f474f22020-02-01 12:14:24 -0800385 const std::vector<std::unique_ptr<Subscriber>> &subscribers,
James Kuszmaul71a81932020-12-15 21:08:01 -0800386 const aos::FlatbufferDetachedBuffer<aos::Configuration> &config,
387 const EventLoop *event_loop)
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700388 : server_(server),
389 sock_(sock),
Alex Perry5f474f22020-02-01 12:14:24 -0800390 subscribers_(subscribers),
James Kuszmaul71a81932020-12-15 21:08:01 -0800391 config_headers_(PackBuffer(config.span())),
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700392 event_loop_(event_loop) {
393 connection_.set_on_negotiation_needed([]() {
394 VLOG(1) << "Negotiation needed, not offering so not creating offer.";
395 });
Alex Perryb3b50792020-01-18 16:13:45 -0800396
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700397 connection_.set_on_local_candidate(
398 [this](struct rawrtc_peer_connection_ice_candidate *const candidate,
399 char const *const url) { LocalCandidate(candidate, url); });
400
401 connection_.set_on_data_channel(
402 [this](std::shared_ptr<ScopedDataChannel> channel) {
403 OnDataChannel(channel);
404 });
405
406 connection_.Open();
407}
408
409ApplicationConnection::~ApplicationConnection() {
410 for (auto &it : channels_) {
411 it.second.data_channel->Close();
412 it.second.data_channel = nullptr;
413 }
414
415 // Eh, we are done, tell the channel to shut down. If we didn't, it would
416 // just hang around until the connection closes, which is rather shortly
417 // after.
418 if (channel_) {
419 channel_->Close();
420 }
421}
422
423void ApplicationConnection::OnSdp(const char *sdp) {
424 struct rawrtc_peer_connection_description *remote_description = NULL;
425
426 auto error = rawrtc_peer_connection_description_create(
427 &remote_description, RAWRTC_SDP_TYPE_OFFER, sdp);
428 if (error) {
429 LOG(WARNING) << "Cannot parse remote description: "
430 << rawrtc_code_to_str(error);
James Kuszmaul48671362020-12-24 13:54:16 -0800431 return;
432 }
Alex Perryb3b50792020-01-18 16:13:45 -0800433
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700434 CHECK_RAWRTC(rawrtc_peer_connection_set_remote_description(
435 connection_.connection(), remote_description));
Alex Perryb3b50792020-01-18 16:13:45 -0800436
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700437 struct rawrtc_peer_connection_description *local_description;
438 CHECK_RAWRTC(rawrtc_peer_connection_create_answer(&local_description,
439 connection_.connection()));
440 CHECK_RAWRTC(rawrtc_peer_connection_set_local_description(
441 connection_.connection(), local_description));
Alex Perryb3b50792020-01-18 16:13:45 -0800442
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700443 enum rawrtc_sdp_type type;
444 char *local_sdp = nullptr;
445 // Get SDP type & the SDP itself
446 CHECK_RAWRTC(rawrtc_peer_connection_description_get_sdp_type(
447 &type, local_description));
448 CHECK_RAWRTC(rawrtc_peer_connection_description_get_sdp(&local_sdp,
449 local_description));
James Kuszmaul8d928d02020-12-25 17:47:49 -0800450
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700451 flatbuffers::FlatBufferBuilder fbb;
Alex Perryb3b50792020-01-18 16:13:45 -0800452 flatbuffers::Offset<WebSocketSdp> sdp_fb =
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700453 CreateWebSocketSdpDirect(fbb, SdpType::ANSWER, local_sdp);
Alex Perryb3b50792020-01-18 16:13:45 -0800454 flatbuffers::Offset<WebSocketMessage> answer_message =
455 CreateWebSocketMessage(fbb, Payload::WebSocketSdp, sdp_fb.Union());
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700456
457 VLOG(1) << aos::FlatbufferToJson(
458 flatbuffers::GetTemporaryPointer(fbb, answer_message));
Alex Perryb3b50792020-01-18 16:13:45 -0800459 fbb.Finish(answer_message);
460
461 server_->execute(std::make_shared<UpdateData>(sock_, fbb.Release()));
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700462 mem_deref(local_sdp);
Alex Perryb3b50792020-01-18 16:13:45 -0800463}
464
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700465void ApplicationConnection::OnIce(const WebSocketIce *ice) {
466 if (!ice->has_candidate()) {
467 return;
468 }
Brian Silverman225c5072021-11-17 19:56:31 -0800469 uint8_t sdp_m_line_index = ice->sdp_m_line_index();
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700470
471 struct rawrtc_peer_connection_ice_candidate *ice_candidate = nullptr;
472 CHECK_RAWRTC(rawrtc_peer_connection_ice_candidate_create(
Brian Silverman225c5072021-11-17 19:56:31 -0800473 &ice_candidate, ice->candidate()->c_str(), ice->sdp_mid()->c_str(),
474 &sdp_m_line_index, nullptr));
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700475
476 rawrtc_peer_connection_add_ice_candidate(connection_.connection(),
477 ice_candidate);
478
479 mem_deref(ice_candidate);
480}
481
482void ApplicationConnection::LocalCandidate(
483 struct rawrtc_peer_connection_ice_candidate *const candidate,
484 char const *const url) {
485 struct rawrtc_ice_candidate *ortc_candidate = nullptr;
486 if (candidate) {
487 CHECK_RAWRTC(rawrtc_peer_connection_ice_candidate_get_ortc_candidate(
488 &ortc_candidate, candidate));
489
490 flatbuffers::FlatBufferBuilder fbb;
491 char *sdpp = nullptr;
492 CHECK_RAWRTC(
493 rawrtc_peer_connection_ice_candidate_get_sdp(&sdpp, candidate));
494 char *midp = nullptr;
495 CHECK_RAWRTC(
496 rawrtc_peer_connection_ice_candidate_get_sdp_mid(&midp, candidate));
497
498 uint8_t media_line_index;
499 enum rawrtc_code error =
500 rawrtc_peer_connection_ice_candidate_get_sdp_media_line_index(
501 &media_line_index, candidate);
502
503 flatbuffers::Offset<flatbuffers::String> sdpp_offset =
504 fbb.CreateString(sdpp);
505 flatbuffers::Offset<flatbuffers::String> sdp_mid_offset =
506 fbb.CreateString(midp);
507
508 WebSocketIce::Builder web_socket_ice_builder(fbb);
509
510 web_socket_ice_builder.add_candidate(sdpp_offset);
Brian Silverman225c5072021-11-17 19:56:31 -0800511 web_socket_ice_builder.add_sdp_mid(sdp_mid_offset);
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700512
513 if (error == RAWRTC_CODE_SUCCESS) {
Brian Silverman225c5072021-11-17 19:56:31 -0800514 web_socket_ice_builder.add_sdp_m_line_index(media_line_index);
James Kuszmaul1ec74432020-07-30 20:26:45 -0700515 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700516 flatbuffers::Offset<WebSocketIce> ice_offset =
517 web_socket_ice_builder.Finish();
518
519 flatbuffers::Offset<WebSocketMessage> ice_message =
520 CreateWebSocketMessage(fbb, Payload::WebSocketIce, ice_offset.Union());
521 VLOG(1) << url << ": "
522 << aos::FlatbufferToJson(
523 flatbuffers::GetTemporaryPointer(fbb, ice_message));
524 fbb.Finish(ice_message);
525
526 server_->execute(std::make_shared<UpdateData>(sock_, fbb.Release()));
527
528 mem_deref(sdpp);
529 mem_deref(midp);
Alex Perry5f474f22020-02-01 12:14:24 -0800530 }
531}
532
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700533void ApplicationConnection::OnDataChannel(
534 std::shared_ptr<ScopedDataChannel> channel) {
535 if (channel->label() == std::string_view("signalling")) {
536 CHECK(!channel_);
537 channel_ = channel;
538
539 channel_->set_on_message(
540 [this](struct mbuf *const buffer,
541 const enum rawrtc_data_channel_message_flag flags) {
542 HandleSignallingData(buffer, flags);
543 });
544
545 channel_->set_on_open([this]() {
546 for (const auto &header : config_headers_) {
547 channel_->Send(header.buffer());
548 }
549 });
550
551 channel_->set_on_error([this]() { LOG(ERROR) << "Error on " << this; });
552
553 // Register an on_close callback which does nothing but keeps channel alive
554 // until it is done. This keeps the memory around until rawrtc can finish
555 // calling the close callback.
556 channel_->set_on_close([channel]() {});
557 } else {
558 channel_->set_on_close([channel]() {});
559 channel->Close();
560 }
561}
562
563void ApplicationConnection::HandleSignallingData(
564 struct mbuf *const
565 buffer, // nullable (in case partial delivery has been requested)
566 const enum rawrtc_data_channel_message_flag /*flags*/) {
James Kuszmaul71a81932020-12-15 21:08:01 -0800567 FlatbufferSpan<SubscriberRequest> message(
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700568 {mbuf_buf(buffer), mbuf_get_left(buffer)});
James Kuszmaul71a81932020-12-15 21:08:01 -0800569 if (!message.Verify()) {
570 LOG(ERROR) << "Invalid flatbuffer received from browser client.";
571 return;
572 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700573 VLOG(1) << "Got a subscription message "
James Kuszmaul71a81932020-12-15 21:08:01 -0800574 << aos::FlatbufferToJson(&message.message());
575 if (!message.message().has_channels_to_transfer()) {
576 LOG(ERROR) << "No channels requested for transfer.";
577 return;
578 }
Alex Perry5f474f22020-02-01 12:14:24 -0800579
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700580 // The client each time sends a full list of everything it wants to be
581 // subscribed to. It is our responsibility to remove channels which aren't
582 // in that list and add ones which need to be.
583 //
584 // Start by clearing a tracking bit on each channel. This takes O(number of
585 // open channels), which should be small.
586 //
587 // Then open any new channels. For any we visit which are already open,
588 // don't update those.
589 //
590 // Finally, iterate over the channel list and purge anything which we didn't
591 // touch.
592 for (auto &it : channels_) {
593 it.second.requested = false;
594 }
595 for (auto channel_request : *message.message().channels_to_transfer()) {
596 const Channel *channel = channel_request->channel();
597 if (channel == nullptr) {
598 LOG(ERROR) << "Got unpopulated channel.";
599 continue;
Alex Perry5f474f22020-02-01 12:14:24 -0800600 }
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700601 const TransferMethod transfer_method = channel_request->method();
602 // Call GetChannel() before comparing the channel name/type to each
603 // subscriber. This allows us to resolve any node or application
604 // specific mappings.
605 const Channel *comparison_channel =
606 configuration::GetChannel(event_loop_->configuration(), channel,
607 event_loop_->name(), event_loop_->node());
608 if (comparison_channel == nullptr) {
James Kuszmaul5e6aa252021-08-28 22:19:29 -0700609 LOG(ERROR) << "Channel does not exist: "
610 << configuration::StrippedChannelToString(channel);
611 continue;
612 }
613 if (!configuration::ChannelIsReadableOnNode(comparison_channel,
614 event_loop_->node())) {
615 LOG(ERROR) << "Channel not available on node "
616 << event_loop_->node()->name()->string_view() << ": "
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700617 << configuration::StrippedChannelToString(channel);
618 continue;
619 }
620
621 size_t channel_index = configuration::ChannelIndex(
622 event_loop_->configuration(), comparison_channel);
623
624 auto it = channels_.find(channel_index);
625 if (it == channels_.end()) {
626 std::shared_ptr<ScopedDataChannel> data_channel =
James Kuszmaula5822682021-12-23 18:39:28 -0800627 ScopedDataChannel::MakeDataChannel();
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700628
629 std::weak_ptr<ScopedDataChannel> data_channel_weak_ptr = data_channel;
630
631 data_channel->set_on_open([this, data_channel_weak_ptr, transfer_method,
632 channel_index]() {
633 std::shared_ptr<ScopedDataChannel> data_channel =
634 data_channel_weak_ptr.lock();
635 CHECK(data_channel) << ": Subscriber got destroyed before we started.";
James Kuszmaula5822682021-12-23 18:39:28 -0800636 // Raw pointer inside the subscriber so we don't have a circular
Austin Schuh52e5e3a2021-04-24 22:30:02 -0700637 // reference. AddListener will close it.
James Kuszmaula5822682021-12-23 18:39:28 -0800638 subscribers_[channel_index]->AddListener(data_channel,
639 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