main.cpp 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461
  1. /*
  2. * libdatachannel client example
  3. * Copyright (c) 2019-2020 Paul-Louis Ageneau
  4. * Copyright (c) 2019 Murat Dogan
  5. * Copyright (c) 2020 Will Munn
  6. * Copyright (c) 2020 Nico Chatzi
  7. * Copyright (c) 2020 Lara Mackey
  8. * Copyright (c) 2020 Erik Cota-Robles
  9. * Copyright (c) 2020 Filip Klembara (in2core)
  10. *
  11. * This program is free software; you can redistribute it and/or
  12. * modify it under the terms of the GNU General Public License
  13. * as published by the Free Software Foundation; either version 2
  14. * of the License, or (at your option) any later version.
  15. *
  16. * This program is distributed in the hope that it will be useful,
  17. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  19. * GNU General Public License for more details.
  20. *
  21. * You should have received a copy of the GNU General Public License
  22. * along with this program; If not, see <http://www.gnu.org/licenses/>.
  23. */
  24. #include "nlohmann/json.hpp"
  25. #include "h264fileparser.hpp"
  26. #include "opusfileparser.hpp"
  27. #include "helpers.hpp"
  28. #include "ArgParser.hpp"
  29. using namespace rtc;
  30. using namespace std;
  31. using namespace std::chrono_literals;
  32. using json = nlohmann::json;
  33. template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
  34. /// all connected clients
  35. unordered_map<string, shared_ptr<Client>> clients{};
  36. /// Creates peer connection and client representation
  37. /// @param config Configuration
  38. /// @param wws Websocket for signaling
  39. /// @param id Client ID
  40. /// @returns Client
  41. shared_ptr<Client> createPeerConnection(const Configuration &config,
  42. weak_ptr<WebSocket> wws,
  43. string id);
  44. /// Creates stream
  45. /// @param h264Samples Directory with H264 samples
  46. /// @param fps Video FPS
  47. /// @param opusSamples Directory with opus samples
  48. /// @returns Stream object
  49. shared_ptr<Stream> createStream(const string h264Samples, const unsigned fps, const string opusSamples);
  50. /// Add client to stream
  51. /// @param client Client
  52. /// @param adding_video True if adding video
  53. void addToStream(shared_ptr<Client> client, bool isAddingVideo);
  54. /// Start stream
  55. void startStream();
  56. /// Main dispatch queue
  57. DispatchQueue MainThread("Main");
  58. /// Audio and video stream
  59. optional<shared_ptr<Stream>> avStream = nullopt;
  60. const string defaultRootDirectory = "../../../examples/streamer/samples/";
  61. const string defaultH264SamplesDirectory = defaultRootDirectory + "h264/";
  62. string h264SamplesDirectory = defaultH264SamplesDirectory;
  63. const string defaultOpusSamplesDirectory = defaultRootDirectory + "opus/";
  64. string opusSamplesDirectory = defaultOpusSamplesDirectory;
  65. const string defaultIPAddress = "127.0.0.1";
  66. const uint16_t defaultPort = 8000;
  67. string ip_address = defaultIPAddress;
  68. uint16_t port = defaultPort;
  69. /// Incomming message handler for websocket
  70. /// @param message Incommint message
  71. /// @param config Configuration
  72. /// @param ws Websocket
  73. void wsOnMessage(json message, Configuration config, shared_ptr<WebSocket> ws) {
  74. auto it = message.find("id");
  75. if (it == message.end())
  76. return;
  77. string id = it->get<string>();
  78. it = message.find("type");
  79. if (it == message.end())
  80. return;
  81. string type = it->get<string>();
  82. if (type == "request") {
  83. clients.emplace(id, createPeerConnection(config, make_weak_ptr(ws), id));
  84. } else if (type == "answer") {
  85. if (auto jt = clients.find(id); jt != clients.end()) {
  86. auto pc = jt->second->peerConnection;
  87. auto sdp = message["sdp"].get<string>();
  88. auto description = Description(sdp, type);
  89. pc->setRemoteDescription(description);
  90. }
  91. }
  92. }
  93. int main(int argc, char **argv) try {
  94. bool enableDebugLogs = false;
  95. bool printHelp = false;
  96. int c = 0;
  97. auto parser = ArgParser({{"a", "audio"}, {"b", "video"}, {"d", "ip"}, {"p","port"}}, {{"h", "help"}, {"v", "verbose"}});
  98. auto parsingResult = parser.parse(argc, argv, [](string key, string value) {
  99. if (key == "audio") {
  100. opusSamplesDirectory = value + "/";
  101. } else if (key == "video") {
  102. h264SamplesDirectory = value + "/";
  103. } else if (key == "ip") {
  104. ip_address = value;
  105. } else if (key == "port") {
  106. port = atoi(value.data());
  107. } else {
  108. cerr << "Invalid option --" << key << " with value " << value << endl;
  109. return false;
  110. }
  111. return true;
  112. }, [&enableDebugLogs, &printHelp](string flag){
  113. if (flag == "verbose") {
  114. enableDebugLogs = true;
  115. } else if (flag == "help") {
  116. printHelp = true;
  117. } else {
  118. cerr << "Invalid flag --" << flag << endl;
  119. return false;
  120. }
  121. return true;
  122. });
  123. if (!parsingResult) {
  124. return 1;
  125. }
  126. if (printHelp) {
  127. cout << "usage: stream-h264 [-a opus_samples_folder] [-b h264_samples_folder] [-d ip_address] [-p port] [-v] [-h]" << endl
  128. << "Arguments:" << endl
  129. << "\t -a " << "Directory with opus samples (default: " << defaultOpusSamplesDirectory << ")." << endl
  130. << "\t -b " << "Directory with H264 samples (default: " << defaultH264SamplesDirectory << ")." << endl
  131. << "\t -d " << "Signaling server IP address (default: " << defaultIPAddress << ")." << endl
  132. << "\t -p " << "Signaling server port (default: " << defaultPort << ")." << endl
  133. << "\t -v " << "Enable debug logs." << endl
  134. << "\t -h " << "Print this help and exit." << endl;
  135. return 0;
  136. }
  137. if (enableDebugLogs) {
  138. InitLogger(LogLevel::Debug);
  139. }
  140. Configuration config;
  141. string stunServer = "stun:stun.l.google.com:19302";
  142. cout << "STUN server is " << stunServer << endl;
  143. config.iceServers.emplace_back(stunServer);
  144. config.disableAutoNegotiation = true;
  145. string localId = "server";
  146. cout << "The local ID is: " << localId << endl;
  147. auto ws = make_shared<WebSocket>();
  148. ws->onOpen([]() { cout << "WebSocket connected, signaling ready" << endl; });
  149. ws->onClosed([]() { cout << "WebSocket closed" << endl; });
  150. ws->onError([](const string &error) { cout << "WebSocket failed: " << error << endl; });
  151. ws->onMessage([&](variant<binary, string> data) {
  152. if (!holds_alternative<string>(data))
  153. return;
  154. json message = json::parse(get<string>(data));
  155. MainThread.dispatch([message, config, ws]() {
  156. wsOnMessage(message, config, ws);
  157. });
  158. });
  159. const string url = "ws://" + ip_address + ":" + to_string(port) + "/" + localId;
  160. cout << "URL is " << url << endl;
  161. ws->open(url);
  162. cout << "Waiting for signaling to be connected..." << endl;
  163. while (!ws->isOpen()) {
  164. if (ws->isClosed())
  165. return 1;
  166. this_thread::sleep_for(100ms);
  167. }
  168. while (true) {
  169. string id;
  170. cout << "Enter to exit" << endl;
  171. cin >> id;
  172. cin.ignore();
  173. cout << "exiting" << endl;
  174. break;
  175. }
  176. cout << "Cleaning up..." << endl;
  177. return 0;
  178. } catch (const std::exception &e) {
  179. std::cout << "Error: " << e.what() << std::endl;
  180. return -1;
  181. }
  182. shared_ptr<ClientTrackData> addVideo(const shared_ptr<PeerConnection> pc, const uint8_t payloadType, const uint32_t ssrc, const string cname, const string msid, const function<void (void)> onOpen) {
  183. auto video = Description::Video(cname);
  184. video.addH264Codec(payloadType);
  185. video.addSSRC(ssrc, cname, msid, cname);
  186. auto track = pc->addTrack(video);
  187. // create RTP configuration
  188. auto rtpConfig = make_shared<RtpPacketizationConfig>(ssrc, cname, payloadType, H264RtpPacketizer::defaultClockRate);
  189. // create packetizer
  190. auto packetizer = make_shared<H264RtpPacketizer>(H264RtpPacketizer::Separator::Length, rtpConfig);
  191. // create H264 handler
  192. auto h264Handler = make_shared<H264PacketizationHandler>(packetizer);
  193. // add RTCP SR handler
  194. auto srReporter = make_shared<RtcpSrReporter>(rtpConfig);
  195. h264Handler->addToChain(srReporter);
  196. // add RTCP NACK handler
  197. auto nackResponder = make_shared<RtcpNackResponder>();
  198. h264Handler->addToChain(nackResponder);
  199. // set handler
  200. track->setMediaHandler(h264Handler);
  201. track->onOpen(onOpen);
  202. auto trackData = make_shared<ClientTrackData>(track, srReporter);
  203. return trackData;
  204. }
  205. shared_ptr<ClientTrackData> addAudio(const shared_ptr<PeerConnection> pc, const uint8_t payloadType, const uint32_t ssrc, const string cname, const string msid, const function<void (void)> onOpen) {
  206. auto audio = Description::Audio(cname);
  207. audio.addOpusCodec(payloadType);
  208. audio.addSSRC(ssrc, cname, msid, cname);
  209. auto track = pc->addTrack(audio);
  210. // create RTP configuration
  211. auto rtpConfig = make_shared<RtpPacketizationConfig>(ssrc, cname, payloadType, OpusRtpPacketizer::defaultClockRate);
  212. // create packetizer
  213. auto packetizer = make_shared<OpusRtpPacketizer>(rtpConfig);
  214. // create opus handler
  215. auto opusHandler = make_shared<OpusPacketizationHandler>(packetizer);
  216. // add RTCP SR handler
  217. auto srReporter = make_shared<RtcpSrReporter>(rtpConfig);
  218. opusHandler->addToChain(srReporter);
  219. // add RTCP NACK handler
  220. auto nackResponder = make_shared<RtcpNackResponder>();
  221. opusHandler->addToChain(nackResponder);
  222. // set handler
  223. track->setMediaHandler(opusHandler);
  224. track->onOpen(onOpen);
  225. auto trackData = make_shared<ClientTrackData>(track, srReporter);
  226. return trackData;
  227. }
  228. // Create and setup a PeerConnection
  229. shared_ptr<Client> createPeerConnection(const Configuration &config,
  230. weak_ptr<WebSocket> wws,
  231. string id) {
  232. auto pc = make_shared<PeerConnection>(config);
  233. auto client = make_shared<Client>(pc);
  234. pc->onStateChange([id](PeerConnection::State state) {
  235. cout << "State: " << state << endl;
  236. if (state == PeerConnection::State::Disconnected ||
  237. state == PeerConnection::State::Failed ||
  238. state == PeerConnection::State::Closed) {
  239. // remove disconnected client
  240. MainThread.dispatch([id]() {
  241. clients.erase(id);
  242. });
  243. }
  244. });
  245. pc->onGatheringStateChange(
  246. [wpc = make_weak_ptr(pc), id, wws](PeerConnection::GatheringState state) {
  247. cout << "Gathering State: " << state << endl;
  248. if (state == PeerConnection::GatheringState::Complete) {
  249. if(auto pc = wpc.lock()) {
  250. auto description = pc->localDescription();
  251. json message = {
  252. {"id", id},
  253. {"type", description->typeString()},
  254. {"sdp", string(description.value())}
  255. };
  256. // Gathering complete, send answer
  257. if (auto ws = wws.lock()) {
  258. ws->send(message.dump());
  259. }
  260. }
  261. }
  262. });
  263. client->video = addVideo(pc, 102, 1, "video-stream", "stream1", [id, wc = make_weak_ptr(client)]() {
  264. MainThread.dispatch([wc]() {
  265. if (auto c = wc.lock()) {
  266. addToStream(c, true);
  267. }
  268. });
  269. cout << "Video from " << id << " opened" << endl;
  270. });
  271. client->audio = addAudio(pc, 111, 2, "audio-stream", "stream1", [id, wc = make_weak_ptr(client)]() {
  272. MainThread.dispatch([wc]() {
  273. if (auto c = wc.lock()) {
  274. addToStream(c, false);
  275. }
  276. });
  277. cout << "Audio from " << id << " opened" << endl;
  278. });
  279. auto dc = pc->createDataChannel("ping-pong");
  280. dc->onOpen([id, wdc = make_weak_ptr(dc)]() {
  281. if (auto dc = wdc.lock()) {
  282. dc->send("Ping");
  283. }
  284. });
  285. dc->onMessage(nullptr, [id, wdc = make_weak_ptr(dc)](string msg) {
  286. cout << "Message from " << id << " received: " << msg << endl;
  287. if (auto dc = wdc.lock()) {
  288. dc->send("Ping");
  289. }
  290. });
  291. client->dataChannel = dc;
  292. pc->setLocalDescription();
  293. return client;
  294. };
  295. /// Create stream
  296. shared_ptr<Stream> createStream(const string h264Samples, const unsigned fps, const string opusSamples) {
  297. // video source
  298. auto video = make_shared<H264FileParser>(h264Samples, fps, true);
  299. // audio source
  300. auto audio = make_shared<OPUSFileParser>(opusSamples, true);
  301. auto stream = make_shared<Stream>(video, audio);
  302. // set callback responsible for sample sending
  303. stream->onSample([ws = make_weak_ptr(stream)](Stream::StreamSourceType type, uint64_t sampleTime, rtc::binary sample) {
  304. vector<ClientTrack> tracks{};
  305. string streamType = type == Stream::StreamSourceType::Video ? "video" : "audio";
  306. // get track for given type
  307. function<optional<shared_ptr<ClientTrackData>> (shared_ptr<Client>)> getTrackData = [type](shared_ptr<Client> client) {
  308. return type == Stream::StreamSourceType::Video ? client->video : client->audio;
  309. };
  310. // get all clients with Ready state
  311. for(auto id_client: clients) {
  312. auto id = id_client.first;
  313. auto client = id_client.second;
  314. auto optTrackData = getTrackData(client);
  315. if (client->getState() == Client::State::Ready && optTrackData.has_value()) {
  316. auto trackData = optTrackData.value();
  317. tracks.push_back(ClientTrack(id, trackData));
  318. }
  319. }
  320. if (!tracks.empty()) {
  321. for (auto clientTrack: tracks) {
  322. auto client = clientTrack.id;
  323. auto trackData = clientTrack.trackData;
  324. auto rtpConfig = trackData->sender->rtpConfig;
  325. // sample time is in us, we need to convert it to seconds
  326. auto elapsedSeconds = double(sampleTime) / (1000 * 1000);
  327. // get elapsed time in clock rate
  328. uint32_t elapsedTimestamp = rtpConfig->secondsToTimestamp(elapsedSeconds);
  329. // set new timestamp
  330. rtpConfig->timestamp = rtpConfig->startTimestamp + elapsedTimestamp;
  331. // get elapsed time in clock rate from last RTCP sender report
  332. auto reportElapsedTimestamp = rtpConfig->timestamp - trackData->sender->lastReportedTimestamp();
  333. // check if last report was at least 1 second ago
  334. if (rtpConfig->timestampToSeconds(reportElapsedTimestamp) > 1) {
  335. trackData->sender->setNeedsToReport();
  336. }
  337. cout << "Sending " << streamType << " sample with size: " << to_string(sample.size()) << " to " << client << endl;
  338. try {
  339. // send sample
  340. trackData->track->send(sample);
  341. } catch (const std::exception &e) {
  342. cerr << "Unable to send "<< streamType << " packet: " << e.what() << endl;
  343. }
  344. }
  345. }
  346. MainThread.dispatch([ws]() {
  347. if (clients.empty()) {
  348. // we have no clients, stop the stream
  349. if (auto stream = ws.lock()) {
  350. stream->stop();
  351. }
  352. }
  353. });
  354. });
  355. return stream;
  356. }
  357. /// Start stream
  358. void startStream() {
  359. shared_ptr<Stream> stream;
  360. if (avStream.has_value()) {
  361. stream = avStream.value();
  362. if (stream->isRunning) {
  363. // stream is already running
  364. return;
  365. }
  366. } else {
  367. stream = createStream(h264SamplesDirectory, 30, opusSamplesDirectory);
  368. avStream = stream;
  369. }
  370. stream->start();
  371. }
  372. /// Send previous key frame so browser can show something to user
  373. /// @param stream Stream
  374. /// @param video Video track data
  375. void sendInitialNalus(shared_ptr<Stream> stream, shared_ptr<ClientTrackData> video) {
  376. auto h264 = dynamic_cast<H264FileParser *>(stream->video.get());
  377. auto initialNalus = h264->initialNALUS();
  378. // send previous NALU key frame so users don't have to wait to see stream works
  379. if (!initialNalus.empty()) {
  380. const double frameDuration_s = double(h264->getSampleDuration_us()) / (1000 * 1000);
  381. const uint32_t frameTimestampDuration = video->sender->rtpConfig->secondsToTimestamp(frameDuration_s);
  382. video->sender->rtpConfig->timestamp = video->sender->rtpConfig->startTimestamp - frameTimestampDuration * 2;
  383. video->track->send(initialNalus);
  384. video->sender->rtpConfig->timestamp += frameTimestampDuration;
  385. // Send initial NAL units again to start stream in firefox browser
  386. video->track->send(initialNalus);
  387. }
  388. }
  389. /// Add client to stream
  390. /// @param client Client
  391. /// @param adding_video True if adding video
  392. void addToStream(shared_ptr<Client> client, bool isAddingVideo) {
  393. if (client->getState() == Client::State::Waiting) {
  394. client->setState(isAddingVideo ? Client::State::WaitingForAudio : Client::State::WaitingForVideo);
  395. } else if ((client->getState() == Client::State::WaitingForAudio && !isAddingVideo)
  396. || (client->getState() == Client::State::WaitingForVideo && isAddingVideo)) {
  397. // Audio and video tracks are collected now
  398. assert(client->video.has_value() && client->audio.has_value());
  399. auto video = client->video.value();
  400. if (avStream.has_value()) {
  401. sendInitialNalus(avStream.value(), video);
  402. }
  403. client->setState(Client::State::Ready);
  404. }
  405. if (client->getState() == Client::State::Ready) {
  406. startStream();
  407. }
  408. }