From e1fac6248e95ab7a25dace78bc311913747f2768 Mon Sep 17 00:00:00 2001
From: Thulinma <jaron@vietors.com>
Date: Thu, 21 Jul 2016 11:39:31 +0200
Subject: [PATCH] Fix RTMP not shutting down on playback.

---
 src/output/output.cpp      |  3 +++
 src/output/output_rtmp.cpp | 43 +++++++++++++++++++++++++++++++++-----
 src/output/output_rtmp.h   |  1 +
 3 files changed, 42 insertions(+), 5 deletions(-)

diff --git a/src/output/output.cpp b/src/output/output.cpp
index 4326cf3e..2670ba2e 100644
--- a/src/output/output.cpp
+++ b/src/output/output.cpp
@@ -657,6 +657,7 @@ namespace Mist {
     }
     stats();
     }
+    onFinish();
     MEDIUM_MSG("MistOut client handler shutting down: %s, %s, %s", myConn.connected() ? "conn_active" : "conn_closed", wantRequest ? "want_request" : "no_want_request", parseData ? "parsing_data" : "not_parsing_data");
     
     stats(true);
@@ -941,11 +942,13 @@ namespace Mist {
       nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
       if (!nProxy.userClient.getData()){
         WARN_MSG("Player connection failure - aborting output");
+        onFinish();
         myConn.close();
         return;
       }
     }
     if (!nProxy.userClient.isAlive()){
+      onFinish();
       myConn.close();
       return;
     }
diff --git a/src/output/output_rtmp.cpp b/src/output/output_rtmp.cpp
index 14a54621..6d10e10b 100644
--- a/src/output/output_rtmp.cpp
+++ b/src/output/output_rtmp.cpp
@@ -51,6 +51,38 @@ namespace Mist {
     }
   }
 
+  bool OutRTMP::onFinish(){
+    MEDIUM_MSG("Finishing stream %s, %s", streamName.c_str(), myConn?"while connected":"already disconnected");
+    if (myConn){
+      myConn.SendNow(RTMPStream::SendUSR(1, 1)); //send UCM StreamEOF (1), stream 1
+      AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
+      amfreply.addContent(AMF::Object("", "onStatus")); //status reply
+      amfreply.addContent(AMF::Object("", (double)0)); //transaction ID
+      amfreply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL)); //null - command info
+      amfreply.addContent(AMF::Object("")); //info
+      amfreply.getContentP(3)->addContent(AMF::Object("level", "status"));
+      amfreply.getContentP(3)->addContent(AMF::Object("code", "NetStream.Play.Stop"));
+      amfreply.getContentP(3)->addContent(AMF::Object("description", "Stream stopped"));
+      amfreply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
+      amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
+      sendCommand(amfreply, 20, 1);
+
+      amfreply = AMF::Object ("container", AMF::AMF0_DDV_CONTAINER);
+      amfreply.addContent(AMF::Object("", "onStatus")); //status reply
+      amfreply.addContent(AMF::Object("", (double)0)); //transaction ID
+      amfreply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL)); //null - command info
+      amfreply.addContent(AMF::Object("")); //info
+      amfreply.getContentP(3)->addContent(AMF::Object("level", "status"));
+      amfreply.getContentP(3)->addContent(AMF::Object("code", "NetStream.Play.UnpublishNotify"));
+      amfreply.getContentP(3)->addContent(AMF::Object("description", "Stream stopped"));
+      amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
+      sendCommand(amfreply, 20, 1);
+
+      myConn.close();
+    }
+    return false;
+  }
+
   void OutRTMP::parseVars(std::string data){
     std::string varname;
     std::string varval;
@@ -445,6 +477,7 @@ namespace Mist {
     } //createStream
     if ((amfData.getContentP(0)->StrValue() == "closeStream") || (amfData.getContentP(0)->StrValue() == "deleteStream")) {
       stop();
+      onFinish();
       return;
     }
     if ((amfData.getContentP(0)->StrValue() == "FCUnpublish") || (amfData.getContentP(0)->StrValue() == "releaseStream")) {
@@ -524,20 +557,20 @@ namespace Mist {
         if (streamCfg){
           if (streamCfg.getMember("source").asString().substr(0, 7) != "push://"){
             FAIL_MSG("Push rejected - stream %s not a push-able stream. (%s != push://*)", streamName.c_str(), streamCfg.getMember("source").asString().c_str());
-            myConn.close();
+            onFinish();
           }else{
             std::string source = streamCfg.getMember("source").asString().substr(7);
             std::string IP = source.substr(0, source.find('@'));
             if (IP != ""){
               if (!myConn.isAddress(IP)){
                 FAIL_MSG("Push from %s to %s rejected - source host not whitelisted", getConnectedHost().c_str(), streamName.c_str());
-                myConn.close();
+                onFinish();
               }
             }
           }
         }else{
           FAIL_MSG("Push from %s rejected - stream '%s' not configured.", getConnectedHost().c_str(), streamName.c_str());
-          myConn.close();
+          onFinish();
         }
         configLock.post();
         configLock.close();
@@ -777,7 +810,7 @@ namespace Mist {
             inputBuffer.get().clear();
           }
           stop();
-          myConn.close();
+          onFinish();
           break; //happens when connection breaks unexpectedly
         case 1: //set chunk size
           RTMPStream::chunk_rec_max = ntohl(*(int *)next.data.c_str());
@@ -850,7 +883,7 @@ namespace Mist {
           static std::map<unsigned int, AMF::Object> pushMeta;
           if (!isInitialized) {
             MEDIUM_MSG("Received useless media data");
-            myConn.close();
+            onFinish();
             break;
           }
           F.ChunkLoader(next);
diff --git a/src/output/output_rtmp.h b/src/output/output_rtmp.h
index a09c6de3..7441de2d 100644
--- a/src/output/output_rtmp.h
+++ b/src/output/output_rtmp.h
@@ -14,6 +14,7 @@ namespace Mist {
       void sendNext();
       void sendHeader();
       bool isReadyForPlay();
+      bool onFinish();
     protected:
       bool isPushing;
       void parseVars(std::string data);