Upintegreate AAH TX and RX players from ICS_AAH

Upintegrate the android at home TX and RX players developed in the
ICS_AAH branch.

Change-Id: I8247d3702e30d8b0e215b31a92675d8ab28dccbb
Signed-off-by: John Grossman <johngro@google.com>

1 files changed, 807 insertions, 0 deletions

diff --git a/media/libaah_rtp/aah_rx_player_core.cpp b/media/libaah_rtp/aah_rx_player_core.cpp
new file mode 100644
index 0000000..d2b3386
--- /dev/null
+++ b/media/libaah_rtp/aah_rx_player_core.cpp
@@ -0,0 +1,807 @@
+/*
+ * Copyright (C) 2011 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#define LOG_TAG "LibAAH_RTP"
+//#define LOG_NDEBUG 0
+#include <utils/Log.h>
+
+#include <fcntl.h>
+#include <poll.h>
+#include <sys/socket.h>
+#include <time.h>
+#include <utils/misc.h>
+
+#include <media/stagefright/Utils.h>
+
+#include "aah_rx_player.h"
+#include "aah_tx_packet.h"
+
+namespace android {
+
+const uint32_t AAH_RXPlayer::kRetransRequestMagic =
+    FOURCC('T','r','e','q');
+const uint32_t AAH_RXPlayer::kRetransNAKMagic =
+    FOURCC('T','n','a','k');
+const uint32_t AAH_RXPlayer::kFastStartRequestMagic =
+    FOURCC('T','f','s','t');
+const uint32_t AAH_RXPlayer::kGapRerequestTimeoutUSec = 75000;
+const uint32_t AAH_RXPlayer::kFastStartTimeoutUSec = 800000;
+const uint32_t AAH_RXPlayer::kRTPActivityTimeoutUSec = 10000000;
+
+static inline int16_t fetchInt16(uint8_t* data) {
+    return static_cast<int16_t>(U16_AT(data));
+}
+
+static inline int32_t fetchInt32(uint8_t* data) {
+    return static_cast<int32_t>(U32_AT(data));
+}
+
+static inline int64_t fetchInt64(uint8_t* data) {
+    return static_cast<int64_t>(U64_AT(data));
+}
+
+uint64_t AAH_RXPlayer::monotonicUSecNow() {
+    struct timespec now;
+    int res = clock_gettime(CLOCK_MONOTONIC, &now);
+    CHECK(res >= 0);
+
+    uint64_t ret = static_cast<uint64_t>(now.tv_sec) * 1000000;
+    ret += now.tv_nsec / 1000;
+
+    return ret;
+}
+
+status_t AAH_RXPlayer::startWorkThread() {
+    status_t res;
+    stopWorkThread();
+    res = thread_wrapper_->run("TRX_Player", PRIORITY_AUDIO);
+
+    if (res != OK) {
+        ALOGE("Failed to start work thread (res = %d)", res);
+    }
+
+    return res;
+}
+
+void AAH_RXPlayer::stopWorkThread() {
+    thread_wrapper_->requestExit();  // set the exit pending flag
+    wakeup_work_thread_evt_.setEvent();
+
+    status_t res;
+    res = thread_wrapper_->requestExitAndWait(); // block until thread exit.
+    if (res != OK) {
+        ALOGE("Failed to stop work thread (res = %d)", res);
+    }
+
+    wakeup_work_thread_evt_.clearPendingEvents();
+}
+
+void AAH_RXPlayer::cleanupSocket() {
+    if (sock_fd_ >= 0) {
+        if (multicast_joined_) {
+            int res;
+            struct ip_mreq mreq;
+            mreq.imr_multiaddr = listen_addr_.sin_addr;
+            mreq.imr_interface.s_addr = htonl(INADDR_ANY);
+            res = setsockopt(sock_fd_,
+                             IPPROTO_IP,
+                             IP_DROP_MEMBERSHIP,
+                             &mreq, sizeof(mreq));
+            if (res < 0) {
+                ALOGW("Failed to leave multicast group. (%d, %d)", res, errno);
+            }
+            multicast_joined_ = false;
+        }
+
+        close(sock_fd_);
+        sock_fd_ = -1;
+    }
+
+    resetPipeline();
+}
+
+void AAH_RXPlayer::resetPipeline() {
+    ring_buffer_.reset();
+
+    // Explicitly shudown all of the active substreams, then call clear out the
+    // collection.  Failure to clear out a substream can result in its decoder
+    // holding a reference to itself and therefor not going away when the
+    // collection is cleared.
+    for (size_t i = 0; i < substreams_.size(); ++i)
+        substreams_.valueAt(i)->shutdown();
+
+    substreams_.clear();
+
+    current_gap_status_ = kGS_NoGap;
+}
+
+bool AAH_RXPlayer::setupSocket() {
+    long flags;
+    int  res, buf_size;
+    socklen_t opt_size;
+
+    cleanupSocket();
+    CHECK(sock_fd_ < 0);
+
+    // Make the socket
+    sock_fd_ = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    if (sock_fd_ < 0) {
+        ALOGE("Failed to create listen socket (errno %d)", errno);
+        goto bailout;
+    }
+
+    // Set non-blocking operation
+    flags = fcntl(sock_fd_, F_GETFL);
+    res   = fcntl(sock_fd_, F_SETFL, flags | O_NONBLOCK);
+    if (res < 0) {
+        ALOGE("Failed to set socket (%d) to non-blocking mode (errno %d)",
+              sock_fd_, errno);
+        goto bailout;
+    }
+
+    // Bind to our port
+    struct sockaddr_in bind_addr;
+    memset(&bind_addr, 0, sizeof(bind_addr));
+    bind_addr.sin_family = AF_INET;
+    bind_addr.sin_addr.s_addr = INADDR_ANY;
+    bind_addr.sin_port = listen_addr_.sin_port;
+    res = bind(sock_fd_,
+               reinterpret_cast<const sockaddr*>(&bind_addr),
+               sizeof(bind_addr));
+    if (res < 0) {
+        uint32_t a = ntohl(bind_addr.sin_addr.s_addr);
+        uint16_t p = ntohs(bind_addr.sin_port);
+        ALOGE("Failed to bind socket (%d) to %d.%d.%d.%d:%hd. (errno %d)",
+              sock_fd_,
+              (a >> 24) & 0xFF,
+              (a >> 16) & 0xFF,
+              (a >>  8) & 0xFF,
+              (a      ) & 0xFF,
+              p,
+              errno);
+
+        goto bailout;
+    }
+
+    buf_size = 1 << 16;   // 64k
+    res = setsockopt(sock_fd_,
+                     SOL_SOCKET, SO_RCVBUF,
+                     &buf_size, sizeof(buf_size));
+    if (res < 0) {
+        ALOGW("Failed to increase socket buffer size to %d.  (errno %d)",
+              buf_size, errno);
+    }
+
+    buf_size = 0;
+    opt_size = sizeof(buf_size);
+    res = getsockopt(sock_fd_,
+                     SOL_SOCKET, SO_RCVBUF,
+                     &buf_size, &opt_size);
+    if (res < 0) {
+        ALOGW("Failed to fetch socket buffer size.  (errno %d)", errno);
+    } else {
+        ALOGI("RX socket buffer size is now %d bytes",  buf_size);
+    }
+
+    if (listen_addr_.sin_addr.s_addr) {
+        // Join the multicast group and we should be good to go.
+        struct ip_mreq mreq;
+        mreq.imr_multiaddr = listen_addr_.sin_addr;
+        mreq.imr_interface.s_addr = htonl(INADDR_ANY);
+        res = setsockopt(sock_fd_,
+                         IPPROTO_IP,
+                         IP_ADD_MEMBERSHIP,
+                         &mreq, sizeof(mreq));
+        if (res < 0) {
+            ALOGE("Failed to join multicast group. (errno %d)", errno);
+            goto bailout;
+        }
+        multicast_joined_ = true;
+    }
+
+    return true;
+
+bailout:
+    cleanupSocket();
+    return false;
+}
+
+bool AAH_RXPlayer::threadLoop() {
+    struct pollfd poll_fds[2];
+    bool process_more_right_now = false;
+
+    if (!setupSocket()) {
+        sendEvent(MEDIA_ERROR);
+        goto bailout;
+    }
+
+    while (!thread_wrapper_->exitPending()) {
+        // Step 1: Wait until there is something to do.
+        int gap_timeout = computeNextGapRetransmitTimeout();
+        int ring_timeout = ring_buffer_.computeInactivityTimeout();
+        int timeout = -1;
+
+        if (!ring_timeout) {
+            ALOGW("RTP inactivity timeout reached, resetting pipeline.");
+            resetPipeline();
+            timeout = gap_timeout;
+        } else {
+            if (gap_timeout < 0) {
+                timeout = ring_timeout;
+            } else if (ring_timeout < 0) {
+                timeout = gap_timeout;
+            } else {
+                timeout = (gap_timeout < ring_timeout) ? gap_timeout
+                                                       : ring_timeout;
+            }
+        }
+
+        if ((0 != timeout) && (!process_more_right_now)) {
+            // Set up the events to wait on.  Start with the wakeup pipe.
+            memset(&poll_fds, 0, sizeof(poll_fds));
+            poll_fds[0].fd     = wakeup_work_thread_evt_.getWakeupHandle();
+            poll_fds[0].events = POLLIN;
+
+            // Add the RX socket.
+            poll_fds[1].fd     = sock_fd_;
+            poll_fds[1].events = POLLIN;
+
+            // Wait for something interesing to happen.
+            int poll_res = poll(poll_fds, NELEM(poll_fds), timeout);
+            if (poll_res < 0) {
+                ALOGE("Fatal error (%d,%d) while waiting on events",
+                      poll_res, errno);
+                sendEvent(MEDIA_ERROR);
+                goto bailout;
+            }
+        }
+
+        if (thread_wrapper_->exitPending()) {
+            break;
+        }
+
+        wakeup_work_thread_evt_.clearPendingEvents();
+        process_more_right_now = false;
+
+        // Step 2: Do we have data waiting in the socket?  If so, drain the
+        // socket moving valid RTP information into the ring buffer to be
+        // processed.
+        if (poll_fds[1].revents) {
+            struct sockaddr_in from;
+            socklen_t from_len;
+
+            ssize_t res = 0;
+            while (!thread_wrapper_->exitPending()) {
+                // Check the size of any pending packet.
+                res = recv(sock_fd_, NULL, 0, MSG_PEEK | MSG_TRUNC);
+
+                // Error?
+                if (res < 0) {
+                    // If the error is anything other than would block,
+                    // something has gone very wrong.
+                    if ((errno != EAGAIN) && (errno != EWOULDBLOCK)) {
+                        ALOGE("Fatal socket error during recvfrom (%d, %d)",
+                              (int)res, errno);
+                        goto bailout;
+                    }
+
+                    // Socket is out of data, just break out of processing and
+                    // wait for more.
+                    break;
+                }
+
+                // Allocate a payload.
+                PacketBuffer* pb = PacketBuffer::allocate(res);
+                if (NULL == pb) {
+                    ALOGE("Fatal error, failed to allocate packet buffer of"
+                          " length %u", static_cast<uint32_t>(res));
+                    goto bailout;
+                }
+
+                // Fetch the data.
+                from_len = sizeof(from);
+                res = recvfrom(sock_fd_, pb->data_, pb->length_, 0,
+                               reinterpret_cast<struct sockaddr*>(&from),
+                               &from_len);
+                if (res != pb->length_) {
+                    ALOGE("Fatal error, fetched packet length (%d) does not"
+                          " match peeked packet length (%u).  This should never"
+                          " happen.  (errno = %d)",
+                          static_cast<int>(res),
+                          static_cast<uint32_t>(pb->length_),
+                          errno);
+                }
+
+                bool drop_packet = false;
+                if (transmitter_known_) {
+                    if (from.sin_addr.s_addr !=
+                        transmitter_addr_.sin_addr.s_addr) {
+                        uint32_t a = ntohl(from.sin_addr.s_addr);
+                        uint16_t p = ntohs(from.sin_port);
+                        ALOGV("Dropping packet from unknown transmitter"
+                              " %u.%u.%u.%u:%hu",
+                              ((a >> 24) & 0xFF),
+                              ((a >> 16) & 0xFF),
+                              ((a >>  8) & 0xFF),
+                              ( a        & 0xFF),
+                              p);
+
+                        drop_packet = true;
+                    } else {
+                        transmitter_addr_.sin_port = from.sin_port;
+                    }
+                } else {
+                    memcpy(&transmitter_addr_, &from, sizeof(from));
+                    transmitter_known_ = true;
+                }
+
+                if (!drop_packet) {
+                    bool serious_error = !processRX(pb);
+
+                    if (serious_error) {
+                        // Something went "seriously wrong".  Currently, the
+                        // only trigger for this should be a ring buffer
+                        // overflow.  The current failsafe behavior for when
+                        // something goes seriously wrong is to just reset the
+                        // pipeline.  The system should behave as if this
+                        // AAH_RXPlayer was just set up for the first time.
+                        ALOGE("Something just went seriously wrong with the"
+                              " pipeline.  Resetting.");
+                        resetPipeline();
+                    }
+                } else {
+                    PacketBuffer::destroy(pb);
+                }
+            }
+        }
+
+        // Step 3: Process any data we mave have accumulated in the ring buffer
+        // so far.
+        if (!thread_wrapper_->exitPending()) {
+            processRingBuffer();
+        }
+
+        // Step 4: At this point in time, the ring buffer should either be
+        // empty, or stalled in front of a gap caused by some dropped packets.
+        // Check on the current gap situation and deal with it in an appropriate
+        // fashion.  If processGaps returns true, it means that it has given up
+        // on a gap and that we should try to process some more data
+        // immediately.
+        if (!thread_wrapper_->exitPending()) {
+            process_more_right_now = processGaps();
+        }
+
+        // Step 5: Check for fatal errors.  If any of our substreams has
+        // encountered a fatal, unrecoverable, error, then propagate the error
+        // up to user level and shut down.
+        for (size_t i = 0; i < substreams_.size(); ++i) {
+            status_t status;
+            CHECK(substreams_.valueAt(i) != NULL);
+
+            status = substreams_.valueAt(i)->getStatus();
+            if (OK != status) {
+                ALOGE("Substream index %d has encountered an unrecoverable"
+                      " error (%d).  Signalling application level and shutting"
+                      " down.", i, status);
+                sendEvent(MEDIA_ERROR);
+                goto bailout;
+            }
+        }
+    }
+
+bailout:
+    cleanupSocket();
+    return false;
+}
+
+bool AAH_RXPlayer::processRX(PacketBuffer* pb) {
+    CHECK(NULL != pb);
+
+    uint8_t* data = pb->data_;
+    ssize_t  amt  = pb->length_;
+    uint32_t nak_magic;
+    uint16_t seq_no;
+    uint32_t epoch;
+
+    // Every packet either starts with an RTP header which is at least 12 bytes
+    // long or is a retry NAK which is 14 bytes long.  If there are fewer than
+    // 12 bytes here, this cannot be a proper RTP packet.
+    if (amt < 12) {
+        ALOGV("Dropping packet, too short to contain RTP header (%u bytes)",
+              static_cast<uint32_t>(amt));
+        goto drop_packet;
+    }
+
+    // Check to see if this is the special case of a NAK packet.
+    nak_magic = ntohl(*(reinterpret_cast<uint32_t*>(data)));
+    if (nak_magic == kRetransNAKMagic) {
+        // Looks like a NAK packet; make sure its long enough.
+
+        if (amt < static_cast<ssize_t>(sizeof(RetransRequest))) {
+            ALOGV("Dropping packet, too short to contain NAK payload (%u bytes)",
+                  static_cast<uint32_t>(amt));
+            goto drop_packet;
+        }
+
+        SeqNoGap gap;
+        RetransRequest* rtr = reinterpret_cast<RetransRequest*>(data);
+        gap.start_seq_ = ntohs(rtr->start_seq_);
+        gap.end_seq_   = ntohs(rtr->end_seq_);
+
+        ALOGV("Process NAK for gap at [%hu, %hu]", gap.start_seq_, gap.end_seq_);
+        ring_buffer_.processNAK(&gap);
+
+        return true;
+    }
+
+    // According to the TRTP spec, version should be 2, padding should be 0,
+    // extension should be 0 and CSRCCnt should be 0.  If any of these tests
+    // fail, we chuck the packet.
+    if (data[0] != 0x80) {
+        ALOGV("Dropping packet, bad V/P/X/CSRCCnt field (0x%02x)",
+              data[0]);
+        goto drop_packet;
+    }
+
+    // Check the payload type.  For TRTP, it should always be 100.
+    if ((data[1] & 0x7F) != 100) {
+        ALOGV("Dropping packet, bad payload type. (%u)",
+              data[1] & 0x7F);
+        goto drop_packet;
+    }
+
+    // Check whether the transmitter has begun a new epoch.
+    epoch = (U32_AT(data + 8) >> 10) & 0x3FFFFF;
+    if (current_epoch_known_) {
+        if (epoch != current_epoch_) {
+            ALOGV("%s: new epoch %u", __PRETTY_FUNCTION__, epoch);
+            current_epoch_ = epoch;
+            resetPipeline();
+        }
+    } else {
+        current_epoch_ = epoch;
+        current_epoch_known_ = true;
+    }
+
+    // Extract the sequence number and hand the packet off to the ring buffer
+    // for dropped packet detection and later processing.
+    seq_no = U16_AT(data + 2);
+    return ring_buffer_.pushBuffer(pb, seq_no);
+
+drop_packet:
+    PacketBuffer::destroy(pb);
+    return true;
+}
+
+void AAH_RXPlayer::processRingBuffer() {
+    PacketBuffer* pb;
+    bool is_discon;
+    sp<Substream> substream;
+    LinearTransform trans;
+    bool foundTrans = false;
+
+    while (NULL != (pb = ring_buffer_.fetchBuffer(&is_discon))) {
+        if (is_discon) {
+            // Abort all partially assembled payloads.
+            for (size_t i = 0; i < substreams_.size(); ++i) {
+                CHECK(substreams_.valueAt(i) != NULL);
+                substreams_.valueAt(i)->cleanupBufferInProgress();
+            }
+        }
+
+        uint8_t* data = pb->data_;
+        ssize_t  amt  = pb->length_;
+
+        // Should not have any non-RTP packets in the ring buffer.  RTP packets
+        // must be at least 12 bytes long.
+        CHECK(amt >= 12);
+
+        // Extract the marker bit and the SSRC field.
+        bool     marker = (data[1] & 0x80) != 0;
+        uint32_t ssrc   = U32_AT(data + 8);
+
+        // Is this the start of a new TRTP payload?  If so, the marker bit
+        // should be set and there are some things we should be checking for.
+        if (marker) {
+            // TRTP headers need to have at least a byte for version, a byte for
+            // payload type and flags, and 4 bytes for length.
+            if (amt < 18) {
+                ALOGV("Dropping packet, too short to contain TRTP header"
+                      " (%u bytes)", static_cast<uint32_t>(amt));
+                goto process_next_packet;
+            }
+
+            // Check the TRTP version and extract the payload type/flags.
+            uint8_t trtp_version =  data[12];
+            uint8_t payload_type = (data[13] >> 4) & 0xF;
+            uint8_t trtp_flags   =  data[13]       & 0xF;
+
+            if (1 != trtp_version) {
+                ALOGV("Dropping packet, bad trtp version %hhu", trtp_version);
+                goto process_next_packet;
+            }
+
+            // Is there a timestamp transformation present on this packet?  If
+            // so, extract it and pass it to the appropriate substreams.
+            if (trtp_flags & 0x02) {
+                ssize_t offset = 18 + ((trtp_flags & 0x01) ? 4 : 0);
+                if (amt < (offset + 24)) {
+                    ALOGV("Dropping packet, too short to contain TRTP Timestamp"
+                          " Transformation (%u bytes)",
+                          static_cast<uint32_t>(amt));
+                    goto process_next_packet;
+                }
+
+                trans.a_zero = fetchInt64(data + offset);
+                trans.b_zero = fetchInt64(data + offset + 16);
+                trans.a_to_b_numer = static_cast<int32_t>(
+                        fetchInt32 (data + offset + 8));
+                trans.a_to_b_denom = U32_AT(data + offset + 12);
+                foundTrans = true;
+
+                uint32_t program_id = (ssrc >> 5) & 0x1F;
+                for (size_t i = 0; i < substreams_.size(); ++i) {
+                    sp<Substream> iter = substreams_.valueAt(i);
+                    CHECK(iter != NULL);
+
+                    if (iter->getProgramID() == program_id) {
+                        iter->processTSTransform(trans);
+                    }
+                }
+            }
+
+            // Is this a command packet?  If so, its not necessarily associate
+            // with one particular substream.  Just give it to the command
+            // packet handler and then move on.
+            if (4 == payload_type) {
+                processCommandPacket(pb);
+                goto process_next_packet;
+            }
+        }
+
+        // If we got to here, then we are a normal packet.  Find (or allocate)
+        // the substream we belong to and send the packet off to be processed.
+        substream = substreams_.valueFor(ssrc);
+        if (substream == NULL) {
+            substream = new Substream(ssrc, omx_);
+            if (substream == NULL) {
+                ALOGE("Failed to allocate substream for SSRC 0x%08x", ssrc);
+                goto process_next_packet;
+            }
+            substreams_.add(ssrc, substream);
+
+            if (foundTrans) {
+                substream->processTSTransform(trans);
+            }
+        }
+
+        CHECK(substream != NULL);
+
+        if (marker) {
+            // Start of a new TRTP payload for this substream.  Extract the
+            // lower 32 bits of the timestamp and hand the buffer to the
+            // substream for processing.
+            uint32_t ts_lower = U32_AT(data + 4);
+            substream->processPayloadStart(data + 12, amt - 12, ts_lower);
+        } else {
+            // Continuation of an existing TRTP payload.  Just hand it off to
+            // the substream for processing.
+            substream->processPayloadCont(data + 12, amt - 12);
+        }
+
+process_next_packet:
+        PacketBuffer::destroy(pb);
+    }  // end of main processing while loop.
+}
+
+void AAH_RXPlayer::processCommandPacket(PacketBuffer* pb) {
+    CHECK(NULL != pb);
+
+    uint8_t* data = pb->data_;
+    ssize_t  amt  = pb->length_;
+
+    // verify that this packet meets the minimum length of a command packet
+    if (amt < 20) {
+        return;
+    }
+
+    uint8_t trtp_version =  data[12];
+    uint8_t trtp_flags   =  data[13]       & 0xF;
+
+    if (1 != trtp_version) {
+        ALOGV("Dropping packet, bad trtp version %hhu", trtp_version);
+        return;
+    }
+
+    // calculate the start of the command payload
+    ssize_t offset = 18;
+    if (trtp_flags & 0x01) {
+        // timestamp is present (4 bytes)
+        offset += 4;
+    }
+    if (trtp_flags & 0x02) {
+        // transform is present (24 bytes)
+        offset += 24;
+    }
+
+    // the packet must contain 2 bytes of command payload beyond the TRTP header
+    if (amt < offset + 2) {
+        return;
+    }
+
+    uint16_t command_id = U16_AT(data + offset);
+
+    switch (command_id) {
+        case TRTPControlPacket::kCommandNop:
+            break;
+
+        case TRTPControlPacket::kCommandEOS:
+        case TRTPControlPacket::kCommandFlush: {
+            uint16_t program_id = (U32_AT(data + 8) >> 5) & 0x1F;
+            ALOGI("*** %s flushing program_id=%d",
+                  __PRETTY_FUNCTION__, program_id);
+
+            Vector<uint32_t> substreams_to_remove;
+            for (size_t i = 0; i < substreams_.size(); ++i) {
+                sp<Substream> iter = substreams_.valueAt(i);
+                if (iter->getProgramID() == program_id) {
+                    iter->shutdown();
+                    substreams_to_remove.add(iter->getSSRC());
+                }
+            }
+
+            for (size_t i = 0; i < substreams_to_remove.size(); ++i) {
+                substreams_.removeItem(substreams_to_remove[i]);
+            }
+        } break;
+    }
+}
+
+bool AAH_RXPlayer::processGaps() {
+    // Deal with the current gap situation.  Specifically...
+    //
+    // 1) If a new gap has shown up, send a retransmit request to the
+    //    transmitter.
+    // 2) If a gap we were working on has had a packet in the middle or at
+    //    the end filled in, send another retransmit request for the begining
+    //    portion of the gap.  TRTP was designed for LANs where packet
+    //    re-ordering is very unlikely; so see the middle or end of a gap
+    //    filled in before the begining is an almost certain indication that
+    //    a retransmission packet was also dropped.
+    // 3) If we have been working on a gap for a while and it still has not
+    //    been filled in, send another retransmit request.
+    // 4) If the are no more gaps in the ring, clear the current_gap_status_
+    //    flag to indicate that all is well again.
+
+    // Start by fetching the active gap status.
+    SeqNoGap gap;
+    bool send_retransmit_request = false;
+    bool ret_val = false;
+    GapStatus gap_status;
+    if (kGS_NoGap != (gap_status = ring_buffer_.fetchCurrentGap(&gap))) {
+        // Note: checking for a change in the end sequence number should cover
+        // moving on to an entirely new gap for case #1 as well as resending the
+        // begining of a gap range for case #2.
+        send_retransmit_request = (kGS_NoGap == current_gap_status_) ||
+                                  (current_gap_.end_seq_ != gap.end_seq_);
+
+        // If this is the same gap we have been working on, and it has timed
+        // out, then check to see if our substreams are about to underflow.  If
+        // so, instead of sending another retransmit request, just give up on
+        // this gap and move on.
+        if (!send_retransmit_request &&
+           (kGS_NoGap != current_gap_status_) &&
+           (0 == computeNextGapRetransmitTimeout())) {
+
+            // If out current gap is the fast-start gap, don't bother to skip it
+            // because substreams look like the are about to underflow.
+            if ((kGS_FastStartGap != gap_status) ||
+                (current_gap_.end_seq_ != gap.end_seq_)) {
+                for (size_t i = 0; i < substreams_.size(); ++i) {
+                    if (substreams_.valueAt(i)->isAboutToUnderflow()) {
+                        ALOGV("About to underflow, giving up on gap [%hu, %hu]",
+                              gap.start_seq_, gap.end_seq_);
+                        ring_buffer_.processNAK();
+                        current_gap_status_ = kGS_NoGap;
+                        return true;
+                    }
+                }
+            }
+
+            // Looks like no one is about to underflow.  Just go ahead and send
+            // the request.
+            send_retransmit_request = true;
+        }
+    } else {
+        current_gap_status_ = kGS_NoGap;
+    }
+
+    if (send_retransmit_request) {
+        // If we have been working on a fast start, and it is still not filled
+        // in, even after the extended retransmit time out, give up and skip it.
+        // The system should fall back into its normal slow-start behavior.
+        if ((kGS_FastStartGap == current_gap_status_) &&
+            (current_gap_.end_seq_ == gap.end_seq_)) {
+            ALOGV("Fast start is taking forever; giving up.");
+            ring_buffer_.processNAK();
+            current_gap_status_ = kGS_NoGap;
+            return true;
+        }
+
+        // Send the request.
+        RetransRequest req;
+        uint32_t magic  = (kGS_FastStartGap == gap_status)
+                        ? kFastStartRequestMagic
+                        : kRetransRequestMagic;
+        req.magic_      = htonl(magic);
+        req.mcast_ip_   = listen_addr_.sin_addr.s_addr;
+        req.mcast_port_ = listen_addr_.sin_port;
+        req.start_seq_  = htons(gap.start_seq_);
+        req.end_seq_    = htons(gap.end_seq_);
+
+        {
+            uint32_t a = ntohl(transmitter_addr_.sin_addr.s_addr);
+            uint16_t p = ntohs(transmitter_addr_.sin_port);
+            ALOGV("Sending to transmitter %u.%u.%u.%u:%hu",
+                    ((a >> 24) & 0xFF),
+                    ((a >> 16) & 0xFF),
+                    ((a >>  8) & 0xFF),
+                    ( a        & 0xFF),
+                    p);
+        }
+
+        int res = sendto(sock_fd_, &req, sizeof(req), 0,
+                         reinterpret_cast<struct sockaddr*>(&transmitter_addr_),
+                         sizeof(transmitter_addr_));
+        if (res < 0) {
+            ALOGE("Error when sending retransmit request (%d)", errno);
+        } else {
+            ALOGV("%s request for range [%hu, %hu] sent",
+                  (kGS_FastStartGap == gap_status) ? "Fast Start" : "Retransmit",
+                  gap.start_seq_, gap.end_seq_);
+        }
+
+        // Update the current gap info.
+        current_gap_ = gap;
+        current_gap_status_ = gap_status;
+        next_retrans_req_time_ = monotonicUSecNow() +
+                               ((kGS_FastStartGap == current_gap_status_)
+                                ? kFastStartTimeoutUSec
+                                : kGapRerequestTimeoutUSec);
+    }
+
+    return false;
+}
+
+// Compute when its time to send the next gap retransmission in milliseconds.
+// Returns < 0 for an infinite timeout (no gap) and 0 if its time to retransmit
+// right now.
+int AAH_RXPlayer::computeNextGapRetransmitTimeout() {
+    if (kGS_NoGap == current_gap_status_) {
+        return -1;
+    }
+
+    int64_t timeout_delta = next_retrans_req_time_ - monotonicUSecNow();
+
+    timeout_delta /= 1000;
+    if (timeout_delta <= 0) {
+        return 0;
+    }
+
+    return static_cast<uint32_t>(timeout_delta);
+}
+
+}  // namespace android