webrtc_wire_stream.h

Go to the documentation of this file.

/*
 * Copyright 2026 The A11 Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef A11_NET_WEBRTC_WIRE_STREAM_H_
#define A11_NET_WEBRTC_WIRE_STREAM_H_
 
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
 
#include <absl/base/thread_annotations.h>
#include <absl/status/status.h>
#include <absl/status/statusor.h>
#include <absl/time/time.h>
 
#include "a11/concurrency/future.h"
#include "a11/net/channel_wire_stream.h"
#include "a11/net/signalling.h"
#include "a11/net/wire_stream.h"
#include "thread/boost_primitives.h"
 
namespace rtc {
class DataChannel;
class PeerConnection;
}  // namespace rtc
 
namespace a11::net {
 
namespace internal {
class MultiplexedBinaryChannel;
class PathMtuDiscovery;
}  // namespace internal
 
enum class TurnRelayType { kUdp, kTcp, kTls };
 
inline constexpr size_t kWebRtcBaseMtu = 1280;
 
inline constexpr size_t kWebRtcMinMtu = 512;
 
struct TurnServer {
  static absl::StatusOr<TurnServer> FromString(std::string_view value);
 
  std::string hostname;       
  std::uint16_t port = 3478;  
  std::string username;       
  std::string password;       
  TurnRelayType relay_type = TurnRelayType::kUdp;  
 
  friend bool operator==(const TurnServer&, const TurnServer&) = default;
};
 
struct WebRtcConfiguration {
  // Large logical WireMessages are fragmented by A11 before reaching SCTP;
  // this remains the advertised local libdatachannel message ceiling.
  std::optional<size_t> max_message_size =
      64 * 1024;  
  size_t channel_split_size =
      48 * 1024;  
  std::optional<size_t> mtu;
  bool enable_ice_udp_mux =
      false;  
  std::vector<std::string> stun_servers;  
  std::vector<TurnServer> turn_servers;   
  std::optional<std::pair<std::uint16_t, std::uint16_t>> preferred_port_range;
  std::optional<std::string>
      bind_address;  
  // Stripe packets across data channels while preserving one reliable
  // WireStream. Clients maintain `desired_channels`; servers admit at most
  // `max_channels` per peer.
  size_t desired_channels =
      8;                    
  size_t max_channels = 8;  
 
  bool path_mtu_discovery = true;
  size_t max_discovered_mtu = 9216;
  absl::Duration probe_timeout = absl::Milliseconds(500);
  absl::Duration path_mtu_raise_interval = absl::Seconds(600);
  absl::Duration path_mtu_startup_retry = absl::Milliseconds(250);
 
  absl::Status Validate() const;
};
 
class WebRtcWireStream final : public ChannelWireStream {
 private:
  struct ConstructorToken {};
 
 public:
  static absl::StatusOr<std::shared_ptr<WebRtcWireStream>> CreateClient(
      std::string identity, std::string peer_identity,
      const std::shared_ptr<SignallingService>& signalling,
      const WebRtcConfiguration& configuration = {},
      WireStreamOptions options = {});
 
  static absl::StatusOr<std::shared_ptr<WebRtcWireStream>> CreateClient(
      std::string peer_identity,
      const std::shared_ptr<SignallingTransport>& signalling,
      const WebRtcConfiguration& configuration = {},
      WireStreamOptions options = {});
 
  static absl::StatusOr<std::shared_ptr<WebRtcWireStream>> Create(
      std::shared_ptr<rtc::DataChannel> data_channel,
      std::shared_ptr<rtc::PeerConnection> connection,
      std::shared_ptr<SignallingTransport> signalling_endpoint,
      ChannelEndpointRole role, WireStreamOptions options = {},
      OpenOperation open_operation = {}, size_t split_size = 48 * 1024);
 
  static absl::StatusOr<std::shared_ptr<WebRtcWireStream>>
  BuildMultiplexedStream(
      std::shared_ptr<internal::BinaryChannel> channel,
      std::shared_ptr<rtc::DataChannel> primary_channel,
      std::shared_ptr<rtc::PeerConnection> connection,
      std::shared_ptr<SignallingTransport> signalling_endpoint, std::string id,
      ChannelEndpointRole role, WireStreamOptions options = {},
      OpenOperation open_operation = {}, size_t split_size = 48 * 1024,
      size_t configured_mtu = kWebRtcBaseMtu);
 
  ~WebRtcWireStream() override;
 
  [[nodiscard]] std::shared_ptr<rtc::DataChannel> data_channel() const;
  [[nodiscard]] std::shared_ptr<rtc::PeerConnection> peer_connection() const;
  [[nodiscard]] std::shared_ptr<SignallingTransport> signalling_endpoint()
      const;
 
  absl::Status SetPathMtu(size_t mtu);
 
  [[nodiscard]] size_t current_path_mtu() const;
 
  void StartPathMtuDiscovery(
      const WebRtcConfiguration& configuration,
      const std::shared_ptr<internal::MultiplexedBinaryChannel>& stream_data);
 
  [[nodiscard]] size_t discovered_path_mtu() const;
 
  WebRtcWireStream(ConstructorToken, std::shared_ptr<State> state,
                   std::shared_ptr<rtc::DataChannel> data_channel,
                   std::shared_ptr<rtc::PeerConnection> connection,
                   std::shared_ptr<SignallingTransport> signalling_endpoint,
                   size_t configured_mtu)
      : ChannelWireStream(std::move(state)),
        data_channel_(std::move(data_channel)),
        connection_(std::move(connection)),
        signalling_endpoint_(std::move(signalling_endpoint)),
        path_mtu_(configured_mtu) {}
 
 private:
  std::shared_ptr<rtc::DataChannel> data_channel_;
  std::shared_ptr<rtc::PeerConnection> connection_;
  std::shared_ptr<SignallingTransport> signalling_endpoint_;
  mutable thread::Mutex path_mtu_mu_;
  size_t path_mtu_ ABSL_GUARDED_BY(path_mtu_mu_);
  std::shared_ptr<internal::PathMtuDiscovery> discovery_
      ABSL_GUARDED_BY(path_mtu_mu_);
};
 
using OnWebRtcStream =
    std::function<a11::Task(std::shared_ptr<WebRtcWireStream>)>;
 
class WebRtcWireServer : public std::enable_shared_from_this<WebRtcWireServer> {
 public:
  static absl::StatusOr<std::shared_ptr<WebRtcWireServer>> Create(
      std::string identity,
      const std::shared_ptr<SignallingService>& signalling,
      OnWebRtcStream on_stream, WebRtcConfiguration configuration = {},
      WireStreamOptions stream_options = {});
 
  static absl::StatusOr<std::shared_ptr<WebRtcWireServer>> Create(
      const std::shared_ptr<SignallingTransport>& signalling,
      OnWebRtcStream on_stream, WebRtcConfiguration configuration = {},
      WireStreamOptions stream_options = {});
 
  ~WebRtcWireServer();
 
  absl::Status Stop();
  [[nodiscard]] std::string identity() const;
  [[nodiscard]] bool running() const;
  [[nodiscard]] size_t pending_peer_count() const;
  [[nodiscard]] std::shared_ptr<SignallingTransport> signalling_endpoint()
      const;
 
 private:
  struct State;
 
  explicit WebRtcWireServer(std::shared_ptr<State> state)
      : state_(std::move(state)) {}
 
  static a11::Task OnSignal(const std::shared_ptr<State>& state,
                            const SignallingMessage& message);
  static absl::Status HandleOffer(const std::shared_ptr<State>& state,
                                  const SignallingMessage& message);
  static absl::Status HandleCandidate(const std::shared_ptr<State>& state,
                                      const SignallingMessage& message);
  static void ReportPeerError(const std::shared_ptr<State>& state,
                              std::string peer_identity, absl::Status status);
 
  std::shared_ptr<State> state_;
};
 
}  // namespace a11::net
 
#endif  // A11_NET_WEBRTC_WIRE_STREAM_H_