Threads-SRTP: Use async encrypt SRTP packet

1. Async SRTP support protect RTCP.
2. Send packet ignore when encrypt size is 0.
3. Callback to send packet if encrypt done.

trunk/src/app/srs_app_rtc_conn.cpp

@@ -214,6 +214,11 @@ srs_error_t SrsSecurityTransport::on_rtp_cipher(char* cipher, int size)
     return session_->on_rtp_cipher(cipher, size);
 }
 
+srs_error_t SrsSecurityTransport::on_rtcp_cipher(char* cipher, int size)
+{
+    return session_->on_rtcp_cipher(cipher, size);
+}
+
 srs_error_t SrsSecurityTransport::protect_rtp(void* packet, int* nb_cipher)
 {
     return srtp_->protect_rtp(packet, nb_cipher);
@@ -2069,6 +2074,17 @@ srs_error_t SrsRtcConnection::on_rtp_cipher(char* cipher, int size)
     return err;
 }
 
+srs_error_t SrsRtcConnection::on_rtcp_cipher(char* cipher, int size)
+{
+    srs_error_t err = srs_success;
+
+    if ((err = sendonly_skt->sendto(cipher, size, 0)) != srs_success) {
+        srs_error_reset(err); // Ignore any error.
+    }
+
+    return err;
+}
+
 srs_error_t SrsRtcConnection::dispatch_rtcp(SrsRtcpCommon* rtcp)
 {
     srs_error_t err = srs_success;
@@ -2418,11 +2434,12 @@ srs_error_t SrsRtcConnection::send_rtcp(char *data, int nb_data)
         return srs_error_wrap(err, "protect rtcp");
     }
 
-    if ((err = sendonly_skt->sendto(data, nb_buf, 0)) != srs_success) {
-        return srs_error_wrap(err, "send");
+    // Async SRTP encrypt.
+    if (nb_buf <= 0) {
+        return err;
     }
 
-    return err;
+    return on_rtcp_cipher(data, nb_buf);
 }
 
 void SrsRtcConnection::check_send_nacks(SrsRtpNackForReceiver* nack, uint32_t ssrc, uint32_t& sent_nacks, uint32_t& timeout_nacks)
@@ -2622,15 +2639,11 @@ srs_error_t SrsRtcConnection::do_send_packet(SrsRtpPacket2* pkt)
 
     ++_srs_pps_srtps->sugar;
 
-    if ((err = sendonly_skt->sendto(iov->iov_base, iov->iov_len, 0)) != srs_success) {
-        srs_error_reset(err); // Ignore any error.
-    }
-
     // Detail log, should disable it in release version.
     srs_info("RTC: SEND PT=%u, SSRC=%#x, SEQ=%u, Time=%u, %u/%u bytes", pkt->header.get_payload_type(), pkt->header.get_ssrc(),
         pkt->header.get_sequence(), pkt->header.get_timestamp(), pkt->nb_bytes(), iov->iov_len);
 
-    return err;
+    return on_rtp_cipher((char*)iov->iov_base, iov->iov_len);
 }
 
 void SrsRtcConnection::set_all_tracks_status(std::string stream_uri, bool is_publish, bool status)

trunk/src/app/srs_app_rtc_conn.hpp

@@ -146,6 +146,7 @@ class SrsSecurityTransport : public ISrsRtcTransport
     srs_error_t on_rtp_plaintext(char* plaintext, int size);
     srs_error_t on_rtcp_plaintext(char* plaintext, int size);
     srs_error_t on_rtp_cipher(char* cipher, int size);
+    srs_error_t on_rtcp_cipher(char* cipher, int size);
 };
 
 // Semi security transport, setup DTLS and SRTP, with SRTP decrypt, without SRTP encrypt.
@@ -499,6 +500,7 @@ class SrsRtcConnection : public ISrsResource
     srs_error_t on_rtcp_plaintext(char* plaintext, int size);
 private:
     srs_error_t on_rtp_cipher(char* cipher, int size);
+    srs_error_t on_rtcp_cipher(char* cipher, int size);
 private:
     srs_error_t dispatch_rtcp(SrsRtcpCommon* rtcp);
 public:

trunk/src/app/srs_app_threads.cpp

@@ -810,8 +810,23 @@ srs_error_t SrsAsyncSRTP::protect_rtcp(void* packet, int* nb_cipher)
         return srs_error_new(ERROR_RTC_SRTP_UNPROTECT, "not ready");
     }
 
-    // TODO: FIMXE: Remove it.
-    return SrsSRTP::protect_rtcp(packet, nb_cipher);
+    int nb_plaintext = *nb_cipher;
+
+    // Note that we must allocate more bytes than the size of packet, because SRTP
+    // will write checksum at the end of buffer.
+    char* buf = new char[kRtcpPacketSize];
+    memcpy(buf, packet, nb_plaintext);
+
+    SrsAsyncSRTPPacket* pkt = new SrsAsyncSRTPPacket(task_);
+    pkt->msg_->wrap(buf, nb_plaintext);
+    pkt->is_rtp_ = false;
+    pkt->do_decrypt_ = false;
+    _srs_async_srtp->add_packet(pkt);
+
+    // Do the job asynchronously.
+    *nb_cipher = 0;
+
+    return srs_success;
 }
 
 srs_error_t SrsAsyncSRTP::unprotect_rtp(void* packet, int* nb_plaintext)
@@ -910,6 +925,8 @@ srs_error_t SrsAsyncSRTPTask::cook(SrsAsyncSRTPPacket* pkt)
     } else {
         if (pkt->is_rtp_) {
             err = impl_->protect_rtp(pkt->msg_->payload, &pkt->nb_consumed_);
+        } else {
+            err = impl_->protect_rtcp(pkt->msg_->payload, &pkt->nb_consumed_);
         }
     }
     if (err != srs_success) {
@@ -939,6 +956,8 @@ srs_error_t SrsAsyncSRTPTask::consume(SrsAsyncSRTPPacket* pkt)
     } else {
         if (pkt->is_rtp_) {
             err = codec_->transport_->on_rtp_cipher(payload, pkt->nb_consumed_);
+        } else {
+            err = codec_->transport_->on_rtcp_cipher(payload, pkt->nb_consumed_);
         }
     }