Skip to content

Commit 4f77c77

Browse files
UriagatUriagat
authored
Feat/data track (#399)
* feat: dataTrack * feat: data Track Api update * fix: streamApi clanup * fix: post review changes
1 parent 7b08ce7 commit 4f77c77

23 files changed

Lines changed: 449 additions & 49 deletions

‎endpoint/programs/stream_testing/single_video_receiver.cpp‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,7 @@ class OnTrackImpl : public stream::OnTrackInterface {
150150
}
151151
}
152152
virtual void OnData(std::shared_ptr<stream::Data> data) override {
153+
153154
if(data->type == stream::DataType::VIDEO) {
154155
auto videoData = std::dynamic_pointer_cast<stream::VideoData>(data);
155156
// selecting most active video track to render
@@ -162,9 +163,13 @@ class OnTrackImpl : public stream::OnTrackInterface {
162163
_renderer.OnFrame(videoData->w, videoData->h, videoData->frameData);
163164
}
164165
--_videoTrackC;
165-
}
166-
if(data->type == stream::DataType::AUDIO) {
166+
} else if(data->type == stream::DataType::AUDIO) {
167167
auto audioData = std::dynamic_pointer_cast<stream::AudioData>(data);
168+
} else if(data->type == stream::DataType::PLAIN) {
169+
auto plainData = std::dynamic_pointer_cast<stream::PlainData>(data);
170+
LOG_INFO("Recived plain data", plainData->data.stdString());
171+
} else {
172+
LOG_FATAL("DataType::UNKNOWN")
168173
}
169174
}
170175
private:
@@ -231,6 +236,7 @@ int main(int argc, char** argv) {
231236
std::cout << "stream.metadata:" << (stream.metadata.has_value() ? stream.metadata.value() : "") << std::endl;
232237
for(auto track : stream.tracks) {
233238
std::cout << "stream.track[].mid:" << track.mid << std::endl;
239+
std::cout << "stream.track[].type:" << track.type << std::endl;
234240
streamsId.push_back(stream::StreamSubscription{stream.id, track.mid});
235241
}
236242
break;

‎endpoint/programs/stream_testing/single_video_sender.cpp‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,9 +103,13 @@ int main(int argc, char** argv) {
103103
streamApi.addTrack(streamHandle, desktopDevice, stream::MediaTrackConstrains{.idealFps=30});
104104
break;
105105
}
106+
auto dataTrack = streamApi.addTrack(streamHandle, stream::MediaDevice{.name="", .id="", .type=stream::DeviceType::Plain}, stream::MediaTrackConstrains{});
106107

107108
streamApi.publishStream(streamHandle);
108-
std::this_thread::sleep_for(std::chrono::seconds(600));
109+
for(int i = 0; i < 300; i++) {
110+
std::this_thread::sleep_for(std::chrono::seconds(2));
111+
streamApi.sendData(streamHandle, core::Buffer::from("ping"));
112+
}
109113
streamApi.unpublishStream(streamHandle);
110114
std::this_thread::sleep_for(std::chrono::seconds(2));
111115
streamApi.leaveStreamRoom(streamRoomId);

‎endpoint/stream/stream/include_pub/privmx/endpoint/stream/StreamException.hpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,8 @@ DECLARE_ENDPOINT_EXCEPTION(EndpointStreamException, StreamIsPublished, "Stream i
8585
DECLARE_ENDPOINT_EXCEPTION(EndpointStreamException, CannotExtractStreamUpdatedEventException, "Cannot extract StreamUpdatedEvent", 0x0023)
8686
DECLARE_ENDPOINT_EXCEPTION(EndpointStreamException, NullCallbackException, "Callback must not be null", 0x0024)
8787
DECLARE_ENDPOINT_EXCEPTION(EndpointStreamException, UnknownTypeException, "Unknown type encountered", 0x0025)
88+
DECLARE_ENDPOINT_EXCEPTION(EndpointStreamException, ThereCanBeOnlyOneDataTrackException, "There can be only one dataTrack per user in StreamRoom", 0x0026)
89+
DECLARE_ENDPOINT_EXCEPTION(EndpointStreamException, DataTrackNotInitialized, "Data track not initialized", 0x0027);
8890
} // stream
8991
} // endpoint
9092
} // privmx

