refine rtmp protocol stack, refer to go.rtmp

trunk/src/app/srs_app_dvr.cpp

@@ -455,7 +455,7 @@ int SrsDvrPlan::on_meta_data(SrsOnMetaDataPacket* metadata)
     return ret;
 }
 
-int SrsDvrPlan::on_audio(SrsSharedPtrMessage* audio)
+int SrsDvrPlan::on_audio(__SrsSharedPtrMessage* audio)
 {
     int ret = ERROR_SUCCESS;
     
@@ -481,7 +481,7 @@ int SrsDvrPlan::on_audio(SrsSharedPtrMessage* audio)
     return ret;
 }
 
-int SrsDvrPlan::on_video(SrsSharedPtrMessage* video)
+int SrsDvrPlan::on_video(__SrsSharedPtrMessage* video)
 {
     int ret = ERROR_SUCCESS;
     
@@ -573,7 +573,7 @@ int SrsDvrPlan::flv_close()
     return ret;
 }
 
-int SrsDvrPlan::update_duration(SrsSharedPtrMessage* msg)
+int SrsDvrPlan::update_duration(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
 
@@ -720,7 +720,7 @@ void SrsDvrSegmentPlan::on_unpublish()
     dvr_enabled = false;
 }
 
-int SrsDvrSegmentPlan::update_duration(SrsSharedPtrMessage* msg)
+int SrsDvrSegmentPlan::update_duration(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -903,7 +903,7 @@ int SrsDvrHssPlan::on_dvr_reap_flv_header(string path)
     return ret;
 }
 
-int SrsDvrHssPlan::update_duration(SrsSharedPtrMessage* msg)
+int SrsDvrHssPlan::update_duration(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -998,11 +998,11 @@ int SrsDvr::on_meta_data(SrsOnMetaDataPacket* metadata)
     return ret;
 }
 
-int SrsDvr::on_audio(SrsSharedPtrMessage* audio)
+int SrsDvr::on_audio(__SrsSharedPtrMessage* audio)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsAutoFree(SrsSharedPtrMessage, audio, false);
+    SrsAutoFree(__SrsSharedPtrMessage, audio, false);
     
     if ((ret = plan->on_audio(audio)) != ERROR_SUCCESS) {
         return ret;
@@ -1011,11 +1011,11 @@ int SrsDvr::on_audio(SrsSharedPtrMessage* audio)
     return ret;
 }
 
-int SrsDvr::on_video(SrsSharedPtrMessage* video)
+int SrsDvr::on_video(__SrsSharedPtrMessage* video)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsAutoFree(SrsSharedPtrMessage, video, false);
+    SrsAutoFree(__SrsSharedPtrMessage, video, false);
     
     if ((ret = plan->on_video(video)) != ERROR_SUCCESS) {
         return ret;

trunk/src/app/srs_app_dvr.hpp

@@ -36,7 +36,7 @@ class SrsRequest;
 class SrsStream;
 class SrsRtmpJitter;
 class SrsOnMetaDataPacket;
-class SrsSharedPtrMessage;
+class __SrsSharedPtrMessage;
 
 /**
 * file stream to read/write file.
@@ -185,13 +185,13 @@ public:
     virtual int on_publish();
     virtual void on_unpublish() = 0;
     virtual int on_meta_data(SrsOnMetaDataPacket* metadata);
-    virtual int on_audio(SrsSharedPtrMessage* audio);
+    virtual int on_audio(__SrsSharedPtrMessage* audio);
-    virtual int on_video(SrsSharedPtrMessage* video);
+    virtual int on_video(__SrsSharedPtrMessage* video);
 protected:
     virtual int flv_open(std::string stream, std::string path);
     virtual int flv_close();
     virtual int open_new_segment();
-    virtual int update_duration(SrsSharedPtrMessage* msg);
+    virtual int update_duration(__SrsSharedPtrMessage* msg);
     virtual int write_flv_header();
     virtual int on_dvr_request_sh();
     virtual int on_video_keyframe();
@@ -233,7 +233,7 @@ public:
     virtual int on_publish();
     virtual void on_unpublish();
 private:
-    virtual int update_duration(SrsSharedPtrMessage* msg);
+    virtual int update_duration(__SrsSharedPtrMessage* msg);
 };
 
 /**
@@ -260,7 +260,7 @@ protected:
     virtual int64_t filter_timestamp(int64_t timestamp);
 private:
     virtual int on_dvr_reap_flv_header(std::string path);
-    virtual int update_duration(SrsSharedPtrMessage* msg);
+    virtual int update_duration(__SrsSharedPtrMessage* msg);
 };
 
 /**
@@ -300,11 +300,11 @@ public:
     /**
     * mux the audio packets to dvr.
     */
-    virtual int on_audio(SrsSharedPtrMessage* audio);
+    virtual int on_audio(__SrsSharedPtrMessage* audio);
     /**
     * mux the video packets to dvr.
     */
-    virtual int on_video(SrsSharedPtrMessage* video);
+    virtual int on_video(__SrsSharedPtrMessage* video);
 };
 
 #endif

trunk/src/app/srs_app_edge.cpp

