datachannel: propagate max-message-size changes which will allow Chromium to implement https://w3c.github.io/webrtc-pc/#datachannel-send properly. Bug: chromium:490588131 Change-Id: I79b11a98569e4eec3f39ab24207523bb68ccfb62 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/454580 Reviewed-by: Danil Chapovalov <danilchap@webrtc.org> Reviewed-by: Henrik Boström <hbos@webrtc.org> Reviewed-by: Victor Boivie <boivie@webrtc.org> Commit-Queue: Philipp Hancke <philipp.hancke@googlemail.com> Cr-Commit-Position: refs/heads/main@{#47373}
diff --git a/api/data_channel_interface.h b/api/data_channel_interface.h index 0533136..1ae6494 100644 --- a/api/data_channel_interface.h +++ b/api/data_channel_interface.h
@@ -103,6 +103,7 @@ virtual void OnMessage(const DataBuffer& buffer) = 0; // The data channel's buffered_amount has changed. virtual void OnBufferedAmountChange(uint64_t /* sent_data_size */) {} + virtual void OnMaxMessageSize(int /* max_message_size */) {} // Override this to get callbacks directly on the network thread. // An implementation that does that must not block the network thread
diff --git a/api/transport/data_channel_transport_interface.h b/api/transport/data_channel_transport_interface.h index c99828b..bdf2fd4 100644 --- a/api/transport/data_channel_transport_interface.h +++ b/api/transport/data_channel_transport_interface.h
@@ -95,6 +95,10 @@ // The data channel's buffered_amount has fallen to or below the threshold // set when calling `SetBufferedAmountLowThreshold` virtual void OnBufferedAmountLow(int channel_id) = 0; + + // The data channel's max-message-size has changed as a result of SDP + // negotiation. + virtual void OnMaxMessageSize(int max_message_size) = 0; }; // Transport for data channels.
diff --git a/media/sctp/dcsctp_transport.cc b/media/sctp/dcsctp_transport.cc index 03ffc69..b8f7249 100644 --- a/media/sctp/dcsctp_transport.cc +++ b/media/sctp/dcsctp_transport.cc
@@ -206,7 +206,7 @@ << (options.remote_init.has_value() ? "(set)" : "(not set)") << ")"; - if (!socket_) { + if (socket_ == nullptr) { dcsctp::DcSctpOptions dcsctp_options = CreateDcSctpOptions(options, env_.field_trials()); if (options.local_init.has_value()) { @@ -251,6 +251,9 @@ for (const auto& [sid, stream_state] : stream_states_) { socket_->SetStreamPriority(sid, stream_state.priority); } + if (data_channel_sink_ != nullptr) { + data_channel_sink_->OnMaxMessageSize(options.max_message_size); + } return true; }
diff --git a/media/sctp/dcsctp_transport_unittest.cc b/media/sctp/dcsctp_transport_unittest.cc index bfe797c..5c23a71 100644 --- a/media/sctp/dcsctp_transport_unittest.cc +++ b/media/sctp/dcsctp_transport_unittest.cc
@@ -65,6 +65,7 @@ MOCK_METHOD(void, OnReadyToSend, ()); MOCK_METHOD(void, OnTransportClosed, (RTCError)); MOCK_METHOD(void, OnBufferedAmountLow, (int channel_id), (override)); + MOCK_METHOD(void, OnMaxMessageSize, (int max_message_size), (override)); }; static_assert(!std::is_abstract_v<MockDataChannelSink>); @@ -111,9 +112,9 @@ &dcsctp::DcSctpSocketCallbacks::OnConnected)); EXPECT_CALL(peer_a.sink_, OnReadyToSend); EXPECT_CALL(peer_a.sink_, OnConnected); - peer_a.sctp_transport_->Start({.local_port = 5000, - .remote_port = 5000, - .max_message_size = 256 * 1024}); + EXPECT_CALL(peer_a.sink_, OnMaxMessageSize(32 * 1024)); + peer_a.sctp_transport_->Start( + {.local_port = 5000, .remote_port = 5000, .max_message_size = 32 * 1024}); } // Tests that the close sequence invoked from one end results in the stream to
diff --git a/pc/data_channel_controller.cc b/pc/data_channel_controller.cc index 19c8195..34bce61 100644 --- a/pc/data_channel_controller.cc +++ b/pc/data_channel_controller.cc
@@ -252,6 +252,16 @@ (*it)->OnBufferedAmountLow(); } +void DataChannelController::OnMaxMessageSize(int max_message_size) { + RTC_DCHECK_RUN_ON(network_thread()); + + max_message_size_ = max_message_size; + // Tell all the channels about their new max-message-size. + for (auto& channel : sctp_data_channels_n_) { + channel->OnMaxMessageSize(max_message_size); + } +} + void DataChannelController::SetupDataChannelTransport_n( DataChannelTransportInterface* transport) { RTC_DCHECK_RUN_ON(network_thread()); @@ -406,6 +416,9 @@ } } sctp_data_channels_n_.push_back(channel); + if (max_message_size_.has_value()) { + channel->OnMaxMessageSize(*max_message_size_); + } return channel; }
diff --git a/pc/data_channel_controller.h b/pc/data_channel_controller.h index de9aa7b..910bafb 100644 --- a/pc/data_channel_controller.h +++ b/pc/data_channel_controller.h
@@ -75,6 +75,7 @@ void OnReadyToSend() override; void OnTransportClosed(RTCError error) override; void OnBufferedAmountLow(int channel_id) override; + void OnMaxMessageSize(int max_message_size) override; // Called as part of destroying the owning PeerConnection. void PrepareForShutdown(); @@ -176,6 +177,9 @@ std::unique_ptr<DataChannelEventObserverInterface> event_observer_; + // Cached value of max-message-size. + std::optional<int> max_message_size_ RTC_GUARDED_BY(network_thread()); + // Owning PeerConnection. PeerConnectionInternal* const pc_; // The weak pointers must be dereferenced and invalidated on the network
diff --git a/pc/peer_connection_end_to_end_unittest.cc b/pc/peer_connection_end_to_end_unittest.cc index ee6799c..cf2aa60 100644 --- a/pc/peer_connection_end_to_end_unittest.cc +++ b/pc/peer_connection_end_to_end_unittest.cc
@@ -734,6 +734,35 @@ channels[kReducedMaxSctpStreams / 2]->state()); } +// Verifies that max-message-size is propagated to the DataChannel. +TEST_P(PeerConnectionEndToEndTest, MaxMessageSizePropagated) { + CreatePcs(MockAudioEncoderFactory::CreateEmptyFactory(), + MockAudioDecoderFactory::CreateEmptyFactory()); + + DataChannelInit init; + scoped_refptr<DataChannelInterface> caller_dc( + caller_->CreateDataChannel("caller", init)); + MockDataChannelObserver caller_dc_observer(caller_dc.get()); + + Negotiate(); + WaitForConnection(); + + EXPECT_TRUE(WaitUntil([&] { return caller_dc_observer.IsOpen(); })); + + // Attempting to send a message larger than our announced max-message-size + // should fail. + DataBuffer buffer(CopyOnWriteBuffer(kSctpSendBufferSize + 64), + /*binary=*/true); + EXPECT_FALSE(caller_dc->Send(buffer)); + + scoped_refptr<DataChannelInterface> callee_dc( + callee_->CreateDataChannel("callee", init)); + MockDataChannelObserver callee_dc_observer(callee_dc.get()); + EXPECT_THAT(WaitUntil([&] { return caller_dc_observer.IsOpen(); }, IsTrue()), + IsRtcOk()); + EXPECT_FALSE(callee_dc->Send(buffer)); +} + #endif // WEBRTC_HAVE_SCTP TEST_P(PeerConnectionEndToEndTest, CanRestartIce) {
diff --git a/pc/sctp_data_channel.cc b/pc/sctp_data_channel.cc index 13d01f2..a7338aa 100644 --- a/pc/sctp_data_channel.cc +++ b/pc/sctp_data_channel.cc
@@ -408,6 +408,9 @@ auto register_observer = [me = std::move(me), observer = observer] { RTC_DCHECK_RUN_ON(me->network_thread_); me->observer_ = observer; + if (me->max_message_size_) { + observer->OnMaxMessageSize(*me->max_message_size_); + } me->DeliverQueuedReceivedData(); }; @@ -681,6 +684,15 @@ } } +void SctpDataChannel::OnMaxMessageSize(int max_message_size) { + RTC_DCHECK_RUN_ON(network_thread_); + + max_message_size_ = max_message_size; + if (observer_) { + observer_->OnMaxMessageSize(max_message_size); + } +} + DataChannelStats SctpDataChannel::GetStats() const { RTC_DCHECK_RUN_ON(network_thread_); DataChannelStats stats{.internal_id = internal_id_, @@ -856,6 +868,7 @@ } state_ = state; + if (observer_) { observer_->OnStateChange(); }
diff --git a/pc/sctp_data_channel.h b/pc/sctp_data_channel.h index 540b342..2ac0a3e 100644 --- a/pc/sctp_data_channel.h +++ b/pc/sctp_data_channel.h
@@ -219,6 +219,9 @@ // Called when the amount of data buffered to be sent falls to or below the // threshold set when calling `SetBufferedAmountLowThreshold`. void OnBufferedAmountLow(); + // Called when the data channel's max-message-size has changed as a result + // of SDP negotiation. + void OnMaxMessageSize(int max_message_size); DataChannelStats GetStats() const; @@ -296,6 +299,7 @@ uint64_t bytes_sent_ RTC_GUARDED_BY(network_thread_) = 0; uint32_t messages_received_ RTC_GUARDED_BY(network_thread_) = 0; uint64_t bytes_received_ RTC_GUARDED_BY(network_thread_) = 0; + std::optional<int> max_message_size_ RTC_GUARDED_BY(network_thread_); WeakPtr<SctpDataChannelControllerInterface> controller_ RTC_GUARDED_BY(network_thread_); HandshakeState handshake_state_ RTC_GUARDED_BY(network_thread_) =
diff --git a/pc/test/mock_peer_connection_observers.h b/pc/test/mock_peer_connection_observers.h index ebe2f21..331e4d9 100644 --- a/pc/test/mock_peer_connection_observers.h +++ b/pc/test/mock_peer_connection_observers.h
@@ -479,6 +479,7 @@ ~MockDataChannelObserver() override { channel_->UnregisterObserver(); } void OnBufferedAmountChange(uint64_t previous_amount) override {} + void OnMaxMessageSize(int max_message_size) override {} void OnStateChange() override { states_.push_back(channel_->state());