‎endpoint/stream/stream/src/ServerApi.cpp‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -94,23 +94,18 @@ void ServerApi::trickle(server::StreamTrickleModel model) {
9494
requestWS("streamTrickle", model);
9595
}
9696

97-
9897
template<class T> T ServerApi::request(const std::string& method, Poco::JSON::Object::Ptr params) { //only typed object
99-
LOG_TRACE("ServerApi::request ", method+ ":class");
10098
return privmx::utils::TypedObjectFactory::createObjectFromVar<T>(_gateway->request("stream." + method, params));
10199
}
102100

103101
Poco::Dynamic::Var ServerApi::request(const std::string& method, Poco::JSON::Object::Ptr params) { //var
104-
LOG_TRACE("ServerApi::request ", method+ ":var");
105102
return _gateway->request("stream." + method, params);
106103
}
107104

108105
template<class T> T ServerApi::requestWS(const std::string& method, Poco::JSON::Object::Ptr params) { //only typed object
109-
LOG_TRACE("ServerApi::requestWS ", method+ ":class");
110106
return privmx::utils::TypedObjectFactory::createObjectFromVar<T>(_gateway->request("stream." + method, params, {.channel_type=privmx::rpc::ChannelType::WEBSOCKET}));
111107
}
112108

113109
Poco::Dynamic::Var ServerApi::requestWS(const std::string& method, Poco::JSON::Object::Ptr params) { //var
114-
LOG_TRACE("ServerApi::requestWS ", method+ ":var");
115110
return _gateway->request("stream." + method, params, {.channel_type=privmx::rpc::ChannelType::WEBSOCKET});
116111
}

‎endpoint/stream/stream/src/StreamApiLow.cpp‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -287,7 +287,6 @@ StreamPublishResult StreamApiLow::publishStream(const StreamHandle& streamHandle
287287
try {
288288
return impl->publishStream(streamHandle);
289289
} catch (const privmx::utils::PrivmxException& e) {
290-
std::cerr << e.what() << std::endl;
291290
core::ExceptionConverter::rethrowAsCoreException(e);
292291
throw core::Exception("ExceptionConverter rethrow error");
293292
}
@@ -298,7 +297,6 @@ StreamPublishResult StreamApiLow::updateStream(const StreamHandle& streamHandle)
298297
try {
299298
return impl->updateStream(streamHandle);
300299
} catch (const privmx::utils::PrivmxException& e) {
301-
std::cerr << e.what() << std::endl;
302300
core::ExceptionConverter::rethrowAsCoreException(e);
303301
throw core::Exception("ExceptionConverter rethrow error");
304302
}

‎endpoint/stream/stream/src/StreamApiLowImpl.cpp‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -239,7 +239,7 @@ void StreamApiLowImpl::processNotificationEvent(const core::NotificationEvent& n
239239
_eventMiddleware->emitApiEvent(event);
240240
}
241241
else {
242-
std::cerr << "UNRESOLVED EVENT in CPP layer: '" << type << "'"<< std::endl;
242+
LOG_ERROR("UNRESOLVED EVENT in CPP layer: '", type, "'");
243243
}
244244
}
245245

