Thread pool; onAudioFrame callback; FakeAudioDevice; fake plaform

.clang-format

@@ -0,0 +1,2 @@
+DisableFormat: true
+SortIncludes: false

tgcalls/AudioFrame.h

@@ -0,0 +1,16 @@
+#pragma once
+
+#include <cinttypes>
+#include <cstring>
+
+namespace tgcalls {
+struct AudioFrame {
+  const int16_t *audio_samples;
+  size_t num_samples;
+  size_t bytes_per_sample;
+  size_t num_channels;
+  uint32_t samples_per_sec;
+  int64_t elapsed_time_ms;
+  int64_t ntp_time_ms;
+};
+}  // namespace tgcalls

tgcalls/FakeAudioDeviceModule.cpp

@@ -0,0 +1,179 @@
+#include "FakeAudioDeviceModule.h"
+
+#include "modules/audio_device/include/audio_device_default.h"
+#include "rtc_base/ref_counted_object.h"
+#include "rtc_base/platform_thread.h"
+#include "rtc_base/time_utils.h"
+
+#include <thread>
+
+namespace tgcalls {
+class FakeAudioDeviceModuleImpl : public webrtc::webrtc_impl::AudioDeviceModuleDefault<webrtc::AudioDeviceModule> {
+ public:
+  static rtc::scoped_refptr<webrtc::AudioDeviceModule> Create(webrtc::TaskQueueFactory* taskQueueFactory,
+                                                              std::shared_ptr<FakeAudioDeviceModule::Renderer> renderer,
+                                                              FakeAudioDeviceModule::Options options) {
+    return rtc::scoped_refptr<webrtc::AudioDeviceModule>(
+        new rtc::RefCountedObject<FakeAudioDeviceModuleImpl>(taskQueueFactory, options, std::move(renderer)));
+  }
+
+  FakeAudioDeviceModuleImpl(webrtc::TaskQueueFactory*, FakeAudioDeviceModule::Options options,
+                            std::shared_ptr<FakeAudioDeviceModule::Renderer> renderer)
+      : num_channels_{options.num_channels}, samples_per_sec_{options.samples_per_sec}, renderer_(std::move(renderer)) {
+    RTC_CHECK(num_channels_ == 1 || num_channels_ == 2);
+    auto good_sample_rate = [](size_t sr) {
+      return sr == 8000 || sr == 16000 || sr == 32000 || sr == 44100 || sr == 48000;
+    };
+    RTC_CHECK(good_sample_rate(samples_per_sec_));
+    samples_per_frame_ = samples_per_sec_ / 100;
+    playout_buffer_.resize(samples_per_frame_ * 2 /* 2 in case stereo will be turned on later */, 0);
+  }
+
+  ~FakeAudioDeviceModuleImpl() override {
+    StopPlayout();
+  }
+
+  int32_t PlayoutIsAvailable(bool* available) override {
+    if (available) {
+      *available = true;
+    }
+    return 0;
+  }
+
+  int32_t StereoPlayoutIsAvailable(bool* available) const override {
+    if (available) {
+      *available = true;
+    }
+    return 0;
+  }
+  int32_t StereoPlayout(bool* enabled) const override {
+    if (enabled) {
+      *enabled = num_channels_ == 2;
+    }
+    return 0;
+  }
+  int32_t SetStereoPlayout(bool enable) override {
+    size_t new_num_channels = enable ? 2 : 1;
+    if (new_num_channels != num_channels_) {
+      return -1;
+    }
+    return 0;
+  }
+
+  int32_t Init() override {
+    return 0;
+  }
+
+  int32_t RegisterAudioCallback(webrtc::AudioTransport* callback) override {
+    webrtc::MutexLock lock(&lock_);
+    audio_callback_ = callback;
+    return 0;
+  }
+
+  int32_t StartPlayout() override {
+    webrtc::MutexLock lock(&lock_);
+    RTC_CHECK(renderer_);
+    if (rendering_) {
+      return 0;
+    }
+    rendering_ = true;
+    renderThread_ = std::make_unique<rtc::PlatformThread>(
+        RenderThreadFunction, this, "webrtc_fake_audio_module_capture_thread", rtc::kRealtimePriority);
+    renderThread_->Start();
+
+    return 0;
+  }
+
+  int32_t StopPlayout() override {
+    if (!rendering_) {
+      return 0;
+    }
+    decltype(renderThread_) thread;
+
+    {
+      webrtc::MutexLock lock(&lock_);
+      thread = std::move(renderThread_);
+      rendering_ = false;
+    }
+
+    thread->Stop();
+    return 0;
+  }
+
+  bool Playing() const override {
+    return rendering_;
+  }
+
+ private:
+  static void RenderThreadFunction(void* pThis) {
+    auto* device = static_cast<FakeAudioDeviceModuleImpl*>(pThis);
+    while (true) {
+      int wait_for_us = device->Render();
+      if (wait_for_us < 0) {
+        break;
+      }
+      std::this_thread::sleep_for(std::chrono::microseconds(wait_for_us));
+    }
+  }
+
+  int32_t Render() {
+    webrtc::MutexLock lock(&lock_);
+    if (!rendering_) {
+      return -1;
+    }
+    size_t samples_out = 0;
+    int64_t elapsed_time_ms = -1;
+    int64_t ntp_time_ms = -1;
+    size_t bytes_per_sample = 2;
+
+    RTC_CHECK(audio_callback_);
+    if (renderer_) {
+      renderer_->BeginFrame(0);
+    }
+    audio_callback_->NeedMorePlayData(samples_per_frame_, bytes_per_sample, num_channels_, samples_per_sec_,
+                                      playout_buffer_.data(), samples_out, &elapsed_time_ms, &ntp_time_ms);
+    if (renderer_) {
+      renderer_->EndFrame();
+    }
+    if (samples_out != 0 && renderer_) {
+      AudioFrame frame;
+      frame.audio_samples = playout_buffer_.data();
+      frame.num_samples = samples_out;
+      frame.bytes_per_sample = bytes_per_sample;
+      frame.num_channels = num_channels_;
+      frame.samples_per_sec = samples_per_sec_;
+      frame.elapsed_time_ms = elapsed_time_ms;
+      frame.ntp_time_ms = ntp_time_ms;
+      renderer_->Render(frame);
+    }
+    int32_t wait_for_us = -1;
+    if (renderer_) {
+      wait_for_us = renderer_->WaitForUs();
+    }
+    return wait_for_us;
+  }
+
+  size_t num_channels_;
+  const uint32_t samples_per_sec_;
+  size_t samples_per_frame_{0};
+
+  mutable webrtc::Mutex lock_;
+  std::atomic<bool> rendering_{false};
+  std::unique_ptr<rtc::PlatformThread> renderThread_;
+
+  webrtc::AudioTransport* audio_callback_{nullptr};
+  const std::shared_ptr<FakeAudioDeviceModule::Renderer> renderer_ RTC_GUARDED_BY(lock_);
+  std::vector<int16_t> playout_buffer_ RTC_GUARDED_BY(lock_);
+};
+
+std::function<rtc::scoped_refptr<webrtc::AudioDeviceModule>(webrtc::TaskQueueFactory*)> FakeAudioDeviceModule::Creator(
+    std::shared_ptr<Renderer> renderer, Options options) {
+  bool is_renderer_empty = bool(renderer);
+  auto boxed_renderer = std::make_shared<std::shared_ptr<Renderer>>(std::move(renderer));
+  return
+      [boxed_renderer = std::move(boxed_renderer), is_renderer_empty, options](webrtc::TaskQueueFactory* task_factory) {
+        RTC_CHECK(is_renderer_empty == bool(*boxed_renderer));  // call only once if renderer exists
+        return FakeAudioDeviceModuleImpl::Create(task_factory, std::move(*boxed_renderer), options);
+      };
+}
+}  // namespace tgcalls

