Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[flatbuffers] Add pause() and resume() in DataProducer and DataConsumer #1104

Merged
merged 9 commits into from
Jun 22, 2023
Merged

[flatbuffers] Add pause() and resume() in DataProducer and DataConsumer #1104

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
bdee272
[flatbuffers] Add pause() and resume() to DataProducer and DataConsumer
ibc Jun 21, 2023
01cf2a8
more
ibc Jun 21, 2023
21f2d7e
more
ibc Jun 21, 2023
e6d708c
done but tests missing
ibc Jun 21, 2023
c69b8b8
revert a rename
ibc Jun 21, 2023
a1ccc16
fix node lint
ibc Jun 21, 2023
0d657c8
Node and C++ done and tested
ibc Jun 21, 2023
cbbc115
Rust mostly done
ibc Jun 21, 2023
7ae8328
Rust done (tests included)
ibc Jun 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion worker/fbs/notification.fbs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ enum Event: uint8 {
// Notifications to worker.
TRANSPORT_SEND_RTCP = 0,
PRODUCER_SEND,
DATA_PRODUCER_SEND,
DATAPRODUCER_SEND,
// Notifications from worker.
WORKER_RUNNING,
TRANSPORT_SCTP_STATE_CHANGE,
Expand All @@ -37,6 +37,8 @@ enum Event: uint8 {
CONSUMER_TRACE,
DATACONSUMER_BUFFERED_AMOUNT_LOW,
DATACONSUMER_SCTP_SENDBUFFER_FULL,
DATACONSUMER_DATAPRODUCER_PAUSE,
DATACONSUMER_DATAPRODUCER_RESUME,
DATACONSUMER_DATAPRODUCER_CLOSE,
DATACONSUMER_MESSAGE,
ACTIVESPEAKEROBSERVER_DOMINANT_SPEAKER,
Expand Down
54 changes: 29 additions & 25 deletions worker/fbs/request.fbs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,19 +15,19 @@ enum Method: uint8 {
WORKER_UPDATE_SETTINGS,
WORKER_CREATE_WEBRTC_SERVER,
WORKER_CREATE_ROUTER,
WORKER_WEBRTC_SERVER_CLOSE,
WORKER_WEBRTCSERVER_CLOSE,
WORKER_CLOSE_ROUTER,
WEBRTC_SERVER_DUMP,
ROUTER_DUMP,
ROUTER_CREATE_WEBRTC_TRANSPORT,
ROUTER_CREATE_WEBRTC_TRANSPORT_WITH_SERVER,
ROUTER_CREATE_PLAIN_TRANSPORT,
ROUTER_CREATE_PIPE_TRANSPORT,
ROUTER_CREATE_DIRECT_TRANSPORT,
ROUTER_CREATE_WEBRTCTRANSPORT,
ROUTER_CREATE_WEBRTCTRANSPORT_WITH_SERVER,
ROUTER_CREATE_PLAINTRANSPORT,
ROUTER_CREATE_PIPETRANSPORT,
ROUTER_CREATE_DIRECTTRANSPORT,
ROUTER_CLOSE_TRANSPORT,
ROUTER_CREATE_ACTIVE_SPEAKER_OBSERVER,
ROUTER_CREATE_AUDIO_LEVEL_OBSERVER,
ROUTER_CLOSE_RTP_OBSERVER,
ROUTER_CREATE_ACTIVESPEAKEROBSERVER,
ROUTER_CREATE_AUDIOLEVELOBSERVER,
ROUTER_CLOSE_RTPOBSERVER,
TRANSPORT_DUMP,
TRANSPORT_GET_STATS,
TRANSPORT_CONNECT,
Expand All @@ -42,11 +42,11 @@ enum Method: uint8 {
TRANSPORT_ENABLE_TRACE_EVENT,
TRANSPORT_CLOSE_PRODUCER,
TRANSPORT_CLOSE_CONSUMER,
TRANSPORT_CLOSE_DATA_PRODUCER,
TRANSPORT_CLOSE_DATA_CONSUMER,
PLAIN_TRANSPORT_CONNECT,
PIPE_TRANSPORT_CONNECT,
WEBRTC_TRANSPORT_CONNECT,
TRANSPORT_CLOSE_DATAPRODUCER,
TRANSPORT_CLOSE_DATACONSUMER,
PLAINTRANSPORT_CONNECT,
PIPETRANSPORT_CONNECT,
WEBRTCTRANSPORT_CONNECT,
PRODUCER_DUMP,
PRODUCER_GET_STATS,
PRODUCER_PAUSE,
Expand All @@ -60,17 +60,21 @@ enum Method: uint8 {
CONSUMER_SET_PRIORITY,
CONSUMER_REQUEST_KEY_FRAME,
CONSUMER_ENABLE_TRACE_EVENT,
DATA_PRODUCER_DUMP,
DATA_PRODUCER_GET_STATS,
DATA_CONSUMER_DUMP,
DATA_CONSUMER_GET_STATS,
DATA_CONSUMER_GET_BUFFERED_AMOUNT,
DATA_CONSUMER_SET_BUFFERED_AMOUNT_LOW_THRESHOLD,
DATA_CONSUMER_SEND,
RTP_OBSERVER_PAUSE,
RTP_OBSERVER_RESUME,
RTP_OBSERVER_ADD_PRODUCER,
RTP_OBSERVER_REMOVE_PRODUCER,
DATAPRODUCER_DUMP,
DATAPRODUCER_GET_STATS,
DATAPRODUCER_PAUSE,
DATAPRODUCER_RESUME,
DATACONSUMER_DUMP,
DATACONSUMER_GET_STATS,
DATACONSUMER_PAUSE,
DATACONSUMER_RESUME,
DATACONSUMER_GET_BUFFERED_AMOUNT,
DATACONSUMER_SET_BUFFERED_AMOUNT_LOW_THRESHOLD,
DATACONSUMER_SEND,
RTPOBSERVER_PAUSE,
RTPOBSERVER_RESUME,
RTPOBSERVER_ADD_PRODUCER,
RTPOBSERVER_REMOVE_PRODUCER,
}

union Body {
Expand Down
12 changes: 12 additions & 0 deletions worker/include/RTC/DataConsumer.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,16 @@ namespace RTC
}
void TransportConnected();
void TransportDisconnected();
bool IsPaused() const
{
return this->paused;
}
bool IsDataProducerPaused() const
{
return this->dataProducerPaused;
}
void DataProducerPaused();
void DataProducerResumed();
void SctpAssociationConnected();
void SctpAssociationClosed();
void SctpAssociationBufferedAmount(uint32_t bufferedAmount);
Expand Down Expand Up @@ -108,6 +118,8 @@ namespace RTC
std::string protocol;
bool transportConnected{ false };
bool sctpAssociationConnected{ false };
bool paused{ false };
bool dataProducerPaused{ false };
bool dataProducerClosed{ false };
size_t messagesSent{ 0u };
size_t bytesSent{ 0u };
Expand Down
7 changes: 7 additions & 0 deletions worker/include/RTC/DataProducer.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ namespace RTC
virtual void OnDataProducerReceiveData(RTC::DataProducer* producer, size_t len) = 0;
virtual void OnDataProducerMessageReceived(
RTC::DataProducer* dataProducer, uint32_t ppid, const uint8_t* msg, size_t len) = 0;
virtual void OnDataProducerPaused(RTC::DataProducer* dataProducer) = 0;
virtual void OnDataProducerResumed(RTC::DataProducer* dataProducer) = 0;
};

public:
Expand Down Expand Up @@ -55,6 +57,10 @@ namespace RTC
{
return this->sctpStreamParameters;
}
bool IsPaused() const
{
return this->paused;
}
void ReceiveMessage(uint32_t ppid, const uint8_t* msg, size_t len);

/* Methods inherited from Channel::ChannelSocket::RequestHandler. */
Expand All @@ -80,6 +86,7 @@ namespace RTC
RTC::SctpStreamParameters sctpStreamParameters;
std::string label;
std::string protocol;
bool paused{ false };
size_t messagesReceived{ 0u };
size_t bytesReceived{ 0u };
};
Expand Down
4 changes: 2 additions & 2 deletions worker/include/RTC/PipeConsumer.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
#ifndef MS_RTC_PIPE_CONSUMER_HPP
#define MS_RTC_PIPE_CONSUMER_HPP
#ifndef MS_RTC_PIPECONSUMER_HPP
#define MS_RTC_PIPECONSUMER_HPP

#include "RTC/Consumer.hpp"
#include "RTC/RtpStreamSend.hpp"
Expand Down
2 changes: 2 additions & 0 deletions worker/include/RTC/Router.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,8 @@ namespace RTC
RTC::Transport* transport, RTC::Consumer* consumer, uint32_t mappedSsrc) override;
void OnTransportNewDataProducer(RTC::Transport* transport, RTC::DataProducer* dataProducer) override;
void OnTransportDataProducerClosed(RTC::Transport* transport, RTC::DataProducer* dataProducer) override;
void OnTransportDataProducerPaused(RTC::Transport* transport, RTC::DataProducer* dataProducer) override;
void OnTransportDataProducerResumed(RTC::Transport* transport, RTC::DataProducer* dataProducer) override;
void OnTransportDataProducerMessageReceived(
RTC::Transport* transport,
RTC::DataProducer* dataProducer,
Expand Down
4 changes: 2 additions & 2 deletions worker/include/RTC/RtpObserver.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
#ifndef MS_RTC_RTP_PACKET_OBSERVER_HPP
#define MS_RTC_RTP_PACKET_OBSERVER_HPP
#ifndef MS_RTC_RTP_OBSERVER_HPP
#define MS_RTC_RTP_OBSERVER_HPP

#include "common.hpp"
#include "RTC/Producer.hpp"
Expand Down
5 changes: 5 additions & 0 deletions worker/include/RTC/Transport.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,8 @@ namespace RTC
virtual void OnTransportConsumerClosed(RTC::Transport* transport, RTC::Consumer* consumer) = 0;
virtual void OnTransportConsumerProducerClosed(
RTC::Transport* transport, RTC::Consumer* consumer) = 0;
virtual void OnTransportDataProducerPaused(RTC::Transport* transport, RTC::DataProducer* dataProducer) = 0;
virtual void OnTransportDataProducerResumed(RTC::Transport* transport, RTC::DataProducer* dataProducer) = 0;
virtual void OnTransportConsumerKeyFrameRequested(
RTC::Transport* transport, RTC::Consumer* consumer, uint32_t mappedSsrc) = 0;
virtual void OnTransportNewDataProducer(
Expand Down Expand Up @@ -235,6 +237,9 @@ namespace RTC
}
void OnDataProducerMessageReceived(
RTC::DataProducer* dataProducer, uint32_t ppid, const uint8_t* msg, size_t len) override;
void OnDataProducerPaused(RTC::DataProducer* dataProducer) override;
void OnDataProducerResumed(RTC::DataProducer* dataProducer) override;


/* Pure virtual methods inherited from RTC::DataConsumer::Listener. */
public:
Expand Down
4 changes: 2 additions & 2 deletions worker/src/Channel/ChannelNotification.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ namespace Channel
absl::flat_hash_map<FBS::Notification::Event, const char*> ChannelNotification::event2String =
{
{ FBS::Notification::Event::TRANSPORT_SEND_RTCP, "transport.sendRtcp" },
{ FBS::Notification::Event::PRODUCER_SEND, "producer.send" },
{ FBS::Notification::Event::DATA_PRODUCER_SEND, "dataProducer.send" },
{ FBS::Notification::Event::PRODUCER_SEND, "producer.send" },
{ FBS::Notification::Event::DATAPRODUCER_SEND, "dataProducer.send" },
};
// clang-format on

Expand Down
Loading