AudioStream: prevent callback after destruction

Use a set of ongoing ops to cancel them when destroying.

GitLab: #888
Change-Id: Iff043d9e5fdfc8bd76aac0eb9eac2f2ed3520814

src/media/audio/pulseaudio/audiostream.cpp

@@ -24,8 +24,11 @@
 #include "logger.h"
 #include "compiler_intrinsics.h"
 
+#include <string_view>
 #include <stdexcept>
 
+using namespace std::literals;
+
 namespace jami {
 
 AudioStream::AudioStream(pa_context* c,
@@ -177,13 +180,19 @@ AudioStream::stop()
         destroyStream(audiostream_);
     }
     audiostream_ = nullptr;
+
+    std::unique_lock<std::mutex> lock(mutex_);
+    for (auto op : ongoing_ops)
+        pa_operation_cancel(op);
+    // wait for all operations to end
+    cond_.wait(lock, [this]{ return ongoing_ops.empty(); });
 }
 
 void
 AudioStream::moved(pa_stream* s)
 {
     audiostream_ = s;
-    JAMI_DBG("[audiostream] Stream moved: %d, %s",
+    JAMI_LOG("[audiostream] Stream moved: {:d}, {:s}",
              pa_stream_get_index(s),
              pa_stream_get_device_name(s));
 
@@ -195,37 +204,48 @@ AudioStream::moved(pa_stream* s)
             return;
         }
 
-        pa_context* context = pa_stream_get_context(s);
         auto* op = pa_context_get_source_info_by_name(
-            context,
+            pa_stream_get_context(s),
             name,
             [](pa_context* /*c*/, const pa_source_info* i, int /*eol*/, void* userdata) {
                 AudioStream* thisPtr = (AudioStream*) userdata;
-                // this whole closure gets called twice by pulse for some reason
+                // this closure gets called twice by pulse for some reason
                 // the 2nd time, i is invalid
                 if (!i) {
-                    // JAMI_ERR("[audiostream] source info not found for %s", realName);
+                    // JAMI_ERROR("[audiostream] source info not found");
+                    return;
+                }
+                if (!thisPtr) {
+                    JAMI_ERROR("[audiostream] AudioStream pointer became invalid during pa_source_info_cb_t callback!");
                     return;
                 }
 
                 // string compare
-                bool usingEchoCancel = std::string_view(i->driver) == "module-echo-cancel.c";
-                JAMI_WARN("[audiostream] capture stream using pulse echo cancel module? %s (%s)",
+                bool usingEchoCancel = std::string_view(i->driver) == "module-echo-cancel.c"sv;
+                JAMI_WARNING("[audiostream] capture stream using pulse echo cancel module? {} ({})",
                           usingEchoCancel ? "yes" : "no",
                           i->name);
-                if (!thisPtr) {
-                    JAMI_ERR("[audiostream] AudioStream pointer became invalid during "
-                             "pa_source_info_cb_t callback!");
-                    return;
-                }
                 thisPtr->echoCancelCb(usingEchoCancel);
             },
             this);
 
-        pa_operation_unref(op);
+        std::lock_guard<std::mutex> lock(mutex_);
+        pa_operation_set_state_callback(op, [](pa_operation *op, void *userdata){
+            static_cast<AudioStream*>(userdata)->opEnded(op);
+        }, this);
+        ongoing_ops.emplace(op);
     }
 }
 
+void
+AudioStream::opEnded(pa_operation* op)
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+    ongoing_ops.erase(op);
+    pa_operation_unref(op);
+    cond_.notify_all();
+}
+
 void
 AudioStream::stateChanged(pa_stream* s)
 {

src/media/audio/pulseaudio/audiostream.h

@@ -102,6 +102,7 @@ private:
      */
     void stateChanged(pa_stream* s);
     void moved(pa_stream* s);
+    void opEnded(pa_operation* s);
 
     /**
      * The pulse audio object
@@ -122,6 +123,10 @@ private:
      * Function called whenever the stream is moved and we check for an echo canceller
      */
     std::function<void(bool)> echoCancelCb;
+
+    std::mutex mutex_;
+    std::condition_variable cond_;
+    std::set<pa_operation*> ongoing_ops;
 };
 
 } // namespace jami