tgcalls/FakeAudioDeviceModule.h

@@ -0,0 +1,42 @@
+#pragma once
+
+#include <functional>
+#include <memory>
+
+#include "AudioFrame.h"
+
+namespace webrtc {
+class AudioDeviceModule;
+class TaskQueueFactory;
+}  // namespace webrtc
+
+namespace rtc {
+template <class T>
+class scoped_refptr;
+}
+
+namespace tgcalls {
+class FakeAudioDeviceModule {
+ public:
+  class Renderer {
+   public:
+    virtual ~Renderer() = default;
+    virtual bool Render(const AudioFrame &samples) = 0;
+    virtual void BeginFrame(double timestamp) {
+    }
+    virtual void AddFrameChannel(uint32_t ssrc, const tgcalls::AudioFrame &frame) {
+    }
+    virtual void EndFrame() {
+    }
+    virtual int32_t WaitForUs() {
+       return 10000;
+    }
+  };
+  struct Options {
+    uint32_t samples_per_sec{48000};
+    uint32_t num_channels{2};
+  };
+  static std::function<rtc::scoped_refptr<webrtc::AudioDeviceModule>(webrtc::TaskQueueFactory *)> Creator(
+      std::shared_ptr<Renderer> renderer, Options options);
+};
+}  // namespace tgcalls

tgcalls/StaticThreads.cpp

@@ -2,46 +2,134 @@
 
 #include "rtc_base/thread.h"
 
+#include <mutex>
+#include <algorithm>
+
 namespace tgcalls {
 
+template <class ValueT, class CreatorT>
+class Pool : public std::enable_shared_from_this<Pool<ValueT, CreatorT>> {
+  struct Entry {
+    std::unique_ptr<ValueT> value;
+    size_t refcnt;
+
+    bool operator < (const Entry &other) const {
+      return refcnt < other.refcnt;
+    }
+  };
+
+public:
+  explicit Pool(CreatorT creator) : creator_(std::move(creator)) {
+  }
+  std::shared_ptr<ValueT> get() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    set_pool_size_locked(1);
+    auto i = std::min_element(entries_.begin(), entries_.end()) - entries_.begin();
+    return std::shared_ptr<ValueT>(entries_[i].value.get(),
+      [i, self = this->shared_from_this()](auto *ptr) {
+        self->dec_ref(i);
+      });
+  }
+
+  void set_pool_size(size_t size) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    set_pool_size_locked(size);
+  }
+
+  void dec_ref(size_t i) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    entries_.at(i).refcnt--;
+  }
+
+private:
+  std::mutex mutex_;
+  std::vector<Entry> entries_;
+
+  CreatorT creator_;
+
+  void set_pool_size_locked(size_t size) {
+    for (size_t i = entries_.size(); i < size; i++) {
+      entries_.emplace_back(Entry{creator_(i + 1), 0});
+    }
+  }
+};
+
+class ThreadsImpl : public Threads {
+  using Thread = std::unique_ptr<rtc::Thread>;
+public:
+  explicit ThreadsImpl(size_t i) {
+    auto suffix = i == 0 ? "" : "#" + std::to_string(i);
+    network_ = create_network("tgc-net" + suffix);
+    media_ = create("tgc-media" + suffix);
+    worker_ = create("tgc-work"  + suffix);
+  }
+
+  rtc::Thread *getNetworkThread() override {
+    return network_.get();
+  }
+  rtc::Thread *getMediaThread() override {
+    return media_.get();
+  }
+  rtc::Thread *getWorkerThread() override {
+    return worker_.get();
+  }
+
+private:
+  Thread network_;
+  Thread media_;
+  Thread worker_;
+
+  static Thread create(const std::string &name) {
+    return init(std::unique_ptr<rtc::Thread>(rtc::Thread::Create()), name);
+  }
+  static Thread create_network(const std::string &name) {
+    return init(std::unique_ptr<rtc::Thread>(rtc::Thread::CreateWithSocketServer()), name);
+  }
+
+  static Thread init(Thread value, const std::string &name) {
+    value->SetName(name, nullptr);
+    value->Start();
+    return value;
+  }
+};
+
+class ThreadsCreator {
+public:
+  std::unique_ptr<Threads> operator()(size_t i) {
+   return std::make_unique<ThreadsImpl>(i);
+  }
+};
+
+Pool<Threads, ThreadsCreator> &get_pool() {
+  static auto pool = std::make_shared<Pool<Threads, ThreadsCreator>>(ThreadsCreator());
+  return *pool;
+}
+
+void Threads::setPoolSize(size_t size){
+  get_pool().set_pool_size(size);
+}
+std::shared_ptr<Threads> Threads::getThreads(){
+  return get_pool().get();
+}
+
 namespace StaticThreads {
 
-static rtc::Thread *makeNetworkThread() {
-    static std::unique_ptr<rtc::Thread> value = rtc::Thread::CreateWithSocketServer();
-    value->SetName("WebRTC-Network", nullptr);
-    value->Start();
-    return value.get();
-}
-
 rtc::Thread *getNetworkThread() {
-    static rtc::Thread *value = makeNetworkThread();
-    return value;
-}
-
-static rtc::Thread *makeMediaThread() {
-    static std::unique_ptr<rtc::Thread> value = rtc::Thread::Create();
-    value->SetName("WebRTC-Media", nullptr);
-    value->Start();
-    return value.get();
+  return getThreads()->getNetworkThread();
 }
 
 rtc::Thread *getMediaThread() {
-    static rtc::Thread *value = makeMediaThread();
-    return value;
-}
-
-static rtc::Thread *makeWorkerThread() {
-    static std::unique_ptr<rtc::Thread> value = rtc::Thread::Create();
-    value->SetName("WebRTC-Worker", nullptr);
-    value->Start();
-    return value.get();
+  return getThreads()->getMediaThread();
 }
 
 rtc::Thread *getWorkerThread() {
-    static rtc::Thread *value = makeWorkerThread();
-    return value;
+  return getThreads()->getWorkerThread();
 }
 
+std::shared_ptr<Threads> &getThreads() {
+  static std::shared_ptr<Threads> threads = std::make_shared<ThreadsImpl>(0);
+  return threads;
+}
 };
 
 }

