Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "libav_deps.h" // MUST BE INCLUDED FIRST
19 : #include "video_receive_thread.h"
20 : #include "media/media_decoder.h"
21 : #include "socket_pair.h"
22 : #include "manager.h"
23 : #include "logger.h"
24 :
25 : extern "C" {
26 : #include <libavutil/display.h>
27 : }
28 :
29 : #include <unistd.h>
30 :
31 : namespace jami {
32 : namespace video {
33 :
34 : using std::string;
35 :
36 55 : VideoReceiveThread::VideoReceiveThread(const std::string& id, bool useSink, const std::string& sdp, uint16_t mtu)
37 : : VideoGenerator::VideoGenerator()
38 55 : , args_()
39 55 : , id_(id)
40 55 : , useSink_(useSink)
41 55 : , stream_(sdp)
42 55 : , sdpContext_(stream_.str().size(), false, &readFunction, 0, 0, this)
43 55 : , sink_ {Manager::instance().createSinkClient(id)}
44 55 : , mtu_(mtu)
45 110 : , loop_(std::bind(&VideoReceiveThread::setup, this),
46 110 : std::bind(&VideoReceiveThread::decodeFrame, this),
47 220 : std::bind(&VideoReceiveThread::cleanup, this))
48 : {
49 55 : JAMI_LOG("[{}] Instance created", fmt::ptr(this));
50 55 : }
51 :
52 165 : VideoReceiveThread::~VideoReceiveThread()
53 : {
54 55 : loop_.join();
55 55 : JAMI_LOG("[{}] Instance destroyed", fmt::ptr(this));
56 110 : }
57 :
58 : void
59 55 : VideoReceiveThread::startLoop()
60 : {
61 55 : JAMI_LOG("[{}] Starting receiver’s loop", fmt::ptr(this));
62 55 : loop_.start();
63 55 : }
64 :
65 : void
66 114 : VideoReceiveThread::stopLoop()
67 : {
68 114 : if (loop_.isStopping())
69 59 : return;
70 55 : JAMI_LOG("[{}] Stopping receiver’s loop and waiting for the thread to exit…", fmt::ptr(this));
71 55 : loop_.stop();
72 55 : loop_.join();
73 55 : JAMI_LOG("[{}] Receiver’s thread exited", fmt::ptr(this));
74 : }
75 :
76 : // We do this setup here instead of the constructor because we don't want the
77 : // main thread to block while this executes, so it happens in the video thread.
78 : bool
79 55 : VideoReceiveThread::setup()
80 : {
81 55 : JAMI_LOG("[{}] Setting up video receiver", fmt::ptr(this));
82 :
83 55 : videoDecoder_.reset(new MediaDecoder([this](const std::shared_ptr<MediaFrame>& frame) mutable {
84 0 : libav_utils::AVBufferPtr displayMatrix;
85 : {
86 0 : std::lock_guard l(rotationMtx_);
87 0 : if (displayMatrix_)
88 0 : displayMatrix.reset(av_buffer_ref(displayMatrix_.get()));
89 0 : }
90 0 : if (displayMatrix)
91 0 : av_frame_new_side_data_from_buf(frame->pointer(), AV_FRAME_DATA_DISPLAYMATRIX, displayMatrix.release());
92 0 : publishFrame(std::static_pointer_cast<VideoFrame>(frame));
93 110 : }));
94 55 : videoDecoder_->setContextCallback([this]() {
95 0 : if (recorderCallback_)
96 0 : recorderCallback_(getInfo());
97 0 : });
98 0 : videoDecoder_->setResolutionChangedCallback([this](int width, int height) {
99 0 : dstWidth_ = width;
100 0 : dstHeight_ = height;
101 0 : sink_->setFrameSize(dstWidth_, dstHeight_);
102 0 : });
103 :
104 55 : dstWidth_ = static_cast<int>(args_.width);
105 55 : dstHeight_ = static_cast<int>(args_.height);
106 :
107 55 : static const std::string SDP_FILENAME = "dummyFilename";
108 55 : if (args_.input.empty()) {
109 55 : args_.format = "sdp";
110 55 : args_.input = SDP_FILENAME;
111 0 : } else if (args_.input.substr(0, strlen("/dev/video")) == "/dev/video") {
112 : // it's a v4l device if starting with /dev/video
113 : // FIXME: This is not a robust way of checking if we mean to use a
114 : // v4l2 device
115 0 : args_.format = "video4linux2";
116 : }
117 :
118 55 : videoDecoder_->setInterruptCallback(interruptCb, this);
119 :
120 55 : if (args_.input == SDP_FILENAME) {
121 : // Force custom_io so the SDP demuxer will not open any UDP connections
122 : // We need it to use ICE transport.
123 55 : args_.sdp_flags = "custom_io";
124 :
125 55 : if (stream_.str().empty()) {
126 0 : JAMI_ERROR("No SDP loaded");
127 0 : return false;
128 : }
129 :
130 55 : videoDecoder_->setIOContext(&sdpContext_);
131 : }
132 :
133 55 : args_.disable_dts_probe_delay = true;
134 :
135 55 : if (videoDecoder_->openInput(args_)) {
136 0 : JAMI_ERROR("Unable to open input \"{}\"", args_.input);
137 0 : return false;
138 : }
139 :
140 55 : videoDecoder_->setKeyFrameRequestCb(keyFrameRequestCallback_);
141 :
142 55 : if (args_.input == SDP_FILENAME) {
143 : // Now replace our custom AVIOContext with one that will read packets
144 55 : videoDecoder_->setIOContext(demuxContext_.get());
145 : }
146 55 : return true;
147 : }
148 :
149 : void
150 55 : VideoReceiveThread::cleanup()
151 : {
152 55 : JAMI_LOG("[{}] Stopping receiver", fmt::ptr(this));
153 :
154 55 : detach(sink_.get());
155 55 : sink_->stop();
156 :
157 55 : videoDecoder_.reset();
158 55 : }
159 :
160 : // This callback is used by libav internally to break out of blocking calls
161 : int
162 56 : VideoReceiveThread::interruptCb(void* data)
163 : {
164 56 : auto* const context = static_cast<VideoReceiveThread*>(data);
165 56 : return not context->loop_.isRunning();
166 : }
167 :
168 : int
169 185 : VideoReceiveThread::readFunction(void* opaque, uint8_t* buf, int buf_size)
170 : {
171 185 : std::istream& is = static_cast<VideoReceiveThread*>(opaque)->stream_;
172 185 : is.read(reinterpret_cast<char*>(buf), buf_size);
173 :
174 185 : auto count = is.gcount();
175 185 : if (count != 0)
176 55 : return static_cast<int>(count);
177 : else
178 130 : return AVERROR_EOF;
179 : }
180 :
181 : void
182 55 : VideoReceiveThread::addIOContext(SocketPair& socketPair)
183 : {
184 55 : demuxContext_.reset(socketPair.createIOContext(mtu_));
185 55 : }
186 :
187 : void
188 60 : VideoReceiveThread::setRecorderCallback(const std::function<void(const MediaStream& ms)>& cb)
189 : {
190 60 : recorderCallback_ = cb;
191 60 : if (videoDecoder_)
192 2 : videoDecoder_->setContextCallback([this]() {
193 0 : if (recorderCallback_)
194 0 : recorderCallback_(getInfo());
195 0 : });
196 60 : }
197 :
198 : void
199 58 : VideoReceiveThread::decodeFrame()
200 : {
201 58 : if (not loop_.isRunning())
202 0 : return;
203 :
204 58 : if (not isVideoConfigured_) {
205 55 : if (!configureVideoOutput()) {
206 0 : JAMI_ERROR("[{:p}] Failed to configure video output", fmt::ptr(this));
207 0 : return;
208 : } else {
209 55 : JAMI_LOG("[{:p}] Decoder configured, starting decoding", fmt::ptr(this));
210 : }
211 : }
212 58 : auto status = videoDecoder_->decode();
213 58 : if (status == MediaDemuxer::Status::EndOfFile) {
214 54 : JAMI_LOG("[{:p}] End of file", fmt::ptr(this));
215 54 : loop_.stop();
216 4 : } else if (status == MediaDemuxer::Status::ReadError) {
217 0 : JAMI_ERROR("[{:p}] Decoding error: %s", fmt::ptr(this), MediaDemuxer::getStatusStr(status));
218 4 : } else if (status == MediaDemuxer::Status::FallBack) {
219 0 : if (keyFrameRequestCallback_)
220 0 : keyFrameRequestCallback_();
221 : }
222 : }
223 :
224 : bool
225 55 : VideoReceiveThread::configureVideoOutput()
226 : {
227 55 : assert(not isVideoConfigured_);
228 :
229 55 : JAMI_LOG("[{}] Configuring video output", fmt::ptr(this));
230 :
231 55 : if (not loop_.isRunning()) {
232 0 : JAMI_WARNING("[{}] Unable to configure video output, the loop is not running!", fmt::ptr(this));
233 0 : return false;
234 : }
235 :
236 55 : if (videoDecoder_->setupVideo() < 0) {
237 0 : JAMI_ERROR("Decoder IO startup failed");
238 0 : stopLoop();
239 0 : return false;
240 : }
241 :
242 : // Default size from input video
243 55 : if (dstWidth_ == 0 and dstHeight_ == 0) {
244 55 : dstWidth_ = videoDecoder_->getWidth();
245 55 : dstHeight_ = videoDecoder_->getHeight();
246 : }
247 :
248 55 : if (not sink_->start()) {
249 0 : JAMI_ERROR("RX: sink startup failed");
250 0 : stopLoop();
251 0 : return false;
252 : }
253 :
254 55 : if (useSink_)
255 54 : startSink();
256 :
257 55 : if (onSuccessfulSetup_)
258 55 : onSuccessfulSetup_(MEDIA_VIDEO, 1);
259 :
260 55 : return isVideoConfigured_ = true;
261 : }
262 :
263 : void
264 115 : VideoReceiveThread::stopSink()
265 : {
266 115 : JAMI_LOG("[{}] Stopping sink", fmt::ptr(this));
267 :
268 115 : if (!loop_.isRunning())
269 114 : return;
270 :
271 1 : detach(sink_.get());
272 1 : sink_->setFrameSize(0, 0);
273 : }
274 :
275 : void
276 55 : VideoReceiveThread::startSink()
277 : {
278 55 : JAMI_LOG("[{}] Starting sink", fmt::ptr(this));
279 :
280 55 : if (!loop_.isRunning())
281 53 : return;
282 :
283 2 : if (dstWidth_ > 0 and dstHeight_ > 0 and attach(sink_.get()))
284 1 : sink_->setFrameSize(dstWidth_, dstHeight_);
285 : }
286 :
287 : int
288 0 : VideoReceiveThread::getWidth() const
289 : {
290 0 : return dstWidth_;
291 : }
292 :
293 : int
294 0 : VideoReceiveThread::getHeight() const
295 : {
296 0 : return dstHeight_;
297 : }
298 :
299 : AVPixelFormat
300 0 : VideoReceiveThread::getPixelFormat() const
301 : {
302 0 : if (videoDecoder_)
303 0 : return videoDecoder_->getPixelFormat();
304 0 : return {};
305 : }
306 :
307 : MediaStream
308 157 : VideoReceiveThread::getInfo() const
309 : {
310 157 : if (videoDecoder_)
311 165 : return videoDecoder_->getStream("v:remote");
312 102 : return {};
313 : }
314 :
315 : void
316 56 : VideoReceiveThread::setRotation(int angle)
317 : {
318 56 : libav_utils::AVBufferPtr displayMatrix(av_buffer_alloc(sizeof(int32_t) * 9));
319 56 : av_display_rotation_set(reinterpret_cast<int32_t*>(displayMatrix->data), angle);
320 56 : std::lock_guard l(rotationMtx_);
321 56 : displayMatrix_ = std::move(displayMatrix);
322 56 : }
323 :
324 : } // namespace video
325 : } // namespace jami
|
