Add ClosableStream support to ReplayBuffer

Source/OBS.cpp

@@ -816,8 +816,7 @@ OBS::OBS()
 
 OBS::~OBS()
 {
-    Stop(true);
+    Stop(true, true);
-    StopReplayBuffer();
 
     bShuttingDown = true;
 

Source/OBS.h

@@ -850,6 +850,7 @@ private:
 
     std::unique_ptr<VideoFileStream> replayBufferStream;
     ReplayBuffer *replayBuffer;
+    StopInfo replayBufferStop;
 
     bool bRequestKeyframe;
     int  keyframeWait;
@@ -1017,11 +1018,11 @@ private:
     void ResetItemCrops();
 
     void Start(bool recordingOnly=false, bool replayBufferOnly=false);
-    void Stop(bool overrideKeepRecording=false);
+    void Stop(bool overrideKeepRecording=false, bool stopReplayBuffer=false);
     bool StartRecording(bool force=false);
     void StopRecording();
     void StartReplayBuffer();
-    void StopReplayBuffer();
+    void StopReplayBuffer(bool immediate=false);
 
     static void STDCALL StartStreamHotkey(DWORD hotkey, UPARAM param, bool bDown);
     static void STDCALL StopStreamHotkey(DWORD hotkey, UPARAM param, bool bDown);

Source/OBSCapture.cpp

@@ -109,20 +109,32 @@ void OBS::StartReplayBuffer()
     ConfigureStreamButtons();
 }
 
-void OBS::StopReplayBuffer()
+void OBS::StopReplayBuffer(bool immediate)
 {
     if (!replayBufferStream) return;
 
+    if (!immediate && replayBufferStop.func) return;
+
+    auto shutdown = [this]()
+    {
         bRecordingReplayBuffer = false;
         ReportStopRecordingReplayBufferTrigger();
 
-    if (!bStreaming && !bRecording && bRunning) Stop(true);
+        if (!bStreaming && !bRecording && bRunning) PostStopMessage(true);
 
         auto stream = move(replayBufferStream);
 
         replayBuffer = nullptr;
 
         ConfigureStreamButtons();
+    };
+
+    if (immediate)
+        return shutdown();
+
+    replayBufferStop.func = shutdown;
+
+    replayBufferStop.time = (DWORD)(GetVideoTime() - firstFrameTimestamp);
 }
 
 String ExpandRecordingFilename(String filename)
@@ -876,7 +888,7 @@ retryHookTestV2:
     ConfigureStreamButtons();
 }
 
-void OBS::Stop(bool overrideKeepRecording)
+void OBS::Stop(bool overrideKeepRecording, bool stopReplayBuffer)
 {
     if((!bStreaming && !bRecording && !bRunning && !bRecordingReplayBuffer) && (!bTestStream)) return;
 
@@ -886,7 +898,7 @@ void OBS::Stop(bool overrideKeepRecording)
 
     int networkMode = AppConfig->GetInt(TEXT("Publish"), TEXT("Mode"), 2);
 
-    if(((!overrideKeepRecording && bRecording && bKeepRecording) || bRecordingReplayBuffer) && networkMode == 0) {
+    if((!overrideKeepRecording && ((bRecording && bKeepRecording) || (!stopReplayBuffer && bRecordingReplayBuffer))) && networkMode == 0) {
         videoEncoder->RequestKeyframe();
 
         Log(TEXT("=====Stream End (recording continues): %s========================="), CurrentDateTimeString().Array());
@@ -967,6 +979,7 @@ void OBS::Stop(bool overrideKeepRecording)
     bStreaming = false;
     
     if(bRecording) StopRecording();
+    if (bRecordingReplayBuffer) StopReplayBuffer(true);
 
     delete micAudio;
     micAudio = NULL;

Source/OBSVideoCapture.cpp

@@ -217,10 +217,13 @@ void OBS::SendFrame(VideoSegment &curSegment, QWORD firstFrameTime)
             if (fileStream)
                 fileStream->AddPacket(shared_data, curSegment.timestamp, curSegment.pts, packet.type);
             if (replayBufferStream)
+            {
+                if (!HandleStreamStopInfo(replayBufferStop, packet.type, curSegment))
                     replayBufferStream->AddPacket(shared_data, curSegment.timestamp, curSegment.pts, packet.type);
             }
         }
     }
+}
 
 bool OBS::HandleStreamStopInfo(OBS::StopInfo &info, PacketType type, const VideoSegment& segment)
 {

Source/ReplayBuffer.cpp

@@ -37,7 +37,7 @@ namespace
     using packet_vec_t = deque<shared_ptr<const packet_t>>;
 }
 
-void CreateRecordingHelper(VideoFileStream *&stream, packet_list_t &packets);
+void CreateRecordingHelper(unique_ptr<VideoFileStream> &stream, packet_list_t &packets);
 
 static DWORD STDCALL SaveReplayBufferThread(void *arg);
 
@@ -66,6 +66,8 @@ struct ReplayBuffer : VideoFileStream
         for (auto &thread : threads)
             if (WaitForSingleObject(thread.second.get(), seconds * 100) != WAIT_OBJECT_0)
                 OSTerminateThread(thread.first.release(), 0);
+            else
+                App->AddPendingStreamThread(thread.first.release());
     }
     
     virtual void AddPacket(const BYTE *data, UINT size, DWORD timestamp, DWORD pts, PacketType type) override
@@ -369,7 +371,7 @@ struct RecordingHelper : VideoFileStream
     }
 };
 
-void CreateRecordingHelper(VideoFileStream *&stream, packet_list_t &packets)
+void CreateRecordingHelper(unique_ptr<VideoFileStream> &stream, packet_list_t &packets)
 {
     if (stream)
     {
@@ -382,7 +384,7 @@ void CreateRecordingHelper(VideoFileStream *&stream, packet_list_t &packets)
 
     auto helper = make_unique<RecordingHelper>(packet_vec_t{begin(packets), end(packets)});
     if (helper->StartRecording())
-        stream = helper.release();
+        stream.reset(helper.release());
 }
 
 pair<ReplayBuffer*, unique_ptr<VideoFileStream>> CreateReplayBuffer(int seconds)