tgcalls/StaticThreads.h

@@ -1,3 +1,7 @@
+#pragma once
+
+#include <cstddef>
+#include <memory>
 
 namespace rtc {
 class Thread;
@@ -5,12 +9,23 @@ class Thread;
 
 namespace tgcalls {
 
-namespace StaticThreads {
+class Threads {
+public:
+  virtual ~Threads() = default;
+  virtual rtc::Thread *getNetworkThread() = 0;
+  virtual rtc::Thread *getMediaThread() = 0;
+  virtual rtc::Thread *getWorkerThread() = 0;
 
+  // it is not possible to decrease pool size
+  static void setPoolSize(size_t size);
+  static std::shared_ptr<Threads> getThreads();
+};
+
+namespace StaticThreads {
 rtc::Thread *getNetworkThread();
 rtc::Thread *getMediaThread();
 rtc::Thread *getWorkerThread();
-
+std::shared_ptr<Threads> &getThreads();
 }
 
 };

tgcalls/VideoCaptureInterface.cpp

@@ -4,8 +4,10 @@
 
 namespace tgcalls {
 
-std::unique_ptr<VideoCaptureInterface> VideoCaptureInterface::Create(std::string deviceId, std::shared_ptr<PlatformContext> platformContext) {
-	return std::make_unique<VideoCaptureInterfaceImpl>(deviceId, platformContext);
+std::unique_ptr<VideoCaptureInterface> VideoCaptureInterface::Create(
+   std::shared_ptr<Threads> threads, std::string deviceId,
+   std::shared_ptr<PlatformContext> platformContext) {
+	return std::make_unique<VideoCaptureInterfaceImpl>(deviceId, platformContext, std::move(threads));
 }
 
 VideoCaptureInterface::~VideoCaptureInterface() = default;

tgcalls/VideoCaptureInterface.h

@@ -16,6 +16,7 @@ class VideoFrame;
 namespace tgcalls {
 
 class PlatformContext;
+class Threads;
 
 enum class VideoState {
 	Inactive,
@@ -29,7 +30,8 @@ protected:
 
 public:
 	static std::unique_ptr<VideoCaptureInterface> Create(
-		std::string deviceId = std::string(),
+                std::shared_ptr<Threads> threads,
+                std::string deviceId = std::string(),
 		std::shared_ptr<PlatformContext> platformContext = nullptr);
 
 	virtual ~VideoCaptureInterface();

tgcalls/VideoCaptureInterfaceImpl.cpp

@@ -8,8 +8,8 @@
 
 namespace tgcalls {
 
-VideoCaptureInterfaceObject::VideoCaptureInterfaceObject(std::string deviceId, std::shared_ptr<PlatformContext> platformContext)
-: _videoSource(PlatformInterface::SharedInstance()->makeVideoSource(StaticThreads::getMediaThread(), StaticThreads::getWorkerThread())) {
+VideoCaptureInterfaceObject::VideoCaptureInterfaceObject(std::string deviceId, std::shared_ptr<PlatformContext> platformContext, Threads &threads)
+: _videoSource(PlatformInterface::SharedInstance()->makeVideoSource(threads.getMediaThread(), threads.getWorkerThread())) {
 	_platformContext = platformContext;
 
 	switchToDevice(deviceId);
@@ -103,9 +103,10 @@ void VideoCaptureInterfaceObject::setStateUpdated(std::function<void(VideoState)
 	_stateUpdated = stateUpdated;
 }
 
-VideoCaptureInterfaceImpl::VideoCaptureInterfaceImpl(std::string deviceId, std::shared_ptr<PlatformContext> platformContext) :
-_impl(StaticThreads::getMediaThread(), [deviceId, platformContext]() {
-	return new VideoCaptureInterfaceObject(deviceId, platformContext);
+VideoCaptureInterfaceImpl::VideoCaptureInterfaceImpl(std::string deviceId,
+   std::shared_ptr<PlatformContext> platformContext, std::shared_ptr<Threads> threads) :
+_impl(threads->getMediaThread(), [deviceId, platformContext, threads]() {
+	return new VideoCaptureInterfaceObject(deviceId, platformContext, *threads);
 }) {
 }
 

tgcalls/VideoCaptureInterfaceImpl.h

@@ -10,10 +10,11 @@
 namespace tgcalls {
 
 class VideoCapturerInterface;
+class Threads;
 
 class VideoCaptureInterfaceObject {
 public:
-	VideoCaptureInterfaceObject(std::string deviceId, std::shared_ptr<PlatformContext> platformContext);
+	VideoCaptureInterfaceObject(std::string deviceId, std::shared_ptr<PlatformContext> platformContext, Threads &threads);
 	~VideoCaptureInterfaceObject();
 
 	void switchToDevice(std::string deviceId);
@@ -39,7 +40,7 @@ private:
 
 class VideoCaptureInterfaceImpl : public VideoCaptureInterface {
 public:
-	VideoCaptureInterfaceImpl(std::string deviceId, std::shared_ptr<PlatformContext> platformContext);
+	VideoCaptureInterfaceImpl(std::string deviceId, std::shared_ptr<PlatformContext> platformContext, std::shared_ptr<Threads> threads);
 	virtual ~VideoCaptureInterfaceImpl();
 
 	void switchToDevice(std::string deviceId) override;

tgcalls/group/GroupInstanceCustomImpl.cpp

@@ -35,6 +35,7 @@
 #include "modules/audio_coding/neteq/default_neteq_factory.h"
 #include "modules/audio_coding/include/audio_coding_module.h"
 
+#include "AudioFrame.h"
 #include "ThreadLocalObject.h"
 #include "Manager.h"
 #include "NetworkManager.h"
@@ -195,6 +196,37 @@ struct VideoSsrcs {
     }
 };
 
+struct ChannelId {
+  uint32_t networkSsrc = 0;
+  uint32_t actualSsrc = 0;
+
+  ChannelId(uint32_t networkSsrc_, uint32_t actualSsrc_) :
+      networkSsrc(networkSsrc_),
+      actualSsrc(actualSsrc_) {
+  }
+
+  explicit ChannelId(uint32_t networkSsrc_) :
+      networkSsrc(networkSsrc_),
+      actualSsrc(networkSsrc_) {
+  }
+
+  bool operator <(const ChannelId& rhs) const {
+    if (networkSsrc != rhs.networkSsrc) {
+      return networkSsrc < rhs.networkSsrc;
+    }
+    return actualSsrc < rhs.actualSsrc;
+  }
+
+  std::string name() {
+    if (networkSsrc == actualSsrc) {
+      return uint32ToString(networkSsrc);
+    } else {
+      return uint32ToString(networkSsrc) + "to" + uint32ToString(actualSsrc);
+    }
+  }
+};
+
+
 class NetworkInterfaceImpl : public cricket::MediaChannel::NetworkInterface {
 public:
     NetworkInterfaceImpl(std::function<void(rtc::CopyOnWriteBuffer const *, rtc::SentPacket)> sendPacket) :
@@ -278,15 +310,27 @@ public:
     };
 
 public:
-    AudioSinkImpl(std::function<void(Update)> update) :
-    _update(update) {
+    AudioSinkImpl(std::function<void(Update)> update,
+        ChannelId channel_id, std::function<void(uint32_t, const AudioFrame &)> onAudioFrame) :
+    _update(update), _channel_id(channel_id), _onAudioFrame(std::move(onAudioFrame)) {
     }
 
     virtual ~AudioSinkImpl() {
     }
 
     virtual void OnData(const Data& audio) override {
-        if (audio.channels == 1) {
+      if (_onAudioFrame) {
+        AudioFrame frame;
+        frame.audio_samples = audio.data;
+        frame.num_samples = audio.samples_per_channel;
+        frame.bytes_per_sample = 2;
+        frame.num_channels = audio.channels;
+        frame.samples_per_sec = audio.sample_rate;
+        frame.elapsed_time_ms = 0;
+        frame.ntp_time_ms = 0;
+        _onAudioFrame(_channel_id.actualSsrc, frame);
+      }
+      if (audio.channels == 1) {
             const int16_t *samples = (const int16_t *)audio.data;
             int numberOfSamplesInFrame = (int)audio.samples_per_channel;
 
@@ -318,8 +362,10 @@ public:
 
 private:
     std::function<void(Update)> _update;
+    ChannelId _channel_id;
+    std::function<void(uint32_t, const AudioFrame &)> _onAudioFrame;
 
-    int _peakCount = 0;
+  int _peakCount = 0;
     uint16_t _peak = 0;
 
     CombinedVad _vad;
@@ -436,37 +482,6 @@ public:
 
     virtual ~AudioCaptureAnalyzer() = default;
 };
-
-struct ChannelId {
-    uint32_t networkSsrc = 0;
-    uint32_t actualSsrc = 0;
-    
-    ChannelId(uint32_t networkSsrc_, uint32_t actualSsrc_) :
-        networkSsrc(networkSsrc_),
-        actualSsrc(actualSsrc_) {
-    }
-    
-    explicit ChannelId(uint32_t networkSsrc_) :
-        networkSsrc(networkSsrc_),
-        actualSsrc(networkSsrc_) {
-    }
-    
-    bool operator <(const ChannelId& rhs) const {
-        if (networkSsrc != rhs.networkSsrc) {
-            return networkSsrc < rhs.networkSsrc;
-        }
-        return actualSsrc < rhs.actualSsrc;
-    }
-    
-    std::string name() {
-        if (networkSsrc == actualSsrc) {
-            return uint32ToString(networkSsrc);
-        } else {
-            return uint32ToString(networkSsrc) + "to" + uint32ToString(actualSsrc);
-        }
-    }
-};
-
 class IncomingAudioChannel : public sigslot::has_slots<> {
 public:
     IncomingAudioChannel(
@@ -476,7 +491,9 @@ public:
         rtc::UniqueRandomIdGenerator *randomIdGenerator,
         bool isRawPcm,
         ChannelId ssrc,
-        std::function<void(AudioSinkImpl::Update)> &&onAudioLevelUpdated) :
+        std::function<void(AudioSinkImpl::Update)> &&onAudioLevelUpdated,
+        std::function<void(uint32_t, const AudioFrame &)> onAudioFrame,
+        Threads &threads) :
     _ssrc(ssrc),
     _channelManager(channelManager),
     _call(call) {
@@ -490,7 +507,7 @@ public:
 
         std::string streamId = std::string("stream") + ssrc.name();
 
-        _audioChannel = _channelManager->CreateVoiceChannel(call, cricket::MediaConfig(), rtpTransport, StaticThreads::getMediaThread(), std::string("audio") + uint32ToString(ssrc.networkSsrc), false, GroupNetworkManager::getDefaulCryptoOptions(), randomIdGenerator, audioOptions);
+        _audioChannel = _channelManager->CreateVoiceChannel(call, cricket::MediaConfig(), rtpTransport, threads.getMediaThread(), std::string("audio") + uint32ToString(ssrc.networkSsrc), false, GroupNetworkManager::getDefaulCryptoOptions(), randomIdGenerator, audioOptions);
 
         const uint8_t opusMinBitrateKbps = 32;
         const uint8_t opusMaxBitrateKbps = 32;
@@ -540,7 +557,7 @@ public:
 
         std::unique_ptr<AudioSinkImpl> audioLevelSink(new AudioSinkImpl([onAudioLevelUpdated = std::move(onAudioLevelUpdated)](AudioSinkImpl::Update update) {
             onAudioLevelUpdated(update);
-        }));
+        }, _ssrc, std::move(onAudioFrame)));
         _audioChannel->media_channel()->SetRawAudioSink(ssrc.networkSsrc, std::move(audioLevelSink));
 
         _audioChannel->SignalSentPacket().connect(this, &IncomingAudioChannel::OnSentPacket_w);
@@ -592,7 +609,8 @@ public:
         webrtc::RtpTransport *rtpTransport,
         rtc::UniqueRandomIdGenerator *randomIdGenerator,
         std::vector<webrtc::SdpVideoFormat> const &availableVideoFormats,
-        GroupParticipantDescription const &description) :
+        GroupParticipantDescription const &description,
+        Threads &threads) :
     _channelManager(channelManager),
     _call(call) {
         _videoSink.reset(new VideoSinkImpl());
@@ -601,7 +619,7 @@ public:
 
         _videoBitrateAllocatorFactory = webrtc::CreateBuiltinVideoBitrateAllocatorFactory();
 
-        _videoChannel = _channelManager->CreateVideoChannel(call, cricket::MediaConfig(), rtpTransport, StaticThreads::getMediaThread(), std::string("video") + uint32ToString(description.audioSsrc), false, GroupNetworkManager::getDefaulCryptoOptions(), randomIdGenerator, cricket::VideoOptions(), _videoBitrateAllocatorFactory.get());
+        _videoChannel = _channelManager->CreateVideoChannel(call, cricket::MediaConfig(), rtpTransport, threads.getMediaThread(), std::string("video") + uint32ToString(description.audioSsrc), false, GroupNetworkManager::getDefaulCryptoOptions(), randomIdGenerator, cricket::VideoOptions(), _videoBitrateAllocatorFactory.get());
 
         auto payloadTypes = assignPayloadTypes(availableVideoFormats);
         if (!payloadTypes.has_value()) {
@@ -756,9 +774,11 @@ struct DecodedBroadcastPart {
 
 class GroupInstanceCustomInternal : public sigslot::has_slots<>, public std::enable_shared_from_this<GroupInstanceCustomInternal> {
 public:
-    GroupInstanceCustomInternal(GroupInstanceDescriptor &&descriptor) :
+    GroupInstanceCustomInternal(GroupInstanceDescriptor &&descriptor, std::shared_ptr<Threads> threads) :
+    _threads(std::move(threads)),
     _networkStateUpdated(descriptor.networkStateUpdated),
     _audioLevelsUpdated(descriptor.audioLevelsUpdated),
+    _onAudioFrame(descriptor.onAudioFrame),
     _incomingVideoSourcesUpdated(descriptor.incomingVideoSourcesUpdated),
     _participantDescriptionsRequired(descriptor.participantDescriptionsRequired),
     _requestBroadcastPart(descriptor.requestBroadcastPart),
@@ -767,7 +787,7 @@ public:
     _taskQueueFactory(webrtc::CreateDefaultTaskQueueFactory()),
 	_createAudioDeviceModule(descriptor.createAudioDeviceModule),
     _missingPacketBuffer(100) {
-        assert(StaticThreads::getMediaThread()->IsCurrent());
+        assert(_threads->getMediaThread()->IsCurrent());
 
         auto generator = std::mt19937(std::random_device()());
         auto distribution = std::uniform_int_distribution<uint32_t>();
@@ -814,10 +834,10 @@ public:
             "WebRTC-Audio-OpusMinPacketLossRate/Enabled-1/"
         );
 
-        _networkManager.reset(new ThreadLocalObject<GroupNetworkManager>(StaticThreads::getNetworkThread(), [weak] () mutable {
+        _networkManager.reset(new ThreadLocalObject<GroupNetworkManager>(_threads->getNetworkThread(), [weak, threads = _threads] () mutable {
             return new GroupNetworkManager(
                 [=](const GroupNetworkManager::State &state) {
-                    StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [=] {
+                    threads->getMediaThread()->PostTask(RTC_FROM_HERE, [=] {
                         const auto strong = weak.lock();
                         if (!strong) {
                             return;
@@ -826,32 +846,32 @@ public:
                     });
                 },
                 [=](rtc::CopyOnWriteBuffer const &message, bool isUnresolved) {
-                    StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, message, isUnresolved]() mutable {
+                    threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, message, isUnresolved]() mutable {
                         if (const auto strong = weak.lock()) {
                             strong->receivePacket(message, isUnresolved);
                         }
                     });
                 },
                 [=](rtc::CopyOnWriteBuffer const &message, int64_t timestamp) {
-                    StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, message, timestamp]() mutable {
+                    threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, message, timestamp]() mutable {
                         if (const auto strong = weak.lock()) {
                             strong->receiveRtcpPacket(message, timestamp);
                         }
                     });
                 },
                 [=](bool isDataChannelOpen) {
-                    StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, isDataChannelOpen]() mutable {
+                    threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, isDataChannelOpen]() mutable {
                         if (const auto strong = weak.lock()) {
                             strong->updateIsDataChannelOpen(isDataChannelOpen);
                         }
                     });
                 },
                 [=](std::string const &message) {
-                    StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, message]() mutable {
+                    threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, message]() mutable {
                         if (const auto strong = weak.lock()) {
                         }
                     });
-                });
+                }, threads);
         }));
 
         PlatformInterface::SharedInstance()->configurePlatformAudio();
@@ -864,8 +884,8 @@ public:
         mediaDeps.video_encoder_factory = PlatformInterface::SharedInstance()->makeVideoEncoderFactory();
         mediaDeps.video_decoder_factory = PlatformInterface::SharedInstance()->makeVideoDecoderFactory();
 
-        auto analyzer = new AudioCaptureAnalyzer([weak](GroupLevelValue const &level) {
-            StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, level](){
+        auto analyzer = new AudioCaptureAnalyzer([weak, threads = _threads](GroupLevelValue const &level) {
+            threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, level](){
                 auto strong = weak.lock();
                 if (!strong) {
                     return;
@@ -889,7 +909,7 @@ public:
 
         std::unique_ptr<cricket::MediaEngineInterface> mediaEngine = cricket::CreateMediaEngine(std::move(mediaDeps));
 
-        _channelManager.reset(new cricket::ChannelManager(std::move(mediaEngine), std::make_unique<cricket::RtpDataEngine>(), StaticThreads::getMediaThread(), StaticThreads::getNetworkThread()));
+        _channelManager.reset(new cricket::ChannelManager(std::move(mediaEngine), std::make_unique<cricket::RtpDataEngine>(), _threads->getMediaThread(), _threads->getNetworkThread()));
         _channelManager->Init();
 
         webrtc::Call::Config callConfig(_eventLog.get());
@@ -900,7 +920,7 @@ public:
 
         _uniqueRandomIdGenerator.reset(new rtc::UniqueRandomIdGenerator());
 
-        StaticThreads::getNetworkThread()->Invoke<void>(RTC_FROM_HERE, [this]() {
+        _threads->getNetworkThread()->Invoke<void>(RTC_FROM_HERE, [this]() {
             _rtpTransport = _networkManager->getSyncAssumingSameThread()->getRtpTransport();
         });
         
@@ -943,7 +963,7 @@ public:
         std::vector<std::string> streamIds;
         streamIds.push_back("1");
         
-        _outgoingAudioChannel = _channelManager->CreateVoiceChannel(_call.get(), cricket::MediaConfig(), _rtpTransport, StaticThreads::getMediaThread(), "0", false, GroupNetworkManager::getDefaulCryptoOptions(), _uniqueRandomIdGenerator.get(), audioOptions);
+        _outgoingAudioChannel = _channelManager->CreateVoiceChannel(_call.get(), cricket::MediaConfig(), _rtpTransport, _threads->getMediaThread(), "0", false, GroupNetworkManager::getDefaulCryptoOptions(), _uniqueRandomIdGenerator.get(), audioOptions);
 
         const uint8_t opusMinBitrateKbps = 32;
         const uint8_t opusMaxBitrateKbps = 32;
@@ -992,7 +1012,7 @@ public:
 
     void beginLevelsTimer(int timeoutMs) {
         const auto weak = std::weak_ptr<GroupInstanceCustomInternal>(shared_from_this());
-        StaticThreads::getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
+        _threads->getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
             auto strong = weak.lock();
             if (!strong) {
                 return;
@@ -1032,7 +1052,7 @@ public:
     
     void beginNetworkStatusTimer(int delayMs) {
         const auto weak = std::weak_ptr<GroupInstanceCustomInternal>(shared_from_this());
-        StaticThreads::getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
+        _threads->getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
             auto strong = weak.lock();
             if (!strong) {
                 return;
@@ -1184,8 +1204,8 @@ public:
     
     void requestNextBroadcastPart() {
         const auto weak = std::weak_ptr<GroupInstanceCustomInternal>(shared_from_this());
-        _currentRequestedBroadcastTask = _requestBroadcastPart(_nextBroadcastTimestampMilliseconds, _broadcastPartDurationMilliseconds, [weak](BroadcastPart &&part) {
-            StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, part = std::move(part)]() mutable {
+        _currentRequestedBroadcastTask = _requestBroadcastPart(_nextBroadcastTimestampMilliseconds, _broadcastPartDurationMilliseconds, [weak, threads = _threads](BroadcastPart &&part) {
+            threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, part = std::move(part)]() mutable {
                 auto strong = weak.lock();
                 if (!strong) {
                     return;
@@ -1198,7 +1218,7 @@ public:
     
     void requestNextBroadcastPartWithDelay(int timeoutMs) {
         const auto weak = std::weak_ptr<GroupInstanceCustomInternal>(shared_from_this());
-        StaticThreads::getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
+        _threads->getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
             auto strong = weak.lock();
             if (!strong) {
                 return;
@@ -1256,7 +1276,7 @@ public:
     
     void beginBroadcastPartsDecodeTimer(int timeoutMs) {
         const auto weak = std::weak_ptr<GroupInstanceCustomInternal>(shared_from_this());
-        StaticThreads::getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
+        _threads->getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
             auto strong = weak.lock();
             if (!strong) {
                 return;
@@ -1579,7 +1599,7 @@ public:
                 _isUnknownSsrcsScheduled = true;
 
                 const auto weak = std::weak_ptr<GroupInstanceCustomInternal>(shared_from_this());
-                StaticThreads::getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
+                _threads->getMediaThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
                     auto strong = weak.lock();
                     if (!strong) {
                         return;
@@ -1934,8 +1954,8 @@ public:
             _uniqueRandomIdGenerator.get(),
             isRawPcm,
             ssrc,
-            [weak, ssrc = ssrc](AudioSinkImpl::Update update) {
-                StaticThreads::getMediaThread()->PostTask(RTC_FROM_HERE, [weak, ssrc, update]() {
+            [weak, ssrc = ssrc, threads = _threads](AudioSinkImpl::Update update) {
+                threads->getMediaThread()->PostTask(RTC_FROM_HERE, [weak, ssrc, update]() {
                     auto strong = weak.lock();
                     if (!strong) {
                         return;
@@ -1945,7 +1965,9 @@ public:
                     mappedUpdate.voice = update.hasSpeech;
                     strong->_audioLevels[ssrc] = mappedUpdate;
                 });
-            }
+            },
+            _onAudioFrame,
+            *_threads
         ));
         
         auto volume = _volumeBySsrc.find(ssrc.actualSsrc);
@@ -1977,7 +1999,8 @@ public:
             _rtpTransport,
             _uniqueRandomIdGenerator.get(),
             _availableVideoFormats,
-            participant
+            participant,
+            *_threads
         ));
         _incomingVideoChannels.insert(std::make_pair(participant.audioSsrc, std::move(channel)));
 
@@ -2063,10 +2086,12 @@ private:
     }
 
 private:
+    std::shared_ptr<Threads> _threads;
     GroupConnectionMode _connectionMode = GroupConnectionMode::GroupConnectionModeNone;
     
     std::function<void(GroupNetworkState)> _networkStateUpdated;
     std::function<void(GroupLevelsUpdate const &)> _audioLevelsUpdated;
+    std::function<void(uint32_t, const AudioFrame &)> _onAudioFrame;
     std::function<void(std::vector<uint32_t> const &)> _incomingVideoSourcesUpdated;
     std::function<void(std::vector<uint32_t> const &)> _participantDescriptionsRequired;
     std::function<std::shared_ptr<BroadcastPartTask>(int64_t, int64_t, std::function<void(BroadcastPart &&)>)> _requestBroadcastPart;
@@ -2144,8 +2169,9 @@ GroupInstanceCustomImpl::GroupInstanceCustomImpl(GroupInstanceDescriptor &&descr
         rtc::LogMessage::AddLogToStream(_logSink.get(), rtc::LS_INFO);
     }
 
-    _internal.reset(new ThreadLocalObject<GroupInstanceCustomInternal>(StaticThreads::getMediaThread(), [descriptor = std::move(descriptor)]() mutable {
-        return new GroupInstanceCustomInternal(std::move(descriptor));
+    _threads = Threads::getThreads();
+    _internal.reset(new ThreadLocalObject<GroupInstanceCustomInternal>(_threads->getMediaThread(), [descriptor = std::move(descriptor), threads = _threads]() mutable {
+        return new GroupInstanceCustomInternal(std::move(descriptor), threads);
     }));
     _internal->perform(RTC_FROM_HERE, [](GroupInstanceCustomInternal *internal) {
         internal->start();
@@ -2159,7 +2185,7 @@ GroupInstanceCustomImpl::~GroupInstanceCustomImpl() {
     _internal.reset();
 
     // Wait until _internal is destroyed
-    StaticThreads::getMediaThread()->Invoke<void>(RTC_FROM_HERE, [] {});
+    _threads->getMediaThread()->Invoke<void>(RTC_FROM_HERE, [] {});
 }
 
 void GroupInstanceCustomImpl::stop() {

tgcalls/group/GroupInstanceCustomImpl.h

@@ -14,6 +14,7 @@ namespace tgcalls {
 
 class LogSinkImpl;
 class GroupInstanceCustomInternal;
+class Threads;
 
 class GroupInstanceCustomImpl final : public GroupInstanceInterface {
 public:
@@ -40,6 +41,7 @@ public:
     void setFullSizeVideoSsrc(uint32_t ssrc);
 
 private:
+    std::shared_ptr<Threads> _threads;
     std::unique_ptr<ThreadLocalObject<GroupInstanceCustomInternal>> _internal;
     std::unique_ptr<LogSinkImpl> _logSink;
 

tgcalls/group/GroupInstanceImpl.h

@@ -23,6 +23,7 @@ namespace tgcalls {
 
 class LogSinkImpl;
 class GroupInstanceManager;
+class AudioFrame;
 
 struct GroupConfig {
     FilePath logPath;
@@ -77,6 +78,7 @@ struct GroupInstanceDescriptor {
     GroupConfig config;
     std::function<void(GroupNetworkState)> networkStateUpdated;
     std::function<void(GroupLevelsUpdate const &)> audioLevelsUpdated;
+    std::function<void(uint32_t, const AudioFrame &)> onAudioFrame;
     std::string initialInputDeviceId;
     std::string initialOutputDeviceId;
     bool debugIgnoreMissingSsrcs = false;

tgcalls/group/GroupNetworkManager.cpp

@@ -41,13 +41,15 @@ public:
     SctpDataChannelProviderInterfaceImpl(
         cricket::DtlsTransport *transportChannel,
         std::function<void(bool)> onStateChanged,
-        std::function<void(std::string const &)> onMessageReceived
+        std::function<void(std::string const &)> onMessageReceived,
+        std::shared_ptr<Threads> threads
     ) :
+    _threads(std::move(threads)),
     _onStateChanged(onStateChanged),
     _onMessageReceived(onMessageReceived) {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
-        _sctpTransportFactory.reset(new cricket::SctpTransportFactory(StaticThreads::getNetworkThread()));
+        _sctpTransportFactory.reset(new cricket::SctpTransportFactory(_threads->getNetworkThread()));
 
         _sctpTransport = _sctpTransportFactory->CreateSctpTransport(transportChannel);
         _sctpTransport->SignalReadyToSendData.connect(this, &SctpDataChannelProviderInterfaceImpl::sctpReadyToSendData);
@@ -59,15 +61,15 @@ public:
             this,
             "data",
             dataChannelInit,
-            StaticThreads::getNetworkThread(),
-            StaticThreads::getNetworkThread()
+            _threads->getNetworkThread(),
+            _threads->getNetworkThread()
         );
 
         _dataChannel->RegisterObserver(this);
     }
 
     virtual ~SctpDataChannelProviderInterfaceImpl() {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         _dataChannel->UnregisterObserver();
         _dataChannel->Close();
@@ -78,7 +80,7 @@ public:
     }
 
     void sendDataChannelMessage(std::string const &message) {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         if (_isDataChannelOpen) {
             RTC_LOG(LS_INFO) << "Outgoing DataChannel message: " << message;
@@ -91,7 +93,7 @@ public:
     }
 
     virtual void OnStateChange() override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         auto state = _dataChannel->state();
         bool isDataChannelOpen = state == webrtc::DataChannelInterface::DataState::kOpen;
@@ -102,7 +104,7 @@ public:
     }
 
     virtual void OnMessage(const webrtc::DataBuffer& buffer) override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         if (!buffer.binary) {
             std::string messageText(buffer.data.data(), buffer.data.data() + buffer.data.size());
@@ -113,7 +115,7 @@ public:
     }
 
     void updateIsConnected(bool isConnected) {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         if (isConnected) {
             if (!_isSctpTransportStarted) {
@@ -124,56 +126,57 @@ public:
     }
 
     void sctpReadyToSendData() {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         _dataChannel->OnTransportReady(true);
     }
 
     void sctpDataReceived(const cricket::ReceiveDataParams& params, const rtc::CopyOnWriteBuffer& buffer) {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         _dataChannel->OnDataReceived(params, buffer);
     }
 
     virtual bool SendData(const cricket::SendDataParams& params, const rtc::CopyOnWriteBuffer& payload, cricket::SendDataResult* result) override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         return _sctpTransport->SendData(params, payload);
     }
 
     virtual bool ConnectDataChannel(webrtc::SctpDataChannel *data_channel) override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         return true;
     }
 
     virtual void DisconnectDataChannel(webrtc::SctpDataChannel* data_channel) override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         return;
     }
 
     virtual void AddSctpDataStream(int sid) override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+      assert(_threads->getNetworkThread()->IsCurrent());
 
         _sctpTransport->OpenStream(sid);
     }
 
     virtual void RemoveSctpDataStream(int sid) override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
-        StaticThreads::getNetworkThread()->Invoke<void>(RTC_FROM_HERE, [this, sid]() {
+        _threads->getNetworkThread()->Invoke<void>(RTC_FROM_HERE, [this, sid]() {
             _sctpTransport->ResetStream(sid);
         });
     }
 
     virtual bool ReadyToSendData() const override {
-        assert(StaticThreads::getNetworkThread()->IsCurrent());
+        assert(_threads->getNetworkThread()->IsCurrent());
 
         return _sctpTransport->ReadyToSendData();
     }
 
 private:
+    std::shared_ptr<Threads> _threads;
     std::function<void(bool)> _onStateChanged;
     std::function<void(std::string const &)> _onMessageReceived;
 
@@ -198,19 +201,21 @@ GroupNetworkManager::GroupNetworkManager(
     std::function<void(rtc::CopyOnWriteBuffer const &, bool)> transportMessageReceived,
     std::function<void(rtc::CopyOnWriteBuffer const &, int64_t)> rtcpPacketReceived,
     std::function<void(bool)> dataChannelStateUpdated,
-    std::function<void(std::string const &)> dataChannelMessageReceived) :
+    std::function<void(std::string const &)> dataChannelMessageReceived,
+    std::shared_ptr<Threads> threads) :
+_threads(std::move(threads)),
 _stateUpdated(std::move(stateUpdated)),
 _transportMessageReceived(std::move(transportMessageReceived)),
 _rtcpPacketReceived(std::move(rtcpPacketReceived)),
 _dataChannelStateUpdated(dataChannelStateUpdated),
 _dataChannelMessageReceived(dataChannelMessageReceived) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
     
     _localIceParameters = PeerIceParameters(rtc::CreateRandomString(cricket::ICE_UFRAG_LENGTH), rtc::CreateRandomString(cricket::ICE_PWD_LENGTH));
     
     _localCertificate = rtc::RTCCertificateGenerator::GenerateCertificate(rtc::KeyParams(rtc::KT_ECDSA), absl::nullopt);
     
-    _socketFactory.reset(new rtc::BasicPacketSocketFactory(StaticThreads::getNetworkThread()));
+    _socketFactory.reset(new rtc::BasicPacketSocketFactory(_threads->getNetworkThread()));
     _networkManager = std::make_unique<rtc::BasicNetworkManager>();
     _asyncResolverFactory = std::make_unique<webrtc::BasicAsyncResolverFactory>();
     
@@ -226,7 +231,7 @@ _dataChannelMessageReceived(dataChannelMessageReceived) {
 }
 
 GroupNetworkManager::~GroupNetworkManager() {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 
     RTC_LOG(LS_INFO) << "GroupNetworkManager::~GroupNetworkManager()";
 
@@ -372,7 +377,7 @@ webrtc::RtpTransport *GroupNetworkManager::getRtpTransport() {
 
 void GroupNetworkManager::checkConnectionTimeout() {
     const auto weak = std::weak_ptr<GroupNetworkManager>(shared_from_this());
-    StaticThreads::getNetworkThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
+    _threads->getNetworkThread()->PostDelayedTask(RTC_FROM_HERE, [weak]() {
         auto strong = weak.lock();
         if (!strong) {
             return;
@@ -393,26 +398,26 @@ void GroupNetworkManager::checkConnectionTimeout() {
 }
 
 void GroupNetworkManager::candidateGathered(cricket::IceTransportInternal *transport, const cricket::Candidate &candidate) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 }
 
 void GroupNetworkManager::candidateGatheringState(cricket::IceTransportInternal *transport) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 }
 
 void GroupNetworkManager::OnTransportWritableState_n(rtc::PacketTransportInternal *transport) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 
     UpdateAggregateStates_n();
 }
 void GroupNetworkManager::OnTransportReceivingState_n(rtc::PacketTransportInternal *transport) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 
     UpdateAggregateStates_n();
 }
 
 void GroupNetworkManager::OnDtlsHandshakeError(rtc::SSLHandshakeError error) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 }
 
 void GroupNetworkManager::DtlsStateChanged() {
@@ -420,7 +425,7 @@ void GroupNetworkManager::DtlsStateChanged() {
 
     if (_dtlsTransport->IsDtlsActive()) {
         const auto weak = std::weak_ptr<GroupNetworkManager>(shared_from_this());
-        StaticThreads::getNetworkThread()->PostTask(RTC_FROM_HERE, [weak]() {
+        _threads->getNetworkThread()->PostTask(RTC_FROM_HERE, [weak]() {
             const auto strong = weak.lock();
             if (!strong) {
                 return;
@@ -435,7 +440,7 @@ void GroupNetworkManager::DtlsReadyToSend(bool isReadyToSend) {
 
     if (isReadyToSend) {
         const auto weak = std::weak_ptr<GroupNetworkManager>(shared_from_this());
-        StaticThreads::getNetworkThread()->PostTask(RTC_FROM_HERE, [weak]() {
+        _threads->getNetworkThread()->PostTask(RTC_FROM_HERE, [weak]() {
             const auto strong = weak.lock();
             if (!strong) {
                 return;
@@ -450,11 +455,11 @@ void GroupNetworkManager::transportStateChanged(cricket::IceTransportInternal *t
 }
 
 void GroupNetworkManager::transportReadyToSend(cricket::IceTransportInternal *transport) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 }
 
 void GroupNetworkManager::transportPacketReceived(rtc::PacketTransportInternal *transport, const char *bytes, size_t size, const int64_t &timestamp, int unused) {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 
     _lastNetworkActivityMs = rtc::TimeMillis();
 }
@@ -472,7 +477,7 @@ void GroupNetworkManager::OnRtcpPacketReceived_n(rtc::CopyOnWriteBuffer *packet,
 }
 
 void GroupNetworkManager::UpdateAggregateStates_n() {
-    assert(StaticThreads::getNetworkThread()->IsCurrent());
+    assert(_threads->getNetworkThread()->IsCurrent());
 
     auto state = _transportChannel->GetIceTransportState();
     bool isConnected = false;

tgcalls/group/GroupNetworkManager.h

@@ -44,6 +44,7 @@ namespace tgcalls {
 
 struct Message;
 class SctpDataChannelProviderInterfaceImpl;
+class Threads;
 
 class GroupNetworkManager : public sigslot::has_slots<>, public std::enable_shared_from_this<GroupNetworkManager> {
 public:
@@ -59,7 +60,8 @@ public:
         std::function<void(rtc::CopyOnWriteBuffer const &, bool)> transportMessageReceived,
         std::function<void(rtc::CopyOnWriteBuffer const &, int64_t)> rtcpPacketReceived,
         std::function<void(bool)> dataChannelStateUpdated,
-        std::function<void(std::string const &)> dataChannelMessageReceived);
+        std::function<void(std::string const &)> dataChannelMessageReceived,
+        std::shared_ptr<Threads> threads);
     ~GroupNetworkManager();
 
     void start();
@@ -93,6 +95,7 @@ private:
     void sctpReadyToSendData();
     void sctpDataReceived(const cricket::ReceiveDataParams& params, const rtc::CopyOnWriteBuffer& buffer);
 
+    std::shared_ptr<Threads> _threads;
     std::function<void(const GroupNetworkManager::State &)> _stateUpdated;
     std::function<void(rtc::CopyOnWriteBuffer const &, bool)> _transportMessageReceived;
     std::function<void(rtc::CopyOnWriteBuffer const &, int64_t)> _rtcpPacketReceived;

tgcalls/platform/fake/FakeInterface.cpp

@@ -0,0 +1,43 @@
+#include "FakeInterface.h"
+
+#include "api/video_codecs/builtin_video_encoder_factory.h"
+#include "api/video_codecs/builtin_video_decoder_factory.h"
+#include "api/video_track_source_proxy.h"
+
+namespace tgcalls {
+
+std::unique_ptr<webrtc::VideoEncoderFactory> FakeInterface::makeVideoEncoderFactory() {
+  return webrtc::CreateBuiltinVideoEncoderFactory();
+}
+
+std::unique_ptr<webrtc::VideoDecoderFactory> FakeInterface::makeVideoDecoderFactory() {
+  return webrtc::CreateBuiltinVideoDecoderFactory();
+}
+
+rtc::scoped_refptr<webrtc::VideoTrackSourceInterface> FakeInterface::makeVideoSource(rtc::Thread *signalingThread,
+                                                                                     rtc::Thread *workerThread) {
+  return nullptr;
+}
+
+bool FakeInterface::supportsEncoding(const std::string &codecName) {
+  return false;
+  //return (codecName == cricket::kH264CodecName) || (codecName == cricket::kVp8CodecName);
+}
+
+void FakeInterface::adaptVideoSource(rtc::scoped_refptr<webrtc::VideoTrackSourceInterface> videoSource, int width,
+                                     int height, int fps) {
+}
+
+std::unique_ptr<VideoCapturerInterface> FakeInterface::makeVideoCapturer(
+    rtc::scoped_refptr<webrtc::VideoTrackSourceInterface> source, std::string deviceId,
+    std::function<void(VideoState)> stateUpdated, std::function<void(PlatformCaptureInfo)> captureInfoUpdated,
+    std::shared_ptr<PlatformContext> platformContext, std::pair<int, int> &outResolution) {
+  return nullptr;
+  //return std::make_unique<VideoCapturerInterfaceImpl>(source, deviceId, stateUpdated, outResolution);
+}
+
+std::unique_ptr<PlatformInterface> CreatePlatformInterface() {
+  return std::make_unique<FakeInterface>();
+}
+
+}  // namespace tgcalls

tgcalls/platform/fake/FakeInterface.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#include "platform/PlatformInterface.h"
+#include "VideoCapturerInterface.h"
+
+namespace tgcalls {
+
+class FakeInterface : public PlatformInterface {
+ public:
+  std::unique_ptr<webrtc::VideoEncoderFactory> makeVideoEncoderFactory() override;
+  std::unique_ptr<webrtc::VideoDecoderFactory> makeVideoDecoderFactory() override;
+  bool supportsEncoding(const std::string &codecName) override;
+  rtc::scoped_refptr<webrtc::VideoTrackSourceInterface> makeVideoSource(rtc::Thread *signalingThread,
+                                                                        rtc::Thread *workerThread) override;
+  void adaptVideoSource(rtc::scoped_refptr<webrtc::VideoTrackSourceInterface> videoSource, int width, int height,
+                        int fps) override;
+  std::unique_ptr<VideoCapturerInterface> makeVideoCapturer(
+      rtc::scoped_refptr<webrtc::VideoTrackSourceInterface> source, std::string deviceId,
+      std::function<void(VideoState)> stateUpdated, std::function<void(PlatformCaptureInfo)> captureInfoUpdated,
+      std::shared_ptr<PlatformContext> platformContext, std::pair<int, int> &outResolution) override;
+};
+
+}  // namespace tgcalls
+