@@ -397,7 +397,6 @@ StreamPublishResult StreamApiLowImpl::publishStream(const StreamHandle& streamHa
397397
auto model = utils::TypedObjectFactory::createNewObject<server::StreamPublishModel>();
398398
model.streamRoomId(room->streamRoomId);
399399
model.offer(sessionDescription);
400-
std::cout << privmx::utils::Utils::stringifyVar(model) << std::endl;
401400
auto result = _serverApi->streamPublish(model);
402401
streamData->sessionId = result.sessionId();
403402
// update/set sessionId in webrtc (for Janus - trickle)

‎endpoint/stream/webrtc/include/privmx/endpoint/stream/DataChannelImpl.hpp‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
#ifndef _PRIVMXLIB_ENDPOINT_STREAM_DATA_CHANEL_IMPL_HPP_
2+
#define _PRIVMXLIB_ENDPOINT_STREAM_DATA_CHANEL_IMPL_HPP_
3+
4+
#include <string>
5+
#include <libwebrtc.h>
6+
#include <rtc_peerconnection.h>
7+
#include "privmx/endpoint/stream/webrtc/Types.hpp"
8+
#include "privmx/endpoint/stream/webrtc/OnTrackInterface.hpp"
9+
#include "privmx/endpoint/stream/PmxDataChannelObserver.hpp"
10+
#include <privmx/utils/Logger.hpp>
11+
12+
namespace privmx {
13+
namespace endpoint {
14+
namespace stream {
15+
16+
17+
class DataChannelImpl {
18+
public:
19+
inline DataChannelImpl(std::shared_ptr<OnTrackInterface> onTrackInterface, libwebrtc::scoped_refptr<libwebrtc::RTCDataChannel> dataChannel) :
20+
_onTrackInterface(onTrackInterface),
21+
_dataChannel(dataChannel),
22+
_dataChannelObserver(std::make_shared<PmxDataChannelObserver>(
23+
onTrackInterface, dataChannel->label().std_string()+":"+std::to_string(dataChannel->id())
24+
))
25+
{
26+
_dataChannel->RegisterObserver(_dataChannelObserver.get());
27+
LOG_TRACE("DataChannelImpl created")
28+
}
29+
inline ~DataChannelImpl() {
30+
LOG_TRACE("DataChannelImpl destroyed")
31+
}
32+
void updateOnTrackInterface(std::shared_ptr<OnTrackInterface> onTrackInterface) {
33+
std::unique_lock<std::mutex> lock(m);
34+
_onTrackInterface = onTrackInterface;
35+
}
36+
private:
37+
std::mutex m;
38+
std::shared_ptr<OnTrackInterface> _onTrackInterface;
39+
libwebrtc::scoped_refptr<libwebrtc::RTCDataChannel> _dataChannel;
40+
std::shared_ptr<PmxDataChannelObserver> _dataChannelObserver;
41+
42+
};
43+
44+
} // stream
45+
} // endpoint
46+
} // privmx
47+
48+
#endif // _PRIVMXLIB_ENDPOINT_STREAM_DATA_CHANEL_IMPL_HPP_

‎endpoint/stream/webrtc/include/privmx/endpoint/stream/PeerConnectionManager.hpp‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ limitations under the License.
1919
#include "privmx/endpoint/stream/webrtc/Types.hpp"
2020
#include "privmx/endpoint/stream/DynamicTypes.hpp"
2121
#include "privmx/endpoint/stream/PmxPeerConnectionObserver.hpp"
22+
#include "privmx/endpoint/stream/PmxDataChannelObserver.hpp"
2223
#include <privmx/utils/ThreadSaveMap.hpp>
2324

2425
namespace privmx {
@@ -42,12 +43,20 @@ struct VideoTrackInfo {
4243
std::shared_ptr<privmx::webrtc::FrameCryptor> frameCryptor;
4344
};
4445

46+
struct DataChannelInfo {
47+
std::shared_ptr<libwebrtc::RTCDataChannelInit> channelInit;
48+
libwebrtc::scoped_refptr<libwebrtc::RTCDataChannel> channel;
49+
std::shared_ptr<PmxDataChannelObserver> observer;
50+
};
51+
4552
struct PeerConnection {
4653
libwebrtc::scoped_refptr<libwebrtc::RTCPeerConnection> pc;
4754
std::shared_ptr<PmxPeerConnectionObserver> observer;
4855
libwebrtc::scoped_refptr<libwebrtc::RTCMediaStream> mediaStream;
4956
std::map<std::string, AudioTrackInfo> audioTracks;
5057
std::map<std::string, VideoTrackInfo> videoTracks;
58+
std::optional<DataChannelInfo> dataChannel;
59+
5160
std::shared_mutex trackMutex;
5261
std::shared_ptr<privmx::webrtc::KeyStore> keys;
5362
};

‎endpoint/stream/webrtc/include/privmx/endpoint/stream/PmxDataChannelObserver.hpp‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
PrivMX Endpoint.
3+
Copyright © 2024 Simplito sp. z o.o.
4+
5+
This file is part of the PrivMX Platform (https://privmx.dev).
6+
This software is Licensed under the PrivMX Free License.
7+
8+
See the License for the specific language governing permissions and
9+
limitations under the License.
10+
*/
11+
12+
#ifndef _PRIVMXLIB_ENDPOINT_STREAM_PMX_DATA_CHANNEL_OBSERVER_HPP_
13+
#define _PRIVMXLIB_ENDPOINT_STREAM_PMX_DATA_CHANNEL_OBSERVER_HPP_
14+
15+
#include <string>
16+
#include <libwebrtc.h>
17+
#include <rtc_data_channel.h>
18+
#include "privmx/endpoint/stream/webrtc/Types.hpp"
19+
#include "privmx/endpoint/stream/webrtc/OnTrackInterface.hpp"
20+
#include <privmx/utils/ThreadSaveMap.hpp>
21+
#include <privmx/utils/Logger.hpp>
22+
23+
namespace privmx {
24+
namespace endpoint {
25+
namespace stream {
26+
27+
class PmxDataChannelObserver : public libwebrtc::RTCDataChannelObserver {
28+
public:
29+
PmxDataChannelObserver(std::shared_ptr<OnTrackInterface> onTrackInterface, const std::string& dataChannelId);
30+
virtual void OnStateChange(libwebrtc::RTCDataChannelState state) override;
31+
virtual void OnMessage(const char* buffer, int length, bool binary) override;
32+
void updateOnTrackInterface(std::shared_ptr<OnTrackInterface> onTrackInterface);
33+
private:
34+
std::mutex _onTrackInterfaceMutex;
35+
std::shared_ptr<OnTrackInterface> _onTrackInterface;
36+
std::string _dataChannelId;
37+
};
38+
39+
} // stream
40+
} // endpoint
41+
} // privmx
42+
43+
#endif // _PRIVMXLIB_ENDPOINT_STREAM_PMX_DATA_CHANNEL_OBSERVER_HPP_

‎endpoint/stream/webrtc/include/privmx/endpoint/stream/PmxPeerConnectionObserver.hpp‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ limitations under the License.
1717
#include <rtc_peerconnection.h>
1818
#include "privmx/endpoint/stream/webrtc/Types.hpp"
1919
#include "privmx/endpoint/stream/webrtc/OnTrackInterface.hpp"
20+
#include "privmx/endpoint/stream/PmxDataChannelObserver.hpp"
21+
#include "privmx/endpoint/stream/DataChannelImpl.hpp"
2022
#include "privmx/endpoint/stream/RTCVideoRendererImpl.hpp"
2123
#include "privmx/endpoint/stream/AudioTrackSinkImpl.hpp"
2224
#include <privmx/utils/ThreadSaveMap.hpp>
@@ -65,6 +67,7 @@ class PmxPeerConnectionObserver : public libwebrtc::RTCPeerConnectionObserver {
6567

6668
privmx::utils::ThreadSaveMap<std::string, std::shared_ptr<RTCVideoRendererImpl>> _RTCVideoRenderers;
6769
privmx::utils::ThreadSaveMap<std::string, std::shared_ptr<AudioTrackSinkImpl>> _audioTrackSinks;
70+
privmx::utils::ThreadSaveMap<std::string, std::shared_ptr<DataChannelImpl>> _dataChannels;
6871
privmx::utils::ThreadSaveMap<std::string, std::shared_ptr<privmx::webrtc::FrameCryptor>> _frameCryptors;
6972

7073
std::optional<std::function<void(libwebrtc::scoped_refptr<libwebrtc::RTCIceCandidate>)>> _onIceCandidate;

0 commit comments

Comments
 (0)