for #730, reset ack follow flash player rules. 2.0.225

README.md

@@ -345,6 +345,7 @@ Remark:
 
 ## History
 
+* v2.0, 2017-01-06, for #730, reset ack follow flash player rules. 2.0.225
 * v2.0, 2016-12-15, for #513, remove hls ram from srs2 to srs3+. 2.0.224
 * <strong>v2.0, 2016-12-13, [2.0 beta3(2.0.223)][r2.0b3] released. 86685 lines.</strong>
 * v2.0, 2016-12-13, fix #713, disable the source cleanup. 2.0.223

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       2
 #define VERSION_MINOR       0
-#define VERSION_REVISION    224
+#define VERSION_REVISION    225
 
 // generated by configure, only macros.
 #include <srs_auto_headers.hpp>

trunk/src/protocol/srs_rtmp_stack.cpp

@@ -210,8 +210,8 @@ int SrsPacket::encode_packet(SrsStream* stream)
 
 SrsProtocol::AckWindowSize::AckWindowSize()
 {
-    ack_window_size = 0;
+    window = 0;
-    acked_size = 0;
+    sequence_number = nb_recv_bytes = 0;
 }
 
 SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io)
@@ -1457,13 +1457,9 @@ int SrsProtocol::on_recv_message(SrsCommonMessage* msg)
     
     srs_assert(msg != NULL);
         
-    // acknowledgement
+    // try to response acknowledgement
-    if (in_ack_size.ack_window_size > 0 
+    if ((ret = response_acknowledgement_message()) != ERROR_SUCCESS) {
-        && skt->get_recv_bytes() - in_ack_size.acked_size > in_ack_size.ack_window_size
+        return ret;
-    ) {
-        if ((ret = response_acknowledgement_message()) != ERROR_SUCCESS) {
-            return ret;
-        }
     }
     
     SrsPacket* packet = NULL;
@@ -1492,7 +1488,7 @@ int SrsProtocol::on_recv_message(SrsCommonMessage* msg)
             srs_assert(pkt != NULL);
             
             if (pkt->ackowledgement_window_size > 0) {
-                in_ack_size.ack_window_size = pkt->ackowledgement_window_size;
+                in_ack_size.window = (uint32_t)pkt->ackowledgement_window_size;
                 // @remark, we ignore this message, for user noneed to care.
                 // but it's important for dev, for client/server will block if required 
                 // ack msg not arrived.
@@ -1526,7 +1522,7 @@ int SrsProtocol::on_recv_message(SrsCommonMessage* msg)
             }
             
             in_chunk_size = pkt->chunk_size;
-            srs_trace("input chunk size to %d", pkt->chunk_size);
+            srs_info("in.chunk=%d", pkt->chunk_size);
 
             break;
         }
