Threads-SRTP: Support async decrypt RTCP

1. SrsAsyncSRTP support unprotect_rtcp packet.
2. Extract on_rtcp_plaintext from on_rtcp.

trunk/src/app/srs_app_rtc_conn.cpp

@@ -204,6 +204,11 @@ srs_error_t SrsSecurityTransport::on_rtp_plaintext(char* plaintext, int size)
     return session_->on_rtp_plaintext(plaintext, size);
 }
 
+srs_error_t SrsSecurityTransport::on_rtcp_plaintext(char* plaintext, int size)
+{
+    return session_->on_rtcp_plaintext(plaintext, size);
+}
+
 srs_error_t SrsSecurityTransport::protect_rtp(void* packet, int* nb_cipher)
 {
     return srtp_->protect_rtp(packet, nb_cipher);
@@ -2001,6 +2006,23 @@ srs_error_t SrsRtcConnection::on_rtcp(char* data, int nb_data)
         return srs_error_wrap(err, "rtcp unprotect");
     }
 
+    // For async SRTP, the nb_unprotected_buf might be zero, which means we do not got the plaintext
+    // right now, and it will callback if get one.
+    if (nb_unprotected_buf == 0) {
+        return err;
+    }
+
+    if ((err = on_rtcp_plaintext(data, nb_unprotected_buf)) != srs_success) {
+        return srs_error_wrap(err, "cipher=%d", nb_data);
+    }
+
+    return err;
+}
+
+srs_error_t SrsRtcConnection::on_rtcp_plaintext(char* data, int nb_unprotected_buf)
+{
+    srs_error_t err = srs_success;
+
     char* unprotected_buf = data;
     if (_srs_blackhole->blackhole) {
         _srs_blackhole->sendto(unprotected_buf, nb_unprotected_buf);
@@ -2022,7 +2044,7 @@ srs_error_t SrsRtcConnection::on_rtcp(char* data, int nb_data)
         SrsAutoFree(SrsRtcpCommon, rtcp);
 
         if(srs_success != err) {
-            return srs_error_wrap(err, "cipher=%u, plaintext=%u, bytes=[%s], rtcp=(%u,%u,%u,%u)", nb_data, nb_unprotected_buf,
+            return srs_error_wrap(err, "plaintext=%u, bytes=[%s], rtcp=(%u,%u,%u,%u)", nb_unprotected_buf,
                 srs_string_dumps_hex(rtcp->data(), rtcp->size(), rtcp->size()).c_str(),
                 rtcp->get_rc(), rtcp->type(), rtcp->get_ssrc(), rtcp->size());
         }

trunk/src/app/srs_app_rtc_conn.hpp

@@ -144,6 +144,7 @@ class SrsSecurityTransport : public ISrsRtcTransport
     srs_error_t srtp_initialize();
 public:
     srs_error_t on_rtp_plaintext(char* plaintext, int size);
+    srs_error_t on_rtcp_plaintext(char* plaintext, int size);
 };
 
 // Semi security transport, setup DTLS and SRTP, with SRTP decrypt, without SRTP encrypt.
@@ -493,6 +494,8 @@ class SrsRtcConnection : public ISrsResource
     srs_error_t find_publisher(char* buf, int size, SrsRtcPublishStream** ppublisher);
 public:
     srs_error_t on_rtcp(char* data, int nb_data);
+private:
+    srs_error_t on_rtcp_plaintext(char* plaintext, int size);
 private:
     srs_error_t dispatch_rtcp(SrsRtcpCommon* rtcp);
 public:

trunk/src/app/srs_app_threads.cpp

@@ -814,7 +814,9 @@ srs_error_t SrsAsyncSRTP::unprotect_rtp(void* packet, int* nb_plaintext)
     _srs_async_srtp->add_packet(pkt);
 
     // Do the job asynchronously.
-    *nb_plaintext = 0;
+    if (nb_plaintext) {
+        *nb_plaintext = 0;
+    }
 
     return srs_success;
 }
@@ -825,8 +827,22 @@ srs_error_t SrsAsyncSRTP::unprotect_rtcp(void* packet, int* nb_plaintext)
         return srs_error_new(ERROR_RTC_SRTP_UNPROTECT, "not ready");
     }
 
-    // TODO: FIMXE: Remove it.
-    return SrsSRTP::unprotect_rtcp(packet, nb_plaintext);
+    int nb_cipher = *nb_plaintext;
+    char* buf = new char[nb_cipher];
+    memcpy(buf, packet, nb_cipher);
+
+    SrsAsyncSRTPPacket* pkt = new SrsAsyncSRTPPacket(task_);
+    pkt->msg_->wrap(buf, nb_cipher);
+    pkt->is_rtp_ = false;
+    pkt->do_decrypt_ = true;
+    _srs_async_srtp->add_packet(pkt);
+
+    // Do the job asynchronously.
+    if (nb_plaintext) {
+        *nb_plaintext = 0;
+    }
+
+    return srs_success;
 }
 
 SrsAsyncSRTPTask::SrsAsyncSRTPTask(SrsAsyncSRTP* codec)
@@ -871,10 +887,12 @@ srs_error_t SrsAsyncSRTPTask::cook(SrsAsyncSRTPPacket* pkt)
         return err;
     }
 
+    pkt->nb_consumed_ = pkt->msg_->size;
     if (pkt->do_decrypt_) {
         if (pkt->is_rtp_) {
-            pkt->nb_consumed_ = pkt->msg_->size;
             err = impl_->unprotect_rtp(pkt->msg_->payload, &pkt->nb_consumed_);
+        } else {
+            err = impl_->unprotect_rtcp(pkt->msg_->payload, &pkt->nb_consumed_);
         }
     }
     if (err != srs_success) {
@@ -898,6 +916,8 @@ srs_error_t SrsAsyncSRTPTask::consume(SrsAsyncSRTPPacket* pkt)
     if (pkt->do_decrypt_) {
         if (pkt->is_rtp_) {
             err = codec_->transport_->on_rtp_plaintext(payload, pkt->nb_consumed_);
+        } else {
+            err = codec_->transport_->on_rtcp_plaintext(payload, pkt->nb_consumed_);
         }
     }