@@ -175,15 +175,15 @@ int SrsEdgeIngester::ingest()
         }
 
         // read from client.
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
-        if ((ret = client->recv_message(&msg)) != ERROR_SUCCESS) {
+        if ((ret = client->__recv_message(&msg)) != ERROR_SUCCESS) {
             srs_error("ingest recv origin server message failed. ret=%d", ret);
             return ret;
         }
         srs_verbose("edge loop recv message. ret=%d", ret);
         
         srs_assert(msg);
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
         
         if ((ret = process_publish_message(msg)) != ERROR_SUCCESS) {
             return ret;
@@ -193,7 +193,7 @@ int SrsEdgeIngester::ingest()
     return ret;
 }
 
-int SrsEdgeIngester::process_publish_message(SrsCommonMessage* msg)
+int SrsEdgeIngester::process_publish_message(__SrsMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -217,12 +217,13 @@ int SrsEdgeIngester::process_publish_message(SrsCommonMessage* msg)
 
     // process onMetaData
     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
-        if ((ret = msg->decode_packet(client->get_protocol())) != ERROR_SUCCESS) {
+        SrsPacket* pkt = NULL;
+        if ((ret = client->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("decode onMetaData message failed. ret=%d", ret);
             return ret;
         }
+        SrsAutoFree(SrsPacket, pkt, false);
     
-        SrsPacket* pkt = msg->get_packet();
         if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
             SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
             if ((ret = source->on_meta_data(msg, metadata)) != ERROR_SUCCESS) {
@@ -419,8 +420,8 @@ int SrsEdgeForwarder::cycle()
 
         // read from client.
         if (true) {
-            SrsCommonMessage* msg = NULL;
+            __SrsMessage* msg = NULL;
-            ret = client->recv_message(&msg);
+            ret = client->__recv_message(&msg);
             
             srs_verbose("edge loop recv message. ret=%d", ret);
             if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
@@ -434,7 +435,7 @@ int SrsEdgeForwarder::cycle()
         
         // forward all messages.
         int count = 0;
-        SrsSharedPtrMessage** msgs = NULL;
+        __SrsSharedPtrMessage** msgs = NULL;
         if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
             srs_error("get message to forward to origin failed. ret=%d", ret);
             return ret;
@@ -455,16 +456,16 @@ int SrsEdgeForwarder::cycle()
             srs_verbose("no packets to forward.");
             continue;
         }
-        SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
+        SrsAutoFree(__SrsSharedPtrMessage*, msgs, true);
     
         // all msgs to forward.
         for (int i = 0; i < count; i++) {
-            SrsSharedPtrMessage* msg = msgs[i];
+            __SrsSharedPtrMessage* msg = msgs[i];
             
             srs_assert(msg);
             msgs[i] = NULL;
             
-            if ((ret = client->send_message(msg)) != ERROR_SUCCESS) {
+            if ((ret = client->__send_and_free_message(msg)) != ERROR_SUCCESS) {
                 srs_error("edge publish forwarder send message to server failed. ret=%d", ret);
                 return ret;
             }
@@ -474,7 +475,7 @@ int SrsEdgeForwarder::cycle()
     return ret;
 }
 
-int SrsEdgeForwarder::proxy(SrsCommonMessage* msg)
+int SrsEdgeForwarder::proxy(__SrsMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -494,8 +495,8 @@ int SrsEdgeForwarder::proxy(SrsCommonMessage* msg)
     }
     
     // TODO: FIXME: use utility to copy msg to shared ptr msg.
-    SrsSharedPtrMessage* copy = new SrsSharedPtrMessage();
+    __SrsSharedPtrMessage* copy = new __SrsSharedPtrMessage();
-    SrsAutoFree(SrsSharedPtrMessage, copy, false);
+    SrsAutoFree(__SrsSharedPtrMessage, copy, false);
     if ((ret = copy->initialize(msg)) != ERROR_SUCCESS) {
         srs_error("initialize the msg failed. ret=%d", ret);
         return ret;
@@ -723,7 +724,7 @@ int SrsPublishEdge::on_client_publish()
     return ret;
 }
 
-int SrsPublishEdge::on_proxy_publish(SrsCommonMessage* msg)
+int SrsPublishEdge::on_proxy_publish(__SrsMessage* msg)
 {
     return forwarder->proxy(msg);
 }

trunk/src/app/srs_app_edge.hpp

@@ -40,7 +40,7 @@ class SrsRequest;
 class SrsPlayEdge;
 class SrsPublishEdge;
 class SrsRtmpClient;
-class SrsCommonMessage;
+class __SrsMessage;
 class SrsMessageQueue;
 class ISrsProtocolReaderWriter;
 
@@ -99,7 +99,7 @@ private:
     virtual int ingest();
     virtual void close_underlayer_socket();
     virtual int connect_server();
-    virtual int process_publish_message(SrsCommonMessage* msg);
+    virtual int process_publish_message(__SrsMessage* msg);
 };
 
 /**
@@ -142,7 +142,7 @@ public:
 public:
     virtual int cycle();
 public:
-    virtual int proxy(SrsCommonMessage* msg);
+    virtual int proxy(__SrsMessage* msg);
 private:
     virtual void close_underlayer_socket();
     virtual int connect_server();
@@ -202,7 +202,7 @@ public:
     /**
     * proxy publish stream to edge
     */
-    virtual int on_proxy_publish(SrsCommonMessage* msg);
+    virtual int on_proxy_publish(__SrsMessage* msg);
     /**
     * proxy unpublish stream to edge.
     */

trunk/src/app/srs_app_forward.cpp

@@ -148,7 +148,7 @@ void SrsForwarder::on_unpublish()
     srs_freep(io);
 }
 
-int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
+int SrsForwarder::on_meta_data(__SrsSharedPtrMessage* metadata)
 {
     int ret = ERROR_SUCCESS;
     
@@ -164,7 +164,7 @@ int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
     return ret;
 }
 
-int SrsForwarder::on_audio(SrsSharedPtrMessage* msg)
+int SrsForwarder::on_audio(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -180,7 +180,7 @@ int SrsForwarder::on_audio(SrsSharedPtrMessage* msg)
     return ret;
 }
 
-int SrsForwarder::on_video(SrsSharedPtrMessage* msg)
+int SrsForwarder::on_video(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -329,7 +329,7 @@ int SrsForwarder::forward()
         
         // forward all messages.
         int count = 0;
-        SrsSharedPtrMessage** msgs = NULL;
+        __SrsSharedPtrMessage** msgs = NULL;
         if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
             srs_error("get message to forward failed. ret=%d", ret);
             return ret;
@@ -348,16 +348,16 @@ int SrsForwarder::forward()
             srs_verbose("no packets to forward.");
             continue;
         }
-        SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
+        SrsAutoFree(__SrsSharedPtrMessage*, msgs, true);
     
         // all msgs to forward.
         for (int i = 0; i < count; i++) {
-            SrsSharedPtrMessage* msg = msgs[i];
+            __SrsSharedPtrMessage* msg = msgs[i];
             
             srs_assert(msg);
             msgs[i] = NULL;
             
-            if ((ret = client->send_message(msg)) != ERROR_SUCCESS) {
+            if ((ret = client->__send_and_free_message(msg)) != ERROR_SUCCESS) {
                 srs_error("forwarder send message to server failed. ret=%d", ret);
                 return ret;
             }

trunk/src/app/srs_app_forward.hpp

@@ -35,7 +35,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_app_thread.hpp>
 
 class ISrsProtocolReaderWriter;
-class SrsSharedPtrMessage;
+class __SrsSharedPtrMessage;
 class SrsOnMetaDataPacket;
 class SrsMessageQueue;
 class SrsRtmpJitter;
@@ -72,9 +72,9 @@ public:
 public:
     virtual int on_publish(SrsRequest* req, std::string forward_server);
     virtual void on_unpublish();
-    virtual int on_meta_data(SrsSharedPtrMessage* metadata);
+    virtual int on_meta_data(__SrsSharedPtrMessage* metadata);
-    virtual int on_audio(SrsSharedPtrMessage* msg);
+    virtual int on_audio(__SrsSharedPtrMessage* msg);
-    virtual int on_video(SrsSharedPtrMessage* msg);
+    virtual int on_video(__SrsSharedPtrMessage* msg);
 // interface ISrsThreadHandler.
 public:
     virtual int cycle();

trunk/src/app/srs_app_hls.cpp

@@ -1379,11 +1379,11 @@ int SrsHls::on_meta_data(SrsAmf0Object* metadata)
     return ret;
 }
 
-int SrsHls::on_audio(SrsSharedPtrMessage* audio)
+int SrsHls::on_audio(__SrsSharedPtrMessage* audio)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsAutoFree(SrsSharedPtrMessage, audio, false);
+    SrsAutoFree(__SrsSharedPtrMessage, audio, false);
     
     if (!hls_enabled) {
         return ret;
@@ -1423,11 +1423,11 @@ int SrsHls::on_audio(SrsSharedPtrMessage* audio)
     return ret;
 }
 
-int SrsHls::on_video(SrsSharedPtrMessage* video)
+int SrsHls::on_video(__SrsSharedPtrMessage* video)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsAutoFree(SrsSharedPtrMessage, video, false);
+    SrsAutoFree(__SrsSharedPtrMessage, video, false);
     
     if (!hls_enabled) {
         return ret;

trunk/src/app/srs_app_hls.hpp

@@ -34,7 +34,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <string>
 #include <vector>
 
-class SrsSharedPtrMessage;
+class __SrsSharedPtrMessage;
 class SrsCodecSample;
 class SrsCodecBuffer;
 class SrsMpegtsFrame;
@@ -314,11 +314,11 @@ public:
     /**
     * mux the audio packets to ts.
     */
-    virtual int on_audio(SrsSharedPtrMessage* audio);
+    virtual int on_audio(__SrsSharedPtrMessage* audio);
     /**
     * mux the video packets to ts.
     */
-    virtual int on_video(SrsSharedPtrMessage* video);
+    virtual int on_video(__SrsSharedPtrMessage* video);
 private:
     virtual void hls_mux();
 };

trunk/src/app/srs_app_rtmp_conn.cpp

@@ -472,8 +472,8 @@ int SrsRtmpConn::playing(SrsSource* source)
         // read from client.
         int ctl_msg_ret = ERROR_SUCCESS;
         if (true) {
-            SrsCommonMessage* msg = NULL;
+            __SrsMessage* msg = NULL;
-            ctl_msg_ret = ret = rtmp->recv_message(&msg);
+            ctl_msg_ret = ret = rtmp->__recv_message(&msg);
             
             srs_verbose("play loop recv message. ret=%d", ret);
             if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
@@ -491,7 +491,7 @@ int SrsRtmpConn::playing(SrsSource* source)
         }
         
         // get messages from consumer.
-        SrsSharedPtrMessage** msgs = NULL;
+        __SrsSharedPtrMessage** msgs = NULL;
         int count = 0;
         if ((ret = consumer->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
             srs_error("get messages from consumer failed. ret=%d", ret);
@@ -510,11 +510,11 @@ int SrsRtmpConn::playing(SrsSource* source)
             srs_verbose("no packets in queue.");
             continue;
         }
-        SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
+        SrsAutoFree(__SrsSharedPtrMessage*, msgs, true);
         
         // sendout messages
         for (int i = 0; i < count; i++) {
-            SrsSharedPtrMessage* msg = msgs[i];
+            __SrsSharedPtrMessage* msg = msgs[i];
             
             // the send_message will free the msg, 
             // so set the msgs[i] to NULL.
@@ -530,7 +530,7 @@ int SrsRtmpConn::playing(SrsSource* source)
             duration += msg->header.timestamp - starttime;
             starttime = msg->header.timestamp;
             
-            if ((ret = rtmp->send_message(msg)) != ERROR_SUCCESS) {
+            if ((ret = rtmp->__send_and_free_message(msg)) != ERROR_SUCCESS) {
                 srs_error("send message to client failed. ret=%d", ret);
                 return ret;
             }
@@ -573,14 +573,13 @@ int SrsRtmpConn::fmle_publish(SrsSource* source)
         // switch to other st-threads.
         st_usleep(0);
         
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
-        if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
+        if ((ret = rtmp->__recv_message(&msg)) != ERROR_SUCCESS) {
             srs_error("fmle recv identify client message failed. ret=%d", ret);
             return ret;
         }
 
-        srs_assert(msg);
+        SrsAutoFree(__SrsMessage, msg, false);
-        SrsAutoFree(SrsCommonMessage, msg, false);
         
         pithy_print.elapse();
 
@@ -594,12 +593,14 @@ int SrsRtmpConn::fmle_publish(SrsSource* source)
     
         // process UnPublish event.
         if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
-            if ((ret = msg->decode_packet(rtmp->get_protocol())) != ERROR_SUCCESS) {
+            SrsPacket* pkt = NULL;
+            if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
                 srs_error("fmle decode unpublish message failed. ret=%d", ret);
                 return ret;
             }
+            
+            SrsAutoFree(SrsPacket, pkt, false);
         
-            SrsPacket* pkt = msg->get_packet();
             if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
                 SrsFMLEStartPacket* unpublish = dynamic_cast<SrsFMLEStartPacket*>(pkt);
                 if ((ret = rtmp->fmle_unpublish(res->stream_id, unpublish->transaction_id)) != ERROR_SUCCESS) {
@@ -647,15 +648,15 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
         // switch to other st-threads.
         st_usleep(0);
         
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
-        if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
+        if ((ret = rtmp->__recv_message(&msg)) != ERROR_SUCCESS) {
             if (!srs_is_client_gracefully_close(ret)) {
                 srs_error("flash recv identify client message failed. ret=%d", ret);
             }
             return ret;
         }
 
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
         
         pithy_print.elapse();
 
@@ -669,11 +670,14 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
     
         // process UnPublish event.
         if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
-            if ((ret = msg->decode_packet(rtmp->get_protocol())) != ERROR_SUCCESS) {
+            SrsPacket* pkt = NULL;
+            if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
                 srs_error("flash decode unpublish message failed. ret=%d", ret);
                 return ret;
             }
             
+            SrsAutoFree(SrsPacket, pkt, false);
+            
             // flash unpublish.
             // TODO: maybe need to support republish.
             srs_trace("flash flash publish finished.");
@@ -690,7 +694,7 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
     return ret;
 }
 
-int SrsRtmpConn::process_publish_message(SrsSource* source, SrsCommonMessage* msg, bool vhost_is_edge)
+int SrsRtmpConn::process_publish_message(SrsSource* source, __SrsMessage* msg, bool vhost_is_edge)
 {
     int ret = ERROR_SUCCESS;
     
@@ -720,12 +724,13 @@ int SrsRtmpConn::process_publish_message(SrsSource* source, SrsCommonMessage* ms
     
     // process onMetaData
     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
-        if ((ret = msg->decode_packet(rtmp->get_protocol())) != ERROR_SUCCESS) {
+        SrsPacket* pkt = NULL;
+        if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("decode onMetaData message failed. ret=%d", ret);
             return ret;
         }
+        SrsAutoFree(SrsPacket, pkt, false);
     
-        SrsPacket* pkt = msg->get_packet();
         if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
             SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
             if ((ret = source->on_meta_data(msg, metadata)) != ERROR_SUCCESS) {
@@ -743,7 +748,7 @@ int SrsRtmpConn::process_publish_message(SrsSource* source, SrsCommonMessage* ms
     return ret;
 }
 
-int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, SrsCommonMessage* msg)
+int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, __SrsMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -751,29 +756,32 @@ int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, SrsCommonMessag
         srs_verbose("ignore all empty message.");
         return ret;
     }
-    SrsAutoFree(SrsCommonMessage, msg, false);
+    SrsAutoFree(__SrsMessage, msg, false);
     
     if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
         srs_info("ignore all message except amf0/amf3 command.");
         return ret;
     }
     
-    if ((ret = msg->decode_packet(rtmp->get_protocol())) != ERROR_SUCCESS) {
+    SrsPacket* pkt = NULL;
+    if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
         srs_error("decode the amf0/amf3 command packet failed. ret=%d", ret);
         return ret;
     }
     srs_info("decode the amf0/amf3 command packet success.");
     
+    SrsAutoFree(SrsPacket, pkt, false);
+    
     // for jwplayer/flowplayer, which send close as pause message.
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/6
-    SrsCloseStreamPacket* close = dynamic_cast<SrsCloseStreamPacket*>(msg->get_packet());
+    SrsCloseStreamPacket* close = dynamic_cast<SrsCloseStreamPacket*>(pkt);
     if (close) {
         ret = ERROR_CONTROL_RTMP_CLOSE;
         srs_trace("system control message: rtmp close stream. ret=%d", ret);
         return ret;
     }
     
-    SrsPausePacket* pause = dynamic_cast<SrsPausePacket*>(msg->get_packet());
+    SrsPausePacket* pause = dynamic_cast<SrsPausePacket*>(pkt);
     if (!pause) {
         srs_info("ignore all amf0/amf3 command except pause.");
         return ret;

trunk/src/app/srs_app_rtmp_conn.hpp

@@ -40,7 +40,7 @@ class SrsResponse;
 class SrsSource;
 class SrsRefer;
 class SrsConsumer;
-class SrsCommonMessage;
+class __SrsMessage;
 class SrsSocket;
 #ifdef SRS_AUTO_HTTP_CALLBACK    
 class SrsHttpHooks;
@@ -80,8 +80,8 @@ private:
     virtual int playing(SrsSource* source);
     virtual int fmle_publish(SrsSource* source);
     virtual int flash_publish(SrsSource* source);
-    virtual int process_publish_message(SrsSource* source, SrsCommonMessage* msg, bool vhost_is_edge);
+    virtual int process_publish_message(SrsSource* source, __SrsMessage* msg, bool vhost_is_edge);
-    virtual int process_play_control_msg(SrsConsumer* consumer, SrsCommonMessage* msg);
+    virtual int process_play_control_msg(SrsConsumer* consumer, __SrsMessage* msg);
 private:
     virtual int http_hooks_on_connect();
     virtual void http_hooks_on_close();

trunk/src/app/srs_app_source.cpp

@@ -52,7 +52,7 @@ SrsRtmpJitter::~SrsRtmpJitter()
 {
 }
 
-int SrsRtmpJitter::correct(SrsSharedPtrMessage* msg, int tba, int tbv)
+int SrsRtmpJitter::correct(__SrsSharedPtrMessage* msg, int tba, int tbv)
 {
     int ret = ERROR_SUCCESS;
 
@@ -130,7 +130,7 @@ void SrsMessageQueue::set_queue_size(double queue_size)
     queue_size_ms = (int)(queue_size * 1000);
 }
 
-int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
+int SrsMessageQueue::enqueue(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -151,7 +151,7 @@ int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
     return ret;
 }
 
-int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
+int SrsMessageQueue::get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count)
 {
     int ret = ERROR_SUCCESS;
     
@@ -169,13 +169,13 @@ int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, in
         return ret;
     }
     
-    pmsgs = new SrsSharedPtrMessage*[count];
+    pmsgs = new __SrsSharedPtrMessage*[count];
     
     for (int i = 0; i < count; i++) {
         pmsgs[i] = msgs[i];
     }
     
-    SrsSharedPtrMessage* last = msgs[count - 1];
+    __SrsSharedPtrMessage* last = msgs[count - 1];
     av_start_time = last->header.timestamp;
     
     if (count == (int)msgs.size()) {
@@ -196,7 +196,7 @@ void SrsMessageQueue::shrink()
     // for when we shrinked, the first is the iframe,
     // we will directly remove the gop next time.
     for (int i = 1; i < (int)msgs.size(); i++) {
-        SrsSharedPtrMessage* msg = msgs[i];
+        __SrsSharedPtrMessage* msg = msgs[i];
         
         if (msg->header.is_video()) {
             if (SrsCodec::video_is_keyframe(msg->payload, msg->size)) {
@@ -222,7 +222,7 @@ void SrsMessageQueue::shrink()
     
     // remove the first gop from the front
     for (int i = 0; i < iframe_index; i++) {
-        SrsSharedPtrMessage* msg = msgs[i];
+        __SrsSharedPtrMessage* msg = msgs[i];
         srs_freep(msg);
     }
     msgs.erase(msgs.begin(), msgs.begin() + iframe_index);
@@ -230,10 +230,10 @@ void SrsMessageQueue::shrink()
 
 void SrsMessageQueue::clear()
 {
-    std::vector<SrsSharedPtrMessage*>::iterator it;
+    std::vector<__SrsSharedPtrMessage*>::iterator it;
 
     for (it = msgs.begin(); it != msgs.end(); ++it) {
-        SrsSharedPtrMessage* msg = *it;
+        __SrsSharedPtrMessage* msg = *it;
         srs_freep(msg);
     }
 
@@ -267,7 +267,7 @@ int SrsConsumer::get_time()
     return jitter->get_time();
 }
 
-int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, int tba, int tbv)
+int SrsConsumer::enqueue(__SrsSharedPtrMessage* msg, int tba, int tbv)
 {
     int ret = ERROR_SUCCESS;
     
@@ -285,7 +285,7 @@ int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, int tba, int tbv)
     return ret;
 }
 
-int SrsConsumer::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
+int SrsConsumer::get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count)
 {
     // paused, return nothing.
     if (paused) {
@@ -329,7 +329,7 @@ void SrsGopCache::set(bool enabled)
     srs_info("enable gop cache");
 }
 
-int SrsGopCache::cache(SrsSharedPtrMessage* msg)
+int SrsGopCache::cache(__SrsSharedPtrMessage* msg)
 {
     int ret = ERROR_SUCCESS;
     
@@ -368,9 +368,9 @@ int SrsGopCache::cache(SrsSharedPtrMessage* msg)
 
 void SrsGopCache::clear()
 {
-    std::vector<SrsSharedPtrMessage*>::iterator it;
+    std::vector<__SrsSharedPtrMessage*>::iterator it;
     for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
-        SrsSharedPtrMessage* msg = *it;
+        __SrsSharedPtrMessage* msg = *it;
         srs_freep(msg);
     }
     gop_cache.clear();
@@ -382,9 +382,9 @@ int SrsGopCache::dump(SrsConsumer* consumer, int tba, int tbv)
 {
     int ret = ERROR_SUCCESS;
     
-    std::vector<SrsSharedPtrMessage*>::iterator it;
+    std::vector<__SrsSharedPtrMessage*>::iterator it;
     for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
-        SrsSharedPtrMessage* msg = *it;
+        __SrsSharedPtrMessage* msg = *it;
         if ((ret = consumer->enqueue(msg->copy(), tba, tbv)) != ERROR_SUCCESS) {
             srs_error("dispatch cached gop failed. ret=%d", ret);
             return ret;
@@ -406,7 +406,7 @@ int64_t SrsGopCache::get_start_time()
         return 0;
     }
     
-    SrsSharedPtrMessage* msg = gop_cache[0];
+    __SrsSharedPtrMessage* msg = gop_cache[0];
     srs_assert(msg);
     
     return msg->header.timestamp;
@@ -789,7 +789,7 @@ bool SrsSource::can_publish()
     return _can_publish;
 }
 
-int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata)
+int SrsSource::on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata)
 {
     int ret = ERROR_SUCCESS;
     
@@ -840,7 +840,7 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
     
     // create a shared ptr message.
     srs_freep(cache_metadata);
-    cache_metadata = new SrsSharedPtrMessage();
+    cache_metadata = new __SrsSharedPtrMessage();
     
     // dump message to shared ptr message.
     if ((ret = cache_metadata->initialize(&msg->header, payload, size)) != ERROR_SUCCESS) {
@@ -877,12 +877,12 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
     return ret;
 }
 
-int SrsSource::on_audio(SrsCommonMessage* audio)
+int SrsSource::on_audio(__SrsMessage* audio)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsSharedPtrMessage* msg = new SrsSharedPtrMessage();
+    __SrsSharedPtrMessage* msg = new __SrsSharedPtrMessage();
-    SrsAutoFree(SrsSharedPtrMessage, msg, false);
+    SrsAutoFree(__SrsSharedPtrMessage, msg, false);
     if ((ret = msg->initialize(audio)) != ERROR_SUCCESS) {
         srs_error("initialize the audio failed. ret=%d", ret);
         return ret;
@@ -966,12 +966,12 @@ int SrsSource::on_audio(SrsCommonMessage* audio)
     return ret;
 }
 
-int SrsSource::on_video(SrsCommonMessage* video)
+int SrsSource::on_video(__SrsMessage* video)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsSharedPtrMessage* msg = new SrsSharedPtrMessage();
+    __SrsSharedPtrMessage* msg = new __SrsSharedPtrMessage();
-    SrsAutoFree(SrsSharedPtrMessage, msg, false);
+    SrsAutoFree(__SrsSharedPtrMessage, msg, false);
     if ((ret = msg->initialize(video)) != ERROR_SUCCESS) {
         srs_error("initialize the video failed. ret=%d", ret);
         return ret;
@@ -1207,7 +1207,7 @@ int SrsSource::on_edge_start_publish()
     return publish_edge->on_client_publish();
 }
 
-int SrsSource::on_edge_proxy_publish(SrsCommonMessage* msg)
+int SrsSource::on_edge_proxy_publish(__SrsMessage* msg)
 {
     return publish_edge->on_proxy_publish(msg);
 }

trunk/src/app/srs_app_source.hpp

@@ -40,9 +40,9 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 class SrsPlayEdge;
 class SrsPublishEdge;
 class SrsSource;
-class SrsCommonMessage;
+class __SrsMessage;
 class SrsOnMetaDataPacket;
-class SrsSharedPtrMessage;
+class __SrsSharedPtrMessage;
 class SrsForwarder;
 class SrsRequest;
 class SrsSocket;
@@ -74,7 +74,7 @@ public:
     /**
     * detect the time jitter and correct it.
     */
-    virtual int correct(SrsSharedPtrMessage* msg, int tba, int tbv);
+    virtual int correct(__SrsSharedPtrMessage* msg, int tba, int tbv);
     /**
     * get current client time, the last packet time.
     */
@@ -91,7 +91,7 @@ private:
     int64_t av_start_time;
     int64_t av_end_time;
     int queue_size_ms;
-    std::vector<SrsSharedPtrMessage*> msgs;
+    std::vector<__SrsSharedPtrMessage*> msgs;
 public:
     SrsMessageQueue();
     virtual ~SrsMessageQueue();
@@ -106,14 +106,14 @@ public:
     * enqueue the message, the timestamp always monotonically.
     * @param msg, the msg to enqueue, user never free it whatever the return code.
     */
-    virtual int enqueue(SrsSharedPtrMessage* msg);
+    virtual int enqueue(__SrsSharedPtrMessage* msg);
     /**
     * get packets in consumer queue.
     * @pmsgs SrsMessages*[], output the prt array.
     * @count the count in array.
     * @max_count the max count to dequeue, 0 to dequeue all.
     */
-    virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
+    virtual int get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count);
 private:
     /**
     * remove a gop from the front.
@@ -150,14 +150,14 @@ public:
     * @param tbv timebase of video.
     *        used to calc the video time delta if time-jitter detected.
     */
-    virtual int enqueue(SrsSharedPtrMessage* msg, int tba, int tbv);
+    virtual int enqueue(__SrsSharedPtrMessage* msg, int tba, int tbv);
     /**
     * get packets in consumer queue.
     * @pmsgs SrsMessages*[], output the prt array.
     * @count the count in array.
     * @max_count the max count to dequeue, 0 to dequeue all.
     */
-    virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
+    virtual int get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count);
     /**
     * when client send the pause message.
     */
@@ -185,7 +185,7 @@ private:
     /**
     * cached gop.
     */
-    std::vector<SrsSharedPtrMessage*> gop_cache;
+    std::vector<__SrsSharedPtrMessage*> gop_cache;
 public:
     SrsGopCache();
     virtual ~SrsGopCache();
@@ -196,7 +196,7 @@ public:
     * 1. cache the gop when got h264 video packet.
     * 2. clear gop when got keyframe.
     */
-    virtual int cache(SrsSharedPtrMessage* msg);
+    virtual int cache(__SrsSharedPtrMessage* msg);
     virtual void clear();
     virtual int dump(SrsConsumer* consumer, int tba, int tbv);
     /**
@@ -267,11 +267,11 @@ private:
     // TODO: FIXME: to support reload atc.
     bool atc;
 private:
-    SrsSharedPtrMessage* cache_metadata;
+    __SrsSharedPtrMessage* cache_metadata;
     // the cached video sequence header.
-    SrsSharedPtrMessage* cache_sh_video;
+    __SrsSharedPtrMessage* cache_sh_video;
     // the cached audio sequence header.
-    SrsSharedPtrMessage* cache_sh_audio;
+    __SrsSharedPtrMessage* cache_sh_audio;
 public:
     /**
     * @param _req the client request object, 
@@ -299,9 +299,9 @@ public:
     virtual int on_dvr_request_sh();
 public:
     virtual bool can_publish();
-    virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
+    virtual int on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata);
-    virtual int on_audio(SrsCommonMessage* audio);
+    virtual int on_audio(__SrsMessage* audio);
-    virtual int on_video(SrsCommonMessage* video);
+    virtual int on_video(__SrsMessage* video);
     /**
     * publish stream event notify.
     * @param _req the request from client, the source will deep copy it,
@@ -322,7 +322,7 @@ public:
     // for edge, when publish edge stream, check the state
     virtual int on_edge_start_publish();
     // for edge, proxy the publish
-    virtual int on_edge_proxy_publish(SrsCommonMessage* msg);
+    virtual int on_edge_proxy_publish(__SrsMessage* msg);
     // for edge, proxy stop publish
     virtual void on_edge_proxy_unpublish();
 private:

trunk/src/core/srs_core.hpp

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 #define VERSION_MAJOR "0"
 #define VERSION_MINOR "9"
-#define VERSION_REVISION "79"
+#define VERSION_REVISION "80"
 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
 // server info.
 #define RTMP_SIG_SRS_KEY "srs"

trunk/src/rtmp/srs_protocol_rtmp.cpp

@@ -382,6 +382,26 @@ int SrsRtmpClient::send_message(ISrsMessage* msg)
     return protocol->send_message(msg);
 }
 
+int SrsRtmpClient::__recv_message(__SrsMessage** pmsg)
+{
+    return protocol->__recv_message(pmsg);
+}
+
+int SrsRtmpClient::__decode_message(__SrsMessage* msg, SrsPacket** ppacket)
+{
+    return protocol->__decode_message(msg, ppacket);
+}
+
+int SrsRtmpClient::__send_and_free_message(__SrsMessage* msg)
+{
+    return protocol->__send_and_free_message(msg);
+}
+
+int SrsRtmpClient::__send_and_free_packet(SrsPacket* packet, int stream_id)
+{
+    return protocol->__send_and_free_packet(packet, stream_id);
+}
+
 int SrsRtmpClient::handshake()
 {
     int ret = ERROR_SUCCESS;
@@ -768,6 +788,26 @@ int SrsRtmpServer::send_message(ISrsMessage* msg)
     return protocol->send_message(msg);
 }
 
+int SrsRtmpServer::__recv_message(__SrsMessage** pmsg)
+{
+    return protocol->__recv_message(pmsg);
+}
+
+int SrsRtmpServer::__decode_message(__SrsMessage* msg, SrsPacket** ppacket)
+{
+    return protocol->__decode_message(msg, ppacket);
+}
+
+int SrsRtmpServer::__send_and_free_message(__SrsMessage* msg)
+{
+    return protocol->__send_and_free_message(msg);
+}
+
+int SrsRtmpServer::__send_and_free_packet(SrsPacket* packet, int stream_id)
+{
+    return protocol->__send_and_free_packet(packet, stream_id);
+}
+
 int SrsRtmpServer::handshake()
 {
     int ret = ERROR_SUCCESS;
@@ -794,13 +834,13 @@ int SrsRtmpServer::connect_app(SrsRequest* req)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsCommonMessage* msg = NULL;
+    __SrsMessage* msg = NULL;
     SrsConnectAppPacket* pkt = NULL;
-    if ((ret = srs_rtmp_expect_message<SrsConnectAppPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
+    if ((ret = __srs_rtmp_expect_message<SrsConnectAppPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
         srs_error("expect connect app message failed. ret=%d", ret);
         return ret;
     }
-    SrsAutoFree(SrsCommonMessage, msg, false);
+    SrsAutoFree(__SrsMessage, msg, false);
     srs_info("get connect app message");
     
     SrsAmf0Any* prop = NULL;
@@ -833,13 +873,9 @@ int SrsRtmpServer::set_window_ack_size(int ack_size)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsCommonMessage* msg = new SrsCommonMessage();
     SrsSetWindowAckSizePacket* pkt = new SrsSetWindowAckSizePacket();
-    
     pkt->ackowledgement_window_size = ack_size;
-    msg->set_packet(pkt, 0);
+    if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-    
-    if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
         srs_error("send ack size message failed. ret=%d", ret);
         return ret;
     }
@@ -852,14 +888,10 @@ int SrsRtmpServer::set_peer_bandwidth(int bandwidth, int type)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsCommonMessage* msg = new SrsCommonMessage();
     SrsSetPeerBandwidthPacket* pkt = new SrsSetPeerBandwidthPacket();
-    
     pkt->bandwidth = bandwidth;
     pkt->type = type;
-    msg->set_packet(pkt, 0);
+    if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-    
-    if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
         srs_error("send set bandwidth message failed. ret=%d", ret);
         return ret;
     }
@@ -873,7 +905,6 @@ int SrsRtmpServer::response_connect_app(SrsRequest *req, const char* server_ip)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsCommonMessage* msg = new SrsCommonMessage();
     SrsConnectAppResPacket* pkt = new SrsConnectAppResPacket();
     
     pkt->props->set("fmsVer", SrsAmf0Any::str("FMS/"RTMP_SIG_FMS_VER));
@@ -903,9 +934,7 @@ int SrsRtmpServer::response_connect_app(SrsRequest *req, const char* server_ip)
         data->set("srs_server_ip", SrsAmf0Any::str(server_ip));
     }
     
-    msg->set_packet(pkt, 0);
+    if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-    
-    if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
         srs_error("send connect app response message failed. ret=%d", ret);
         return ret;
     }
@@ -939,12 +968,8 @@ int SrsRtmpServer::on_bw_done()
 {
     int ret = ERROR_SUCCESS;
     
-    SrsCommonMessage* msg = new SrsCommonMessage();
     SrsOnBWDonePacket* pkt = new SrsOnBWDonePacket();
-    
+    if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-    msg->set_packet(pkt, 0);
-    
-    if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
         srs_error("send onBWDone message failed. ret=%d", ret);
         return ret;
     }
@@ -959,13 +984,13 @@ int SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string&
     int ret = ERROR_SUCCESS;
     
     while (true) {
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
-        if ((ret = protocol->recv_message(&msg)) != ERROR_SUCCESS) {
+        if ((ret = protocol->__recv_message(&msg)) != ERROR_SUCCESS) {
             srs_error("recv identify client message failed. ret=%d", ret);
             return ret;
         }
 
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
 
         if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
             srs_trace("identify ignore messages except "
@@ -973,12 +998,14 @@ int SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string&
             continue;
         }
         
-        if ((ret = msg->decode_packet(protocol)) != ERROR_SUCCESS) {
+        SrsPacket* pkt = NULL;
+        if ((ret = protocol->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("identify decode message failed. ret=%d", ret);
             return ret;
         }
         
-        SrsPacket* pkt = msg->get_packet();
+        SrsAutoFree(SrsPacket, pkt, false);
+        
         if (dynamic_cast<SrsCreateStreamPacket*>(pkt)) {
             srs_info("identify client by create stream, play or flash publish.");
             return identify_create_stream_client(dynamic_cast<SrsCreateStreamPacket*>(pkt), stream_id, type, stream_name, duration);
@@ -1002,13 +1029,9 @@ int SrsRtmpServer::set_chunk_size(int chunk_size)
 {
     int ret = ERROR_SUCCESS;
     
-    SrsCommonMessage* msg = new SrsCommonMessage();
     SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket();
-    
     pkt->chunk_size = chunk_size;
-    msg->set_packet(pkt, 0);
+    if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-    
-    if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
         srs_error("send set chunk size message failed. ret=%d", ret);
         return ret;
     }
@@ -1023,14 +1046,10 @@ int SrsRtmpServer::start_play(int stream_id)
     
     // StreamBegin
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsUserControlPacket* pkt = new SrsUserControlPacket();
-        
         pkt->event_type = SrcPCUCStreamBegin;
         pkt->event_data = stream_id;
-        msg->set_packet(pkt, 0);
+        if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send PCUC(StreamBegin) message failed. ret=%d", ret);
             return ret;
         }
@@ -1039,7 +1058,6 @@ int SrsRtmpServer::start_play(int stream_id)
     
     // onStatus(NetStream.Play.Reset)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
@@ -1048,9 +1066,7 @@ int SrsRtmpServer::start_play(int stream_id)
         pkt->data->set(StatusDetails, SrsAmf0Any::str("stream"));
         pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onStatus(NetStream.Play.Reset) message failed. ret=%d", ret);
             return ret;
         }
@@ -1059,7 +1075,6 @@ int SrsRtmpServer::start_play(int stream_id)
     
     // onStatus(NetStream.Play.Start)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
@@ -1068,9 +1083,7 @@ int SrsRtmpServer::start_play(int stream_id)
         pkt->data->set(StatusDetails, SrsAmf0Any::str("stream"));
         pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onStatus(NetStream.Play.Reset) message failed. ret=%d", ret);
             return ret;
         }
@@ -1079,12 +1092,8 @@ int SrsRtmpServer::start_play(int stream_id)
     
     // |RtmpSampleAccess(false, false)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsSampleAccessPacket* pkt = new SrsSampleAccessPacket();
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, stream_id);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send |RtmpSampleAccess(false, false) message failed. ret=%d", ret);
             return ret;
         }
@@ -1093,14 +1102,9 @@ int SrsRtmpServer::start_play(int stream_id)
     
     // onStatus(NetStream.Data.Start)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusDataPacket* pkt = new SrsOnStatusDataPacket();
-        
         pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeDataStart));
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, stream_id);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onStatus(NetStream.Data.Start) message failed. ret=%d", ret);
             return ret;
         }
@@ -1119,16 +1123,13 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
     if (is_pause) {
         // onStatus(NetStream.Pause.Notify)
         if (true) {
-            SrsCommonMessage* msg = new SrsCommonMessage();
             SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
             
             pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
             pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeStreamPause));
             pkt->data->set(StatusDescription, SrsAmf0Any::str("Paused stream."));
             
-            msg->set_packet(pkt, stream_id);
+            if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-            
-            if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
                 srs_error("send onStatus(NetStream.Pause.Notify) message failed. ret=%d", ret);
                 return ret;
             }
@@ -1136,14 +1137,12 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
         }
         // StreamEOF
         if (true) {
-            SrsCommonMessage* msg = new SrsCommonMessage();
             SrsUserControlPacket* pkt = new SrsUserControlPacket();
             
             pkt->event_type = SrcPCUCStreamEOF;
             pkt->event_data = stream_id;
-            msg->set_packet(pkt, 0);
             
-            if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
+            if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
                 srs_error("send PCUC(StreamEOF) message failed. ret=%d", ret);
                 return ret;
             }
@@ -1152,16 +1151,13 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
     } else {
         // onStatus(NetStream.Unpause.Notify)
         if (true) {
-            SrsCommonMessage* msg = new SrsCommonMessage();
             SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
             
             pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
             pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeStreamUnpause));
             pkt->data->set(StatusDescription, SrsAmf0Any::str("Unpaused stream."));
             
-            msg->set_packet(pkt, stream_id);
+            if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-            
-            if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
                 srs_error("send onStatus(NetStream.Unpause.Notify) message failed. ret=%d", ret);
                 return ret;
             }
@@ -1169,14 +1165,12 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
         }
         // StreanBegin
         if (true) {
-            SrsCommonMessage* msg = new SrsCommonMessage();
             SrsUserControlPacket* pkt = new SrsUserControlPacket();
             
             pkt->event_type = SrcPCUCStreamBegin;
             pkt->event_data = stream_id;
-            msg->set_packet(pkt, 0);
             
-            if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
+            if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
                 srs_error("send PCUC(StreanBegin) message failed. ret=%d", ret);
                 return ret;
             }
@@ -1194,25 +1188,21 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
     // FCPublish
     double fc_publish_tid = 0;
     if (true) {
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
         SrsFMLEStartPacket* pkt = NULL;
-        if ((ret = srs_rtmp_expect_message<SrsFMLEStartPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
+        if ((ret = __srs_rtmp_expect_message<SrsFMLEStartPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("recv FCPublish message failed. ret=%d", ret);
             return ret;
         }
         srs_info("recv FCPublish request message success.");
         
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
         fc_publish_tid = pkt->transaction_id;
     }
     // FCPublish response
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(fc_publish_tid);
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, 0);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send FCPublish response message failed. ret=%d", ret);
             return ret;
         }
@@ -1222,25 +1212,21 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
     // createStream
     double create_stream_tid = 0;
     if (true) {
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
         SrsCreateStreamPacket* pkt = NULL;
-        if ((ret = srs_rtmp_expect_message<SrsCreateStreamPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
+        if ((ret = __srs_rtmp_expect_message<SrsCreateStreamPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("recv createStream message failed. ret=%d", ret);
             return ret;
         }
         srs_info("recv createStream request message success.");
         
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
         create_stream_tid = pkt->transaction_id;
     }
     // createStream response
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsCreateStreamResPacket* pkt = new SrsCreateStreamResPacket(create_stream_tid, stream_id);
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, 0);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send createStream response message failed. ret=%d", ret);
             return ret;
         }
@@ -1249,28 +1235,25 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
     
     // publish
     if (true) {
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
         SrsPublishPacket* pkt = NULL;
-        if ((ret = srs_rtmp_expect_message<SrsPublishPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
+        if ((ret = __srs_rtmp_expect_message<SrsPublishPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("recv publish message failed. ret=%d", ret);
             return ret;
         }
         srs_info("recv publish request message success.");
         
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
     }
     // publish response onFCPublish(NetStream.Publish.Start)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->command_name = RTMP_AMF0_COMMAND_ON_FC_PUBLISH;
         pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodePublishStart));
         pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onFCPublish(NetStream.Publish.Start) message failed. ret=%d", ret);
             return ret;
         }
@@ -1278,7 +1261,6 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
     }
     // publish response onStatus(NetStream.Publish.Start)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
@@ -1286,9 +1268,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
         pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
         pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret);
             return ret;
         }
@@ -1306,16 +1286,13 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
     
     // publish response onFCUnpublish(NetStream.unpublish.Success)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->command_name = RTMP_AMF0_COMMAND_ON_FC_UNPUBLISH;
         pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeUnpublishSuccess));
         pkt->data->set(StatusDescription, SrsAmf0Any::str("Stop publishing stream."));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onFCUnpublish(NetStream.unpublish.Success) message failed. ret=%d", ret);
             return ret;
         }
@@ -1323,12 +1300,8 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
     }
     // FCUnpublish response
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(unpublish_tid);
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, stream_id);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send FCUnpublish response message failed. ret=%d", ret);
             return ret;
         }
@@ -1336,7 +1309,6 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
     }
     // publish response onStatus(NetStream.Unpublish.Success)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
@@ -1344,9 +1316,7 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
         pkt->data->set(StatusDescription, SrsAmf0Any::str("Stream is now unpublished"));
         pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onStatus(NetStream.Unpublish.Success) message failed. ret=%d", ret);
             return ret;
         }
@@ -1364,7 +1334,6 @@ int SrsRtmpServer::start_flash_publish(int stream_id)
     
     // publish response onStatus(NetStream.Publish.Start)
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
         
         pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
@@ -1372,9 +1341,7 @@ int SrsRtmpServer::start_flash_publish(int stream_id)
         pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
         pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
         
-        msg->set_packet(pkt, stream_id);
+        if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret);
             return ret;
         }
@@ -1391,12 +1358,8 @@ int SrsRtmpServer::identify_create_stream_client(SrsCreateStreamPacket* req, int
     int ret = ERROR_SUCCESS;
     
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsCreateStreamResPacket* pkt = new SrsCreateStreamResPacket(req->transaction_id, stream_id);
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, 0);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send createStream response message failed. ret=%d", ret);
             return ret;
         }
@@ -1404,13 +1367,13 @@ int SrsRtmpServer::identify_create_stream_client(SrsCreateStreamPacket* req, int
     }
     
     while (true) {
-        SrsCommonMessage* msg = NULL;
+        __SrsMessage* msg = NULL;
-        if ((ret = protocol->recv_message(&msg)) != ERROR_SUCCESS) {
+        if ((ret = protocol->__recv_message(&msg)) != ERROR_SUCCESS) {
             srs_error("recv identify client message failed. ret=%d", ret);
             return ret;
         }
 
-        SrsAutoFree(SrsCommonMessage, msg, false);
+        SrsAutoFree(__SrsMessage, msg, false);
 
         if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
             srs_trace("identify ignore messages except "
@@ -1418,12 +1381,14 @@ int SrsRtmpServer::identify_create_stream_client(SrsCreateStreamPacket* req, int
             continue;
         }
         
-        if ((ret = msg->decode_packet(protocol)) != ERROR_SUCCESS) {
+        SrsPacket* pkt = NULL;
+        if ((ret = protocol->__decode_message(msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("identify decode message failed. ret=%d", ret);
             return ret;
         }
+
+        SrsAutoFree(SrsPacket, pkt, false);
         
-        SrsPacket* pkt = msg->get_packet();
         if (dynamic_cast<SrsPlayPacket*>(pkt)) {
             srs_info("level1 identify client by play.");
             return identify_play_client(dynamic_cast<SrsPlayPacket*>(pkt), type, stream_name, duration);
@@ -1448,12 +1413,8 @@ int SrsRtmpServer::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsRtmp
     
     // releaseStream response
     if (true) {
-        SrsCommonMessage* msg = new SrsCommonMessage();
         SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(req->transaction_id);
-        
+        if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
-        msg->set_packet(pkt, 0);
-        
-        if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
             srs_error("send releaseStream response message failed. ret=%d", ret);
             return ret;
         }

trunk/src/rtmp/srs_protocol_rtmp.hpp

@@ -42,6 +42,8 @@ class SrsPublishPacket;
 class SrsSharedPtrMessage;
 class SrsOnMetaDataPacket;
 class SrsPlayPacket;
+class __SrsMessage;
+class SrsPacket;
 
 /**
 * the original request from client.
@@ -164,6 +166,10 @@ public:
     virtual int get_send_kbps();
     virtual int recv_message(SrsCommonMessage** pmsg);
     virtual int send_message(ISrsMessage* msg);
+    virtual int __recv_message(__SrsMessage** pmsg);
+    virtual int __decode_message(__SrsMessage* msg, SrsPacket** ppacket);
+    virtual int __send_and_free_message(__SrsMessage* msg);
+    virtual int __send_and_free_packet(SrsPacket* packet, int stream_id);
 public:
     // try complex, then simple handshake.
     virtual int handshake();
@@ -209,6 +215,10 @@ public:
     virtual int get_send_kbps();
     virtual int recv_message(SrsCommonMessage** pmsg);
     virtual int send_message(ISrsMessage* msg);
+    virtual int __recv_message(__SrsMessage** pmsg);
+    virtual int __decode_message(__SrsMessage* msg, SrsPacket** ppacket);
+    virtual int __send_and_free_message(__SrsMessage* msg);
+    virtual int __send_and_free_packet(SrsPacket* packet, int stream_id);
 public:
     virtual int handshake();
     virtual int connect_app(SrsRequest* req);

trunk/src/rtmp/srs_protocol_rtmp_stack.hpp

@@ -45,6 +45,9 @@ class SrsChunkStream;
 class SrsAmf0Object;
 class SrsAmf0Any;
 class ISrsMessage;
+class SrsMessageHeader;
+class __SrsMessage;
+class __SrsChunkStream;
  
 // the following is the timeout for rtmp protocol, 
 // to avoid death connection.
@@ -110,6 +113,9 @@ private:
 // peer in
 private:
     std::map<int, SrsChunkStream*> chunk_streams;
+    // TODO: FIXME: rename to chunk_streams
+    std::map<int, __SrsChunkStream*> __chunk_streams;
+    SrsStream* decode_stream;
     SrsBuffer* buffer;
     int32_t in_chunk_size;
     AckWindowSize in_ack_size;
@@ -125,6 +131,7 @@ public:
     SrsProtocol(ISrsProtocolReaderWriter* io);
     virtual ~SrsProtocol();
 public:
+    // TODO: FIXME: to private.
     std::string get_request_name(double transcationId);
     /**
     * set the timeout in us.
@@ -138,6 +145,82 @@ public:
     virtual int64_t get_send_bytes();
     virtual int get_recv_kbps();
     virtual int get_send_kbps();
+public:
+    /**
+    * recv a RTMP message, which is bytes oriented.
+    * user can use decode_message to get the decoded RTMP packet.
+    * @param pmsg, set the received message, 
+    *       always NULL if error, 
+    *       NULL for unknown packet but return success.
+    *       never NULL if decode success.
+    */
+    virtual int __recv_message(__SrsMessage** pmsg);
+    /**
+    * decode bytes oriented RTMP message to RTMP packet,
+    * @param ppacket, output decoded packet, 
+    *       always NULL if error, never NULL if success.
+    * @return error when unknown packet, error when decode failed.
+    */
+    virtual int __decode_message(__SrsMessage* msg, SrsPacket** ppacket);
+    /**
+    * send the RTMP message and always free it.
+    * user must never free or use the msg after this method,
+    * for it will always free the msg.
+    * @param msg, the msg to send out, never be NULL.
+    */
+    virtual int __send_and_free_message(__SrsMessage* msg);
+    /**
+    * send the RTMP packet and always free it.
+    * user must never free or use the packet after this method,
+    * for it will always free the packet.
+    * @param packet, the packet to send out, never be NULL.
+    * @param stream_id, the stream id of packet to send over, 0 for control message.
+    */
+    virtual int __send_and_free_packet(SrsPacket* packet, int stream_id);
+private:
+    /**
+    * imp for __send_and_free_message
+    * @param packet the packet of message, NULL for raw message.
+    */
+    virtual int __do_send_and_free_message(__SrsMessage* msg, SrsPacket* packet);
+    /**
+    * imp for __decode_message
+    */
+    virtual int __do_decode_message(SrsMessageHeader& header, SrsStream* stream, SrsPacket** ppacket);
+    /**
+    * recv bytes oriented RTMP message from protocol stack.
+    * return error if error occur and nerver set the pmsg,
+    * return success and pmsg set to NULL if no entire message got,
+    * return success and pmsg set to entire message if got one.
+    */
+    virtual int __recv_interlaced_message(__SrsMessage** pmsg);
+    /**
+    * read the chunk basic header(fmt, cid) from chunk stream.
+    * user can discovery a SrsChunkStream by cid.
+    * @bh_size return the chunk basic header size, to remove the used bytes when finished.
+    */
+    virtual int __read_basic_header(char& fmt, int& cid, int& bh_size);
+    /**
+    * read the chunk message header(timestamp, payload_length, message_type, stream_id) 
+    * from chunk stream and save to SrsChunkStream.
+    * @mh_size return the chunk message header size, to remove the used bytes when finished.
+    */
+    virtual int __read_message_header(__SrsChunkStream* chunk, char fmt, int bh_size, int& mh_size);
+    /**
+    * read the chunk payload, remove the used bytes in buffer,
+    * if got entire message, set the pmsg.
+    * @payload_size read size in this roundtrip, generally a chunk size or left message size.
+    */
+    virtual int __read_message_payload(__SrsChunkStream* chunk, int bh_size, int mh_size, int& payload_size, __SrsMessage** pmsg);
+    /**
+    * when recv message, update the context.
+    */
+    virtual int __on_recv_message(__SrsMessage* msg);
+    /**
+    * when message sentout, update the context.
+    */
+    virtual int __on_send_message(__SrsMessage* msg, SrsPacket* packet);
+public:
     /**
     * recv a message with raw/undecoded payload from peer.
     * the payload is not decoded, use srs_rtmp_expect_message<T> if requires 
@@ -226,6 +309,16 @@ struct SrsMessageHeader
     */
     int64_t timestamp;
     
+public:
+    /**
+    * get the perfered cid(chunk stream id) which sendout over.
+    * set at decoding, and canbe used for directly send message,
+    * for example, dispatch to all connections.
+    * @see: SrsSharedPtrMessage.SrsSharedPtr.perfer_cid
+    */
+    int perfer_cid;
+    
+public:
     SrsMessageHeader();
     virtual ~SrsMessageHeader();
 
@@ -283,6 +376,109 @@ public:
     virtual ~SrsChunkStream();
 };
 
+/**
+* incoming chunk stream maybe interlaced,
+* use the chunk stream to cache the input RTMP chunk streams.
+*/
+class __SrsChunkStream
+{
+public:
+    /**
+    * represents the basic header fmt,
+    * which used to identify the variant message header type.
+    */
+    char fmt;
+    /**
+    * represents the basic header cid,
+    * which is the chunk stream id.
+    */
+    int cid;
+    /**
+    * cached message header
+    */
+    SrsMessageHeader header;
+    /**
+    * whether the chunk message header has extended timestamp.
+    */
+    bool extended_timestamp;
+    /**
+    * partially read message.
+    */
+    __SrsMessage* msg;
+    /**
+    * decoded msg count, to identify whether the chunk stream is fresh.
+    */
+    int64_t msg_count;
+public:
+    __SrsChunkStream(int _cid);
+    virtual ~__SrsChunkStream();
+};
+
+/**
+* message is raw data RTMP message, bytes oriented,
+* protcol always recv RTMP message, and can send RTMP message or RTMP packet.
+* the shared-ptr message is a special RTMP message, use ref-count for performance issue.
+*/
+class __SrsMessage
+{
+// 4.1. Message Header
+public:
+    SrsMessageHeader header;
+// 4.2. Message Payload
+public:
+    /**
+    * The other part which is the payload is the actual data that is
+    * contained in the message. For example, it could be some audio samples
+    * or compressed video data. The payload format and interpretation are
+    * beyond the scope of this document.
+    */
+    int32_t size;
+    int8_t* payload;
+public:
+    __SrsMessage();
+    virtual ~__SrsMessage();
+};
+
+/**
+* shared ptr message.
+* for audio/video/data message that need less memory copy.
+* and only for output.
+*/
+class __SrsSharedPtrMessage : public __SrsMessage
+{
+private:
+    struct __SrsSharedPtr
+    {
+        char* payload;
+        int size;
+        int shared_count;
+        
+        __SrsSharedPtr();
+        virtual ~__SrsSharedPtr();
+    };
+    __SrsSharedPtr* ptr;
+public:
+    __SrsSharedPtrMessage();
+    virtual ~__SrsSharedPtrMessage();
+public:
+    /**
+    * set the shared payload.
+    * we will detach the payload of source,
+    * so ensure donot use it before.
+    */
+    virtual int initialize(__SrsMessage* source);
+    /**
+    * set the shared payload.
+    * use source header, and specified param payload.
+    */
+    virtual int initialize(SrsMessageHeader* source, char* payload, int size);
+public:
+    /**
+    * copy current shared ptr message, use ref-count.
+    */
+    virtual __SrsSharedPtrMessage* copy();
+};
+
 /**
 * message to output.
 */
@@ -1215,5 +1411,44 @@ int srs_rtmp_expect_message(SrsProtocol* protocol, SrsCommonMessage** pmsg, T**
     
     return ret;
 }
+template<class T>
+int __srs_rtmp_expect_message(SrsProtocol* protocol, __SrsMessage** pmsg, T** ppacket)
+{
+    *pmsg = NULL;
+    *ppacket = NULL;
+    
+    int ret = ERROR_SUCCESS;
+    
+    while (true) {
+        __SrsMessage* msg = NULL;
+        if ((ret = protocol->__recv_message(&msg)) != ERROR_SUCCESS) {
+            srs_error("recv message failed. ret=%d", ret);
+            return ret;
+        }
+        srs_verbose("recv message success.");
+        
+        SrsPacket* packet = NULL;
+        if ((ret = protocol->__decode_message(msg, &packet)) != ERROR_SUCCESS) {
+            srs_error("decode message failed. ret=%d", ret);
+            srs_freep(msg);
+            return ret;
+        }
+        
+        T* pkt = dynamic_cast<T*>(packet);
+        if (!pkt) {
+            srs_trace("drop message(type=%d, size=%d, time=%"PRId64", sid=%d).", 
+                msg->header.message_type, msg->header.payload_length,
+                msg->header.timestamp, msg->header.stream_id);
+            srs_freep(msg);
+            continue;
+        }
+        
+        *pmsg = msg;
+        *ppacket = pkt;
+        break;
+    }
+    
+    return ret;
+}
 
 #endif