@@ -1535,7 +1531,8 @@ int SrsProtocol::on_recv_message(SrsCommonMessage* msg)
             srs_assert(pkt != NULL);
             
             if (pkt->event_type == SrcPCUCSetBufferLength) {
-                srs_trace("ignored. set buffer length to %d", pkt->extra_data);
+                srs_trace("buffer=%d, in.ack=%d, out.ack=%d, in.chunk=%d, out.chunk=%d", pkt->extra_data,
+                    in_ack_size.window, out_ack_size.window, in_chunk_size, out_chunk_size);
             }
             if (pkt->event_type == SrcPCUCPingRequest) {
                 if ((ret = response_ping_message(pkt->event_data)) != ERROR_SUCCESS) {
@@ -1563,11 +1560,13 @@ int SrsProtocol::on_send_packet(SrsMessageHeader* mh, SrsPacket* packet)
     switch (mh->message_type) {
         case RTMP_MSG_SetChunkSize: {
             SrsSetChunkSizePacket* pkt = dynamic_cast<SrsSetChunkSizePacket*>(packet);
-            srs_assert(pkt != NULL);
-            
             out_chunk_size = pkt->chunk_size;
-            
+            srs_info("out.chunk=%d", pkt->chunk_size);
-            srs_trace("out chunk size to %d", pkt->chunk_size);
+            break;
+        }
+        case RTMP_MSG_WindowAcknowledgementSize: {
+            SrsSetWindowAckSizePacket* pkt = dynamic_cast<SrsSetWindowAckSizePacket*>(packet);
+            out_ack_size.window = (uint32_t)pkt->ackowledgement_window_size;
             break;
         }
         case RTMP_MSG_AMF0CommandMessage:
@@ -1606,9 +1605,27 @@ int SrsProtocol::response_acknowledgement_message()
 {
     int ret = ERROR_SUCCESS;
     
+    if (in_ack_size.window <= 0) {
+        return ret;
+    }
+    
+    // ignore when delta bytes not exceed half of window(ack size).
+    uint32_t delta = (uint32_t)(skt->get_recv_bytes() - in_ack_size.nb_recv_bytes);
+    if (delta < in_ack_size.window / 2) {
+        return ret;
+    }
+    in_ack_size.nb_recv_bytes = skt->get_recv_bytes();
+    
+    // when the sequence number overflow, reset it.
+    uint32_t sequence_number = in_ack_size.sequence_number + delta;
+    if (sequence_number > 0xf0000000) {
+        sequence_number = delta;
+    }
+    in_ack_size.sequence_number = sequence_number;
+    
     SrsAcknowledgementPacket* pkt = new SrsAcknowledgementPacket();
-    in_ack_size.acked_size = skt->get_recv_bytes();
+    pkt->sequence_number = sequence_number;
-    pkt->sequence_number = (int32_t)in_ack_size.acked_size;
+    srs_warn("ack sequence=%#x", sequence_number);
     
     // cache the message and use flush to send.
     if (!auto_response_when_recv) {
@@ -5017,7 +5034,7 @@ SrsAcknowledgementPacket::~SrsAcknowledgementPacket()
 {
 }
 
-int SrsAcknowledgementPacket::decode(SrsBuffer* stream)
+int SrsAcknowledgementPacket::decode(SrsStream* stream)
 {
     int ret = ERROR_SUCCESS;
     
@@ -5027,7 +5044,7 @@ int SrsAcknowledgementPacket::decode(SrsBuffer* stream)
         return ret;
     }
     
-    sequence_number = stream->read_4bytes();
+    sequence_number = (uint32_t)stream->read_4bytes();
     srs_info("decode acknowledgement success");
     
     return ret;

trunk/src/protocol/srs_rtmp_stack.hpp

@@ -190,8 +190,11 @@ private:
     class AckWindowSize
     {
     public:
-        int ack_window_size;
+        uint32_t window;
-        int64_t acked_size;
+        // number of received bytes.
+        int64_t nb_recv_bytes;
+        // previous responsed sequence number.
+        uint32_t sequence_number;
         
         AckWindowSize();
     };
@@ -227,10 +230,11 @@ private:
     * input chunk size, default to 128, set by peer packet.
     */
     int32_t in_chunk_size;
-    /**
+    // The input ack window, to response acknowledge to peer,
-    * input ack size, when to send the acked packet.
+    // for example, to respose the encoder, for server got lots of packets.
-    */
     AckWindowSize in_ack_size;
+    // The output ack window, to require peer to response the ack.
+    AckWindowSize out_ack_size;
     /**
     * whether auto response when recv messages.
     * default to true for it's very easy to use the protocol stack.
@@ -1843,13 +1847,13 @@ protected:
 class SrsAcknowledgementPacket : public SrsPacket
 {
 public:
-    int32_t sequence_number;
+    uint32_t sequence_number;
 public:
     SrsAcknowledgementPacket();
     virtual ~SrsAcknowledgementPacket();
 // decode functions for concrete packet to override.
 public:
-    virtual int decode(SrsBuffer* stream);
+    virtual int decode(SrsStream* stream);
 // encode functions for concrete packet to override.
 public:
     virtual int get_prefer_cid();