Add ClosableStream interface and pending stream helpers

Source/OBS.cpp

@@ -823,6 +823,8 @@ OBS::~OBS()
 
     OSTerminateThread(hHotkeyThread, 2500);
 
+    ClosePendingStreams();
+
     for(UINT i=0; i<plugins.Num(); i++)
     {
         PluginInfo &pluginInfo = plugins[i];

Source/OBS.h

@@ -17,6 +17,7 @@
 ********************************************************************************/
 
 #include <functional>
+#include <list>
 
 #pragma once
 
@@ -86,6 +87,13 @@ bool GetDefaultSpeakerID(String &strVal);
 
 //-------------------------------------------------------------------
 
+struct ClosableStream
+{
+    virtual ~ClosableStream() {}
+};
+
+//-------------------------------------------------------------------
+
 struct DataPacket
 {
     LPBYTE lpPacket;
@@ -1263,6 +1271,18 @@ public:
 
     inline void ResetMic() {if (bRunning && micAudio) ResetWASAPIAudioDevice(micAudio);}
     void GetThreadHandles (HANDLE *videoThread, HANDLE *encodeThread);
+
+    struct PendingStreams
+    {
+        using thread_t = std::unique_ptr<void, ThreadDeleter<>>;
+        std::list<thread_t> streams;
+        std::unique_ptr<void, MutexDeleter> mutex;
+        PendingStreams() : mutex(OSCreateMutex()) {}
+    } pendingStreams;
+
+    void AddPendingStream(ClosableStream *stream, std::function<void()> finishedCallback = {});
+    void AddPendingStreamThread(HANDLE thread);
+    void ClosePendingStreams();
 };
 
 LONG CALLBACK OBSExceptionHandler (PEXCEPTION_POINTERS exceptionInfo);

Source/OBSCapture.cpp

@@ -1509,5 +1509,61 @@ void OBS::RequestKeyframe(int waitTime)
     keyframeWait = waitTime;
 }
 
+void OBS::AddPendingStream(ClosableStream *stream, std::function<void()> finishedCallback)
+{
+    using namespace std;
+    struct args_t
+    {
+        using stream_t = remove_pointer_t<decltype(stream)>;
+        unique_ptr<stream_t> stream;
+        decltype(finishedCallback) finishedCallback;
+        args_t(stream_t *stream, decltype(finishedCallback) finishedCallback) : stream(stream), finishedCallback(move(finishedCallback)) {}
+    };
 
+    auto args = make_unique<args_t>(stream, move(finishedCallback));
 
+    ScopedLock l(pendingStreams.mutex);
+    pendingStreams.streams.emplace_back(OSCreateThread([](void *arg) -> DWORD
+    {
+        unique_ptr<args_t> args(static_cast<args_t*>(arg));
+        args->stream.reset();
+        if (args->finishedCallback)
+            args->finishedCallback();
+        return 0;
+    }, args.release()));
+}
+
+void OBS::AddPendingStreamThread(HANDLE thread)
+{
+    ScopedLock l(pendingStreams.mutex);
+    pendingStreams.streams.emplace_back(thread);
+}
+
+void OBS::ClosePendingStreams()
+{
+    ScopedLock l(pendingStreams.mutex);
+    if (pendingStreams.streams.empty())
+        return;
+
+    using namespace std;
+    vector<HANDLE> handles;
+    for (auto &pendingStream : pendingStreams.streams)
+        handles.push_back(pendingStream.get());
+
+    if (WaitForMultipleObjects(handles.size(), handles.data(), true, 5) != WAIT_OBJECT_0)
+    {
+        using ::locale;
+        int res = IDNO;
+        do
+        {
+            auto res = OBSMessageBox(hwndMain, Str("StreamClosePending"), nullptr, MB_YESNO | MB_ICONEXCLAMATION);
+
+            if (res != IDYES)
+                return;
+
+            if (WaitForMultipleObjects(handles.size(), handles.data(), true, 15 * 1000) == WAIT_OBJECT_0)
+                return;
+
+        } while (res == IDYES);
+    }
+}

rundir/locale/en.txt

@@ -43,6 +43,7 @@ LogWindow="Log Window"
 StreamReport="Stream Report"
 MessageBoxWarningCaption="Warning"
 NoSourcesFound="You haven't added any sources! Are you sure you want to stream a black screen?"
+StreamClosePending="Stream or file output unfinished, closing OBS may cause the stream to end prematurely or the file to be corrupted. Do you want to wait another 15 seconds for the output to finish?"
 
 ImportGlobalSourceNameExists="The global source '$1' already exists in the current scene collection."
 ImportGlobalSources="Import Global Sources"