Support for acting as a wifi display sink.

Change-Id: I0beac87025b93c60164daa865c89f16b72197a47

11 files changed, 2344 insertions, 12 deletions

diff --git a/media/libmediaplayerservice/nuplayer/NuPlayer.cpp b/media/libmediaplayerservice/nuplayer/NuPlayer.cpp
index a02732b..91aaafe 100644
--- a/media/libmediaplayerservice/nuplayer/NuPlayer.cpp
+++ b/media/libmediaplayerservice/nuplayer/NuPlayer.cpp
@@ -293,8 +293,8 @@ void NuPlayer::onMessageReceived(const sp<AMessage> &msg) {
                 break;
             }
 
-            if (mAudioDecoder == NULL && mAudioSink != NULL ||
-                mVideoDecoder == NULL && mNativeWindow != NULL) {
+            if ((mAudioDecoder == NULL && mAudioSink != NULL)
+                    || (mVideoDecoder == NULL && mNativeWindow != NULL)) {
                 msg->post(100000ll);
                 mScanSourcesPending = true;
             }
diff --git a/media/libstagefright/wifi-display/Android.mk b/media/libstagefright/wifi-display/Android.mk
index b035a51..0e59b9e 100644
--- a/media/libstagefright/wifi-display/Android.mk
+++ b/media/libstagefright/wifi-display/Android.mk
@@ -5,6 +5,10 @@ include $(CLEAR_VARS)
 LOCAL_SRC_FILES:= \
         ANetworkSession.cpp             \
         ParsedMessage.cpp               \
+        sink/LinearRegression.cpp       \
+        sink/RTPSink.cpp                \
+        sink/TunnelRenderer.cpp         \
+        sink/WifiDisplaySink.cpp        \
         source/Converter.cpp            \
         source/PlaybackSession.cpp      \
         source/RepeaterSource.cpp       \
diff --git a/media/libstagefright/wifi-display/sink/LinearRegression.cpp b/media/libstagefright/wifi-display/sink/LinearRegression.cpp
new file mode 100644
index 0000000..8cfce37
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/LinearRegression.cpp
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2012, 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_NDEBUG 0
+#define LOG_TAG "LinearRegression"
+#include <utils/Log.h>
+
+#include "LinearRegression.h"
+
+#include <math.h>
+#include <string.h>
+
+namespace android {
+
+LinearRegression::LinearRegression(size_t historySize)
+    : mHistorySize(historySize),
+      mCount(0),
+      mHistory(new Point[mHistorySize]),
+      mSumX(0.0),
+      mSumY(0.0) {
+}
+
+LinearRegression::~LinearRegression() {
+    delete[] mHistory;
+    mHistory = NULL;
+}
+
+void LinearRegression::addPoint(float x, float y) {
+    if (mCount == mHistorySize) {
+        const Point &oldest = mHistory[0];
+
+        mSumX -= oldest.mX;
+        mSumY -= oldest.mY;
+
+        memmove(&mHistory[0], &mHistory[1], (mHistorySize - 1) * sizeof(Point));
+        --mCount;
+    }
+
+    Point *newest = &mHistory[mCount++];
+    newest->mX = x;
+    newest->mY = y;
+
+    mSumX += x;
+    mSumY += y;
+}
+
+bool LinearRegression::approxLine(float *n1, float *n2, float *b) const {
+    static const float kEpsilon = 1.0E-4;
+
+    if (mCount < 2) {
+        return false;
+    }
+
+    float sumX2 = 0.0f;
+    float sumY2 = 0.0f;
+    float sumXY = 0.0f;
+
+    float meanX = mSumX / (float)mCount;
+    float meanY = mSumY / (float)mCount;
+
+    for (size_t i = 0; i < mCount; ++i) {
+        const Point &p = mHistory[i];
+
+        float x = p.mX - meanX;
+        float y = p.mY - meanY;
+
+        sumX2 += x * x;
+        sumY2 += y * y;
+        sumXY += x * y;
+    }
+
+    float T = sumX2 + sumY2;
+    float D = sumX2 * sumY2 - sumXY * sumXY;
+    float root = sqrt(T * T * 0.25 - D);
+
+    float L1 = T * 0.5 - root;
+
+    if (fabs(sumXY) > kEpsilon) {
+        *n1 = 1.0;
+        *n2 = (2.0 * L1 - sumX2) / sumXY;
+
+        float mag = sqrt((*n1) * (*n1) + (*n2) * (*n2));
+
+        *n1 /= mag;
+        *n2 /= mag;
+    } else {
+        *n1 = 0.0;
+        *n2 = 1.0;
+    }
+
+    *b = (*n1) * meanX + (*n2) * meanY;
+
+    return true;
+}
+
+}  // namespace android
+
diff --git a/media/libstagefright/wifi-display/sink/LinearRegression.h b/media/libstagefright/wifi-display/sink/LinearRegression.h
new file mode 100644
index 0000000..ca6f5a1
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/LinearRegression.h
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2012, 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.
+ */
+
+#ifndef LINEAR_REGRESSION_H_
+
+#define LINEAR_REGRESSION_H_
+
+#include <sys/types.h>
+#include <media/stagefright/foundation/ABase.h>
+
+namespace android {
+
+// Helper class to fit a line to a set of points minimizing the sum of
+// squared (orthogonal) distances from line to individual points.
+struct LinearRegression {
+    LinearRegression(size_t historySize);
+    ~LinearRegression();
+
+    void addPoint(float x, float y);
+
+    bool approxLine(float *n1, float *n2, float *b) const;
+
+private:
+    struct Point {
+        float mX, mY;
+    };
+
+    size_t mHistorySize;
+    size_t mCount;
+    Point *mHistory;
+
+    float mSumX, mSumY;
+
+    DISALLOW_EVIL_CONSTRUCTORS(LinearRegression);
+};
+
+}  // namespace android
+
+#endif  // LINEAR_REGRESSION_H_
diff --git a/media/libstagefright/wifi-display/sink/RTPSink.cpp b/media/libstagefright/wifi-display/sink/RTPSink.cpp
new file mode 100644
index 0000000..0918034
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/RTPSink.cpp
@@ -0,0 +1,806 @@
+/*
+ * Copyright 2012, 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_NDEBUG 0
+#define LOG_TAG "RTPSink"
+#include <utils/Log.h>
+
+#include "RTPSink.h"
+
+#include "ANetworkSession.h"
+#include "TunnelRenderer.h"
+
+#include <media/stagefright/foundation/ABuffer.h>
+#include <media/stagefright/foundation/ADebug.h>
+#include <media/stagefright/foundation/AMessage.h>
+#include <media/stagefright/foundation/hexdump.h>
+#include <media/stagefright/MediaErrors.h>
+#include <media/stagefright/Utils.h>
+
+namespace android {
+
+struct RTPSink::Source : public RefBase {
+    Source(uint16_t seq, const sp<ABuffer> &buffer,
+           const sp<AMessage> queueBufferMsg);
+
+    bool updateSeq(uint16_t seq, const sp<ABuffer> &buffer);
+
+    void addReportBlock(uint32_t ssrc, const sp<ABuffer> &buf);
+
+protected:
+    virtual ~Source();
+
+private:
+    static const uint32_t kMinSequential = 2;
+    static const uint32_t kMaxDropout = 3000;
+    static const uint32_t kMaxMisorder = 100;
+    static const uint32_t kRTPSeqMod = 1u << 16;
+
+    sp<AMessage> mQueueBufferMsg;
+
+    uint16_t mMaxSeq;
+    uint32_t mCycles;
+    uint32_t mBaseSeq;
+    uint32_t mBadSeq;
+    uint32_t mProbation;
+    uint32_t mReceived;
+    uint32_t mExpectedPrior;
+    uint32_t mReceivedPrior;
+
+    void initSeq(uint16_t seq);
+    void queuePacket(const sp<ABuffer> &buffer);
+
+    DISALLOW_EVIL_CONSTRUCTORS(Source);
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+RTPSink::Source::Source(
+        uint16_t seq, const sp<ABuffer> &buffer,
+        const sp<AMessage> queueBufferMsg)
+    : mQueueBufferMsg(queueBufferMsg),
+      mProbation(kMinSequential) {
+    initSeq(seq);
+    mMaxSeq = seq - 1;
+
+    buffer->setInt32Data(mCycles | seq);
+    queuePacket(buffer);
+}
+
+RTPSink::Source::~Source() {
+}
+
+void RTPSink::Source::initSeq(uint16_t seq) {
+    mMaxSeq = seq;
+    mCycles = 0;
+    mBaseSeq = seq;
+    mBadSeq = kRTPSeqMod + 1;
+    mReceived = 0;
+    mExpectedPrior = 0;
+    mReceivedPrior = 0;
+}
+
+bool RTPSink::Source::updateSeq(uint16_t seq, const sp<ABuffer> &buffer) {
+    uint16_t udelta = seq - mMaxSeq;
+
+    if (mProbation) {
+        // Startup phase
+
+        if (seq == mMaxSeq + 1) {
+            buffer->setInt32Data(mCycles | seq);
+            queuePacket(buffer);
+
+            --mProbation;
+            mMaxSeq = seq;
+            if (mProbation == 0) {
+                initSeq(seq);
+                ++mReceived;
+
+                return true;
+            }
+        } else {
+            // Packet out of sequence, restart startup phase
+
+            mProbation = kMinSequential - 1;
+            mMaxSeq = seq;
+
+#if 0
+            mPackets.clear();
+            mTotalBytesQueued = 0;
+            ALOGI("XXX cleared packets");
+#endif
+
+            buffer->setInt32Data(mCycles | seq);
+            queuePacket(buffer);
+        }
+
+        return false;
+    }
+
+    if (udelta < kMaxDropout) {
+        // In order, with permissible gap.
+
+        if (seq < mMaxSeq) {
+            // Sequence number wrapped - count another 64K cycle
+            mCycles += kRTPSeqMod;
+        }
+
+        mMaxSeq = seq;
+    } else if (udelta <= kRTPSeqMod - kMaxMisorder) {
+        // The sequence number made a very large jump
+
+        if (seq == mBadSeq) {
+            // Two sequential packets -- assume that the other side
+            // restarted without telling us so just re-sync
+            // (i.e. pretend this was the first packet)
+
+            initSeq(seq);
+        } else {
+            mBadSeq = (seq + 1) & (kRTPSeqMod - 1);
+
+            return false;
+        }
+    } else {
+        // Duplicate or reordered packet.
+    }
+
+    ++mReceived;
+
+    buffer->setInt32Data(mCycles | seq);
+    queuePacket(buffer);
+
+    return true;
+}
+
+void RTPSink::Source::queuePacket(const sp<ABuffer> &buffer) {
+    sp<AMessage> msg = mQueueBufferMsg->dup();
+    msg->setBuffer("buffer", buffer);
+    msg->post();
+}
+
+void RTPSink::Source::addReportBlock(
+        uint32_t ssrc, const sp<ABuffer> &buf) {
+    uint32_t extMaxSeq = mMaxSeq | mCycles;
+    uint32_t expected = extMaxSeq - mBaseSeq + 1;
+
+    int64_t lost = (int64_t)expected - (int64_t)mReceived;
+    if (lost > 0x7fffff) {
+        lost = 0x7fffff;
+    } else if (lost < -0x800000) {
+        lost = -0x800000;
+    }
+
+    uint32_t expectedInterval = expected - mExpectedPrior;
+    mExpectedPrior = expected;
+
+    uint32_t receivedInterval = mReceived - mReceivedPrior;
+    mReceivedPrior = mReceived;
+
+    int64_t lostInterval = expectedInterval - receivedInterval;
+
+    uint8_t fractionLost;
+    if (expectedInterval == 0 || lostInterval <=0) {
+        fractionLost = 0;
+    } else {
+        fractionLost = (lostInterval << 8) / expectedInterval;
+    }
+
+    uint8_t *ptr = buf->data() + buf->size();
+
+    ptr[0] = ssrc >> 24;
+    ptr[1] = (ssrc >> 16) & 0xff;
+    ptr[2] = (ssrc >> 8) & 0xff;
+    ptr[3] = ssrc & 0xff;
+
+    ptr[4] = fractionLost;
+
+    ptr[5] = (lost >> 16) & 0xff;
+    ptr[6] = (lost >> 8) & 0xff;
+    ptr[7] = lost & 0xff;
+
+    ptr[8] = extMaxSeq >> 24;
+    ptr[9] = (extMaxSeq >> 16) & 0xff;
+    ptr[10] = (extMaxSeq >> 8) & 0xff;
+    ptr[11] = extMaxSeq & 0xff;
+
+    // XXX TODO:
+
+    ptr[12] = 0x00;  // interarrival jitter
+    ptr[13] = 0x00;
+    ptr[14] = 0x00;
+    ptr[15] = 0x00;
+
+    ptr[16] = 0x00;  // last SR
+    ptr[17] = 0x00;
+    ptr[18] = 0x00;
+    ptr[19] = 0x00;
+
+    ptr[20] = 0x00;  // delay since last SR
+    ptr[21] = 0x00;
+    ptr[22] = 0x00;
+    ptr[23] = 0x00;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+RTPSink::RTPSink(
+        const sp<ANetworkSession> &netSession,
+        const sp<ISurfaceTexture> &surfaceTex)
+    : mNetSession(netSession),
+      mSurfaceTex(surfaceTex),
+      mRTPPort(0),
+      mRTPSessionID(0),
+      mRTCPSessionID(0),
+      mFirstArrivalTimeUs(-1ll),
+      mNumPacketsReceived(0ll),
+      mRegression(1000),
+      mMaxDelayMs(-1ll) {
+}
+
+RTPSink::~RTPSink() {
+    if (mRTCPSessionID != 0) {
+        mNetSession->destroySession(mRTCPSessionID);
+    }
+
+    if (mRTPSessionID != 0) {
+        mNetSession->destroySession(mRTPSessionID);
+    }
+}
+
+status_t RTPSink::init(bool useTCPInterleaving) {
+    if (useTCPInterleaving) {
+        return OK;
+    }
+
+    int clientRtp;
+
+    sp<AMessage> rtpNotify = new AMessage(kWhatRTPNotify, id());
+    sp<AMessage> rtcpNotify = new AMessage(kWhatRTCPNotify, id());
+    for (clientRtp = 15550;; clientRtp += 2) {
+        int32_t rtpSession;
+        status_t err = mNetSession->createUDPSession(
+                    clientRtp, rtpNotify, &rtpSession);
+
+        if (err != OK) {
+            ALOGI("failed to create RTP socket on port %d", clientRtp);
+            continue;
+        }
+
+        int32_t rtcpSession;
+        err = mNetSession->createUDPSession(
+                clientRtp + 1, rtcpNotify, &rtcpSession);
+
+        if (err == OK) {
+            mRTPPort = clientRtp;
+            mRTPSessionID = rtpSession;
+            mRTCPSessionID = rtcpSession;
+            break;
+        }
+
+        ALOGI("failed to create RTCP socket on port %d", clientRtp + 1);
+        mNetSession->destroySession(rtpSession);
+    }
+
+    if (mRTPPort == 0) {
+        return UNKNOWN_ERROR;
+    }
+
+    return OK;
+}
+
+int32_t RTPSink::getRTPPort() const {
+    return mRTPPort;
+}
+
+void RTPSink::onMessageReceived(const sp<AMessage> &msg) {
+    switch (msg->what()) {
+        case kWhatRTPNotify:
+        case kWhatRTCPNotify:
+        {
+            int32_t reason;
+            CHECK(msg->findInt32("reason", &reason));
+
+            switch (reason) {
+                case ANetworkSession::kWhatError:
+                {
+                    int32_t sessionID;
+                    CHECK(msg->findInt32("sessionID", &sessionID));
+
+                    int32_t err;
+                    CHECK(msg->findInt32("err", &err));
+
+                    AString detail;
+                    CHECK(msg->findString("detail", &detail));
+
+                    ALOGE("An error occurred in session %d (%d, '%s/%s').",
+                          sessionID,
+                          err,
+                          detail.c_str(),
+                          strerror(-err));
+
+                    mNetSession->destroySession(sessionID);
+
+                    if (sessionID == mRTPSessionID) {
+                        mRTPSessionID = 0;
+                    } else if (sessionID == mRTCPSessionID) {
+                        mRTCPSessionID = 0;
+                    }
+                    break;
+                }
+
+                case ANetworkSession::kWhatDatagram:
+                {
+                    int32_t sessionID;
+                    CHECK(msg->findInt32("sessionID", &sessionID));
+
+                    sp<ABuffer> data;
+                    CHECK(msg->findBuffer("data", &data));
+
+                    status_t err;
+                    if (msg->what() == kWhatRTPNotify) {
+                        err = parseRTP(data);
+                    } else {
+                        err = parseRTCP(data);
+                    }
+                    break;
+                }
+
+                default:
+                    TRESPASS();
+            }
+            break;
+        }
+
+        case kWhatSendRR:
+        {
+            onSendRR();
+            break;
+        }
+
+        case kWhatPacketLost:
+        {
+            onPacketLost(msg);
+            break;
+        }
+
+        case kWhatInject:
+        {
+            int32_t isRTP;
+            CHECK(msg->findInt32("isRTP", &isRTP));
+
+            sp<ABuffer> buffer;
+            CHECK(msg->findBuffer("buffer", &buffer));
+
+            status_t err;
+            if (isRTP) {
+                err = parseRTP(buffer);
+            } else {
+                err = parseRTCP(buffer);
+            }
+            break;
+        }
+
+        default:
+            TRESPASS();
+    }
+}
+
+status_t RTPSink::injectPacket(bool isRTP, const sp<ABuffer> &buffer) {
+    sp<AMessage> msg = new AMessage(kWhatInject, id());
+    msg->setInt32("isRTP", isRTP);
+    msg->setBuffer("buffer", buffer);
+    msg->post();
+
+    return OK;
+}
+
+status_t RTPSink::parseRTP(const sp<ABuffer> &buffer) {
+    size_t size = buffer->size();
+    if (size < 12) {
+        // Too short to be a valid RTP header.
+        return ERROR_MALFORMED;
+    }
+
+    const uint8_t *data = buffer->data();
+
+    if ((data[0] >> 6) != 2) {
+        // Unsupported version.
+        return ERROR_UNSUPPORTED;
+    }
+
+    if (data[0] & 0x20) {
+        // Padding present.
+
+        size_t paddingLength = data[size - 1];
+
+        if (paddingLength + 12 > size) {
+            // If we removed this much padding we'd end up with something
+            // that's too short to be a valid RTP header.
+            return ERROR_MALFORMED;
+        }
+
+        size -= paddingLength;
+    }
+
+    int numCSRCs = data[0] & 0x0f;
+
+    size_t payloadOffset = 12 + 4 * numCSRCs;
+
+    if (size < payloadOffset) {
+        // Not enough data to fit the basic header and all the CSRC entries.
+        return ERROR_MALFORMED;
+    }
+
+    if (data[0] & 0x10) {
+        // Header eXtension present.
+
+        if (size < payloadOffset + 4) {
+            // Not enough data to fit the basic header, all CSRC entries
+            // and the first 4 bytes of the extension header.
+
+            return ERROR_MALFORMED;
+        }
+
+        const uint8_t *extensionData = &data[payloadOffset];
+
+        size_t extensionLength =
+            4 * (extensionData[2] << 8 | extensionData[3]);
+
+        if (size < payloadOffset + 4 + extensionLength) {
+            return ERROR_MALFORMED;
+        }
+
+        payloadOffset += 4 + extensionLength;
+    }
+
+    uint32_t srcId = U32_AT(&data[8]);
+    uint32_t rtpTime = U32_AT(&data[4]);
+    uint16_t seqNo = U16_AT(&data[2]);
+
+    int64_t arrivalTimeUs;
+    CHECK(buffer->meta()->findInt64("arrivalTimeUs", &arrivalTimeUs));
+
+    if (mFirstArrivalTimeUs < 0ll) {
+        mFirstArrivalTimeUs = arrivalTimeUs;
+    }
+    arrivalTimeUs -= mFirstArrivalTimeUs;
+
+    int64_t arrivalTimeMedia = (arrivalTimeUs * 9ll) / 100ll;
+
+    ALOGV("seqNo: %d, SSRC 0x%08x, diff %lld",
+            seqNo, srcId, rtpTime - arrivalTimeMedia);
+
+    mRegression.addPoint((float)rtpTime, (float)arrivalTimeMedia);
+
+    ++mNumPacketsReceived;
+
+    float n1, n2, b;
+    if (mRegression.approxLine(&n1, &n2, &b)) {
+        ALOGV("Line %lld: %.2f %.2f %.2f, slope %.2f",
+              mNumPacketsReceived, n1, n2, b, -n1 / n2);
+
+        float expectedArrivalTimeMedia = (b - n1 * (float)rtpTime) / n2;
+        float latenessMs = (arrivalTimeMedia - expectedArrivalTimeMedia) / 90.0;
+
+        if (mMaxDelayMs < 0ll || latenessMs > mMaxDelayMs) {
+            mMaxDelayMs = latenessMs;
+            ALOGI("packet was %.2f ms late", latenessMs);
+        }
+    }
+
+    sp<AMessage> meta = buffer->meta();
+    meta->setInt32("ssrc", srcId);
+    meta->setInt32("rtp-time", rtpTime);
+    meta->setInt32("PT", data[1] & 0x7f);
+    meta->setInt32("M", data[1] >> 7);
+
+    buffer->setRange(payloadOffset, size - payloadOffset);
+
+    ssize_t index = mSources.indexOfKey(srcId);
+    if (index < 0) {
+        if (mRenderer == NULL) {
+            sp<AMessage> notifyLost = new AMessage(kWhatPacketLost, id());
+            notifyLost->setInt32("ssrc", srcId);
+
+            mRenderer = new TunnelRenderer(notifyLost, mSurfaceTex);
+            looper()->registerHandler(mRenderer);
+        }
+
+        sp<AMessage> queueBufferMsg =
+            new AMessage(TunnelRenderer::kWhatQueueBuffer, mRenderer->id());
+
+        sp<Source> source = new Source(seqNo, buffer, queueBufferMsg);
+        mSources.add(srcId, source);
+    } else {
+        mSources.valueAt(index)->updateSeq(seqNo, buffer);
+    }
+
+    return OK;
+}
+
+status_t RTPSink::parseRTCP(const sp<ABuffer> &buffer) {
+    const uint8_t *data = buffer->data();
+    size_t size = buffer->size();
+
+    while (size > 0) {
+        if (size < 8) {
+            // Too short to be a valid RTCP header
+            return ERROR_MALFORMED;
+        }
+
+        if ((data[0] >> 6) != 2) {
+            // Unsupported version.
+            return ERROR_UNSUPPORTED;
+        }
+
+        if (data[0] & 0x20) {
+            // Padding present.
+
+            size_t paddingLength = data[size - 1];
+
+            if (paddingLength + 12 > size) {
+                // If we removed this much padding we'd end up with something
+                // that's too short to be a valid RTP header.
+                return ERROR_MALFORMED;
+            }
+
+            size -= paddingLength;
+        }
+
+        size_t headerLength = 4 * (data[2] << 8 | data[3]) + 4;
+
+        if (size < headerLength) {
+            // Only received a partial packet?
+            return ERROR_MALFORMED;
+        }
+
+        switch (data[1]) {
+            case 200:
+            {
+                parseSR(data, headerLength);
+                break;
+            }
+
+            case 201:  // RR
+            case 202:  // SDES
+            case 204:  // APP
+                break;
+
+            case 205:  // TSFB (transport layer specific feedback)
+            case 206:  // PSFB (payload specific feedback)
+                // hexdump(data, headerLength);
+                break;
+
+            case 203:
+            {
+                parseBYE(data, headerLength);
+                break;
+            }
+
+            default:
+            {
+                ALOGW("Unknown RTCP packet type %u of size %d",
+                     (unsigned)data[1], headerLength);
+                break;
+            }
+        }
+
+        data += headerLength;
+        size -= headerLength;
+    }
+
+    return OK;
+}
+
+status_t RTPSink::parseBYE(const uint8_t *data, size_t size) {
+    size_t SC = data[0] & 0x3f;
+
+    if (SC == 0 || size < (4 + SC * 4)) {
+        // Packet too short for the minimal BYE header.
+        return ERROR_MALFORMED;
+    }
+
+    uint32_t id = U32_AT(&data[4]);
+
+    return OK;
+}
+
+status_t RTPSink::parseSR(const uint8_t *data, size_t size) {
+    size_t RC = data[0] & 0x1f;
+
+    if (size < (7 + RC * 6) * 4) {
+        // Packet too short for the minimal SR header.
+        return ERROR_MALFORMED;
+    }
+
+    uint32_t id = U32_AT(&data[4]);
+    uint64_t ntpTime = U64_AT(&data[8]);
+    uint32_t rtpTime = U32_AT(&data[16]);
+
+    ALOGV("SR: ssrc 0x%08x, ntpTime 0x%016llx, rtpTime 0x%08x",
+          id, ntpTime, rtpTime);
+
+    return OK;
+}
+
+status_t RTPSink::connect(
+        const char *host, int32_t remoteRtpPort, int32_t remoteRtcpPort) {
+    ALOGI("connecting RTP/RTCP sockets to %s:{%d,%d}",
+          host, remoteRtpPort, remoteRtcpPort);
+
+    status_t err =
+        mNetSession->connectUDPSession(mRTPSessionID, host, remoteRtpPort);
+
+    if (err != OK) {
+        return err;
+    }
+
+    err = mNetSession->connectUDPSession(mRTCPSessionID, host, remoteRtcpPort);
+
+    if (err != OK) {
+        return err;
+    }
+
+#if 0
+    sp<ABuffer> buf = new ABuffer(1500);
+    memset(buf->data(), 0, buf->size());
+
+    mNetSession->sendRequest(
+            mRTPSessionID, buf->data(), buf->size());
+
+    mNetSession->sendRequest(
+            mRTCPSessionID, buf->data(), buf->size());
+#endif
+
+    scheduleSendRR();
+
+    return OK;
+}
+
+void RTPSink::scheduleSendRR() {
+    (new AMessage(kWhatSendRR, id()))->post(2000000ll);
+}
+
+void RTPSink::addSDES(const sp<ABuffer> &buffer) {
+    uint8_t *data = buffer->data() + buffer->size();
+    data[0] = 0x80 | 1;
+    data[1] = 202;  // SDES
+    data[4] = 0xde;  // SSRC
+    data[5] = 0xad;
+    data[6] = 0xbe;
+    data[7] = 0xef;
+
+    size_t offset = 8;
+
+    data[offset++] = 1;  // CNAME
+
+    AString cname = "stagefright@somewhere";
+    data[offset++] = cname.size();
+
+    memcpy(&data[offset], cname.c_str(), cname.size());
+    offset += cname.size();
+
+    data[offset++] = 6;  // TOOL
+
+    AString tool = "stagefright/1.0";
+    data[offset++] = tool.size();
+
+    memcpy(&data[offset], tool.c_str(), tool.size());
+    offset += tool.size();
+
+    data[offset++] = 0;
+
+    if ((offset % 4) > 0) {
+        size_t count = 4 - (offset % 4);
+        switch (count) {
+            case 3:
+                data[offset++] = 0;
+            case 2:
+                data[offset++] = 0;
+            case 1:
+                data[offset++] = 0;
+        }
+    }
+
+    size_t numWords = (offset / 4) - 1;
+    data[2] = numWords >> 8;
+    data[3] = numWords & 0xff;
+
+    buffer->setRange(buffer->offset(), buffer->size() + offset);
+}
+
+void RTPSink::onSendRR() {
+    sp<ABuffer> buf = new ABuffer(1500);
+    buf->setRange(0, 0);
+
+    uint8_t *ptr = buf->data();
+    ptr[0] = 0x80 | 0;
+    ptr[1] = 201;  // RR
+    ptr[2] = 0;
+    ptr[3] = 1;
+    ptr[4] = 0xde;  // SSRC
+    ptr[5] = 0xad;
+    ptr[6] = 0xbe;
+    ptr[7] = 0xef;
+
+    buf->setRange(0, 8);
+
+    size_t numReportBlocks = 0;
+    for (size_t i = 0; i < mSources.size(); ++i) {
+        uint32_t ssrc = mSources.keyAt(i);
+        sp<Source> source = mSources.valueAt(i);
+
+        if (numReportBlocks > 31 || buf->size() + 24 > buf->capacity()) {
+            // Cannot fit another report block.
+            break;
+        }
+
+        source->addReportBlock(ssrc, buf);
+        ++numReportBlocks;
+    }
+
+    ptr[0] |= numReportBlocks;  // 5 bit
+
+    size_t sizeInWordsMinus1 = 1 + 6 * numReportBlocks;
+    ptr[2] = sizeInWordsMinus1 >> 8;
+    ptr[3] = sizeInWordsMinus1 & 0xff;
+
+    buf->setRange(0, (sizeInWordsMinus1 + 1) * 4);
+
+    addSDES(buf);
+
+    mNetSession->sendRequest(mRTCPSessionID, buf->data(), buf->size());
+
+    scheduleSendRR();
+}
+
+void RTPSink::onPacketLost(const sp<AMessage> &msg) {
+    uint32_t srcId;
+    CHECK(msg->findInt32("ssrc", (int32_t *)&srcId));
+
+    int32_t seqNo;
+    CHECK(msg->findInt32("seqNo", &seqNo));
+
+    int32_t blp = 0;
+
+    sp<ABuffer> buf = new ABuffer(1500);
+    buf->setRange(0, 0);
+
+    uint8_t *ptr = buf->data();
+    ptr[0] = 0x80 | 1;  // generic NACK
+    ptr[1] = 205;  // RTPFB
+    ptr[2] = 0;
+    ptr[3] = 3;
+    ptr[4] = 0xde;  // sender SSRC
+    ptr[5] = 0xad;
+    ptr[6] = 0xbe;
+    ptr[7] = 0xef;
+    ptr[8] = (srcId >> 24) & 0xff;
+    ptr[9] = (srcId >> 16) & 0xff;
+    ptr[10] = (srcId >> 8) & 0xff;
+    ptr[11] = (srcId & 0xff);
+    ptr[12] = (seqNo >> 8) & 0xff;
+    ptr[13] = (seqNo & 0xff);
+    ptr[14] = (blp >> 8) & 0xff;
+    ptr[15] = (blp & 0xff);
+
+    buf->setRange(0, 16);
+
+    mNetSession->sendRequest(mRTCPSessionID, buf->data(), buf->size());
+}
+
+}  // namespace android
+
diff --git a/media/libstagefright/wifi-display/sink/RTPSink.h b/media/libstagefright/wifi-display/sink/RTPSink.h
new file mode 100644
index 0000000..a1d127d
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/RTPSink.h
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2012, 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.
+ */
+
+#ifndef RTP_SINK_H_
+
+#define RTP_SINK_H_
+
+#include <media/stagefright/foundation/AHandler.h>
+
+#include "LinearRegression.h"
+
+#include <gui/Surface.h>
+
+namespace android {
+
+struct ABuffer;
+struct ANetworkSession;
+struct TunnelRenderer;
+
+// Creates a pair of sockets for RTP/RTCP traffic, instantiates a renderer
+// for incoming transport stream data and occasionally sends statistics over
+// the RTCP channel.
+struct RTPSink : public AHandler {
+    RTPSink(const sp<ANetworkSession> &netSession,
+            const sp<ISurfaceTexture> &surfaceTex);
+
+    // If TCP interleaving is used, no UDP sockets are created, instead
+    // incoming RTP/RTCP packets (arriving on the RTSP control connection)
+    // are manually injected by WifiDisplaySink.
+    status_t init(bool useTCPInterleaving);
+
+    status_t connect(
+            const char *host, int32_t remoteRtpPort, int32_t remoteRtcpPort);
+
+    int32_t getRTPPort() const;
+
+    status_t injectPacket(bool isRTP, const sp<ABuffer> &buffer);
+
+protected:
+    virtual void onMessageReceived(const sp<AMessage> &msg);
+    virtual ~RTPSink();
+
+private:
+    enum {
+        kWhatRTPNotify,
+        kWhatRTCPNotify,
+        kWhatSendRR,
+        kWhatPacketLost,
+        kWhatInject,
+    };
+
+    struct Source;
+    struct StreamSource;
+
+    sp<ANetworkSession> mNetSession;
+    sp<ISurfaceTexture> mSurfaceTex;
+    KeyedVector<uint32_t, sp<Source> > mSources;
+
+    int32_t mRTPPort;
+    int32_t mRTPSessionID;
+    int32_t mRTCPSessionID;
+
+    int64_t mFirstArrivalTimeUs;
+    int64_t mNumPacketsReceived;
+    LinearRegression mRegression;
+    int64_t mMaxDelayMs;
+
+    sp<TunnelRenderer> mRenderer;
+
+    status_t parseRTP(const sp<ABuffer> &buffer);
+    status_t parseRTCP(const sp<ABuffer> &buffer);
+    status_t parseBYE(const uint8_t *data, size_t size);
+    status_t parseSR(const uint8_t *data, size_t size);
+
+    void addSDES(const sp<ABuffer> &buffer);
+    void onSendRR();
+    void onPacketLost(const sp<AMessage> &msg);
+    void scheduleSendRR();
+
+    DISALLOW_EVIL_CONSTRUCTORS(RTPSink);
+};
+
+}  // namespace android
+
+#endif  // RTP_SINK_H_
diff --git a/media/libstagefright/wifi-display/sink/TunnelRenderer.cpp b/media/libstagefright/wifi-display/sink/TunnelRenderer.cpp
new file mode 100644
index 0000000..bc35aef
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/TunnelRenderer.cpp
@@ -0,0 +1,396 @@
+/*
+ * Copyright 2012, 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_NDEBUG 0
+#define LOG_TAG "TunnelRenderer"
+#include <utils/Log.h>
+
+#include "TunnelRenderer.h"
+
+#include "ATSParser.h"
+
+#include <binder/IMemory.h>
+#include <binder/IServiceManager.h>
+#include <gui/SurfaceComposerClient.h>
+#include <media/IMediaPlayerService.h>
+#include <media/IStreamSource.h>
+#include <media/stagefright/foundation/ABuffer.h>
+#include <media/stagefright/foundation/ADebug.h>
+#include <media/stagefright/foundation/AMessage.h>
+#include <ui/DisplayInfo.h>
+
+namespace android {
+
+struct TunnelRenderer::PlayerClient : public BnMediaPlayerClient {
+    PlayerClient() {}
+
+    virtual void notify(int msg, int ext1, int ext2, const Parcel *obj) {
+        ALOGI("notify %d, %d, %d", msg, ext1, ext2);
+    }
+
+protected:
+    virtual ~PlayerClient() {}
+
+private:
+    DISALLOW_EVIL_CONSTRUCTORS(PlayerClient);
+};
+
+struct TunnelRenderer::StreamSource : public BnStreamSource {
+    StreamSource(TunnelRenderer *owner);
+
+    virtual void setListener(const sp<IStreamListener> &listener);
+    virtual void setBuffers(const Vector<sp<IMemory> > &buffers);
+
+    virtual void onBufferAvailable(size_t index);
+
+    virtual uint32_t flags() const;
+
+    void doSomeWork();
+
+protected:
+    virtual ~StreamSource();
+
+private:
+    mutable Mutex mLock;
+
+    TunnelRenderer *mOwner;
+
+    sp<IStreamListener> mListener;
+
+    Vector<sp<IMemory> > mBuffers;
+    List<size_t> mIndicesAvailable;
+
+    size_t mNumDeqeued;
+
+    DISALLOW_EVIL_CONSTRUCTORS(StreamSource);
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+TunnelRenderer::StreamSource::StreamSource(TunnelRenderer *owner)
+    : mOwner(owner),
+      mNumDeqeued(0) {
+}
+
+TunnelRenderer::StreamSource::~StreamSource() {
+}
+
+void TunnelRenderer::StreamSource::setListener(
+        const sp<IStreamListener> &listener) {
+    mListener = listener;
+}
+
+void TunnelRenderer::StreamSource::setBuffers(
+        const Vector<sp<IMemory> > &buffers) {
+    mBuffers = buffers;
+}
+
+void TunnelRenderer::StreamSource::onBufferAvailable(size_t index) {
+    CHECK_LT(index, mBuffers.size());
+
+    {
+        Mutex::Autolock autoLock(mLock);
+        mIndicesAvailable.push_back(index);
+    }
+
+    doSomeWork();
+}
+
+uint32_t TunnelRenderer::StreamSource::flags() const {
+    return kFlagAlignedVideoData;
+}
+
+void TunnelRenderer::StreamSource::doSomeWork() {
+    Mutex::Autolock autoLock(mLock);
+
+    while (!mIndicesAvailable.empty()) {
+        sp<ABuffer> srcBuffer = mOwner->dequeueBuffer();
+        if (srcBuffer == NULL) {
+            break;
+        }
+
+        ++mNumDeqeued;
+
+        if (mNumDeqeued == 1) {
+            ALOGI("fixing real time now.");
+
+            sp<AMessage> extra = new AMessage;
+
+            extra->setInt32(
+                    IStreamListener::kKeyDiscontinuityMask,
+                    ATSParser::DISCONTINUITY_ABSOLUTE_TIME);
+
+            extra->setInt64("timeUs", ALooper::GetNowUs());
+
+            mListener->issueCommand(
+                    IStreamListener::DISCONTINUITY,
+                    false /* synchronous */,
+                    extra);
+        }
+
+        ALOGV("dequeue TS packet of size %d", srcBuffer->size());
+
+        size_t index = *mIndicesAvailable.begin();
+        mIndicesAvailable.erase(mIndicesAvailable.begin());
+
+        sp<IMemory> mem = mBuffers.itemAt(index);
+        CHECK_LE(srcBuffer->size(), mem->size());
+        CHECK_EQ((srcBuffer->size() % 188), 0u);
+
+        memcpy(mem->pointer(), srcBuffer->data(), srcBuffer->size());
+        mListener->queueBuffer(index, srcBuffer->size());
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+TunnelRenderer::TunnelRenderer(
+        const sp<AMessage> &notifyLost,
+        const sp<ISurfaceTexture> &surfaceTex)
+    : mNotifyLost(notifyLost),
+      mSurfaceTex(surfaceTex),
+      mTotalBytesQueued(0ll),
+      mLastDequeuedExtSeqNo(-1),
+      mFirstFailedAttemptUs(-1ll),
+      mRequestedRetransmission(false) {
+}
+
+TunnelRenderer::~TunnelRenderer() {
+    destroyPlayer();
+}
+
+void TunnelRenderer::queueBuffer(const sp<ABuffer> &buffer) {
+    Mutex::Autolock autoLock(mLock);
+
+    mTotalBytesQueued += buffer->size();
+
+    if (mPackets.empty()) {
+        mPackets.push_back(buffer);
+        return;
+    }
+
+    int32_t newExtendedSeqNo = buffer->int32Data();
+
+    List<sp<ABuffer> >::iterator firstIt = mPackets.begin();
+    List<sp<ABuffer> >::iterator it = --mPackets.end();
+    for (;;) {
+        int32_t extendedSeqNo = (*it)->int32Data();
+
+        if (extendedSeqNo == newExtendedSeqNo) {
+            // Duplicate packet.
+            return;
+        }
+
+        if (extendedSeqNo < newExtendedSeqNo) {
+            // Insert new packet after the one at "it".
+            mPackets.insert(++it, buffer);
+            return;
+        }
+
+        if (it == firstIt) {
+            // Insert new packet before the first existing one.
+            mPackets.insert(it, buffer);
+            return;
+        }
+
+        --it;
+    }
+}
+
+sp<ABuffer> TunnelRenderer::dequeueBuffer() {
+    Mutex::Autolock autoLock(mLock);
+
+    sp<ABuffer> buffer;
+    int32_t extSeqNo;
+    while (!mPackets.empty()) {
+        buffer = *mPackets.begin();
+        extSeqNo = buffer->int32Data();
+
+        if (mLastDequeuedExtSeqNo < 0 || extSeqNo > mLastDequeuedExtSeqNo) {
+            break;
+        }
+
+        // This is a retransmission of a packet we've already returned.
+
+        mTotalBytesQueued -= buffer->size();
+        buffer.clear();
+        extSeqNo = -1;
+
+        mPackets.erase(mPackets.begin());
+    }
+
+    if (mPackets.empty()) {
+        if (mFirstFailedAttemptUs < 0ll) {
+            mFirstFailedAttemptUs = ALooper::GetNowUs();
+            mRequestedRetransmission = false;
+        } else {
+            ALOGV("no packets available for %.2f secs",
+                    (ALooper::GetNowUs() - mFirstFailedAttemptUs) / 1E6);
+        }
+
+        return NULL;
+    }
+
+    if (mLastDequeuedExtSeqNo < 0 || extSeqNo == mLastDequeuedExtSeqNo + 1) {
+        if (mRequestedRetransmission) {
+            ALOGI("Recovered after requesting retransmission of %d",
+                  extSeqNo);
+        }
+
+        mLastDequeuedExtSeqNo = extSeqNo;
+        mFirstFailedAttemptUs = -1ll;
+        mRequestedRetransmission = false;
+
+        mPackets.erase(mPackets.begin());
+
+        mTotalBytesQueued -= buffer->size();
+
+        return buffer;
+    }
+
+    if (mFirstFailedAttemptUs < 0ll) {
+        mFirstFailedAttemptUs = ALooper::GetNowUs();
+
+        ALOGI("failed to get the correct packet the first time.");
+        return NULL;
+    }
+
+    if (mFirstFailedAttemptUs + 50000ll > ALooper::GetNowUs()) {
+        // We're willing to wait a little while to get the right packet.
+
+        if (!mRequestedRetransmission) {
+            ALOGI("requesting retransmission of seqNo %d",
+                  (mLastDequeuedExtSeqNo + 1) & 0xffff);
+
+            sp<AMessage> notify = mNotifyLost->dup();
+            notify->setInt32("seqNo", (mLastDequeuedExtSeqNo + 1) & 0xffff);
+            notify->post();
+
+            mRequestedRetransmission = true;
+        } else {
+            ALOGI("still waiting for the correct packet to arrive.");
+        }
+
+        return NULL;
+    }
+
+    ALOGI("dropping packet. extSeqNo %d didn't arrive in time",
+            mLastDequeuedExtSeqNo + 1);
+
+    // Permanent failure, we never received the packet.
+    mLastDequeuedExtSeqNo = extSeqNo;
+    mFirstFailedAttemptUs = -1ll;
+    mRequestedRetransmission = false;
+
+    mTotalBytesQueued -= buffer->size();
+
+    mPackets.erase(mPackets.begin());
+
+    return buffer;
+}
+
+void TunnelRenderer::onMessageReceived(const sp<AMessage> &msg) {
+    switch (msg->what()) {
+        case kWhatQueueBuffer:
+        {
+            sp<ABuffer> buffer;
+            CHECK(msg->findBuffer("buffer", &buffer));
+
+            queueBuffer(buffer);
+
+            if (mStreamSource == NULL) {
+                if (mTotalBytesQueued > 0ll) {
+                    initPlayer();
+                } else {
+                    ALOGI("Have %lld bytes queued...", mTotalBytesQueued);
+                }
+            } else {
+                mStreamSource->doSomeWork();
+            }
+            break;
+        }
+
+        default:
+            TRESPASS();
+    }
+}
+
+void TunnelRenderer::initPlayer() {
+    if (mSurfaceTex == NULL) {
+        mComposerClient = new SurfaceComposerClient;
+        CHECK_EQ(mComposerClient->initCheck(), (status_t)OK);
+
+        DisplayInfo info;
+        SurfaceComposerClient::getDisplayInfo(0, &info);
+        ssize_t displayWidth = info.w;
+        ssize_t displayHeight = info.h;
+
+        mSurfaceControl =
+            mComposerClient->createSurface(
+                    String8("A Surface"),
+                    displayWidth,
+                    displayHeight,
+                    PIXEL_FORMAT_RGB_565,
+                    0);
+
+        CHECK(mSurfaceControl != NULL);
+        CHECK(mSurfaceControl->isValid());
+
+        SurfaceComposerClient::openGlobalTransaction();
+        CHECK_EQ(mSurfaceControl->setLayer(INT_MAX), (status_t)OK);
+        CHECK_EQ(mSurfaceControl->show(), (status_t)OK);
+        SurfaceComposerClient::closeGlobalTransaction();
+
+        mSurface = mSurfaceControl->getSurface();
+        CHECK(mSurface != NULL);
+    }
+
+    sp<IServiceManager> sm = defaultServiceManager();
+    sp<IBinder> binder = sm->getService(String16("media.player"));
+    sp<IMediaPlayerService> service = interface_cast<IMediaPlayerService>(binder);
+    CHECK(service.get() != NULL);
+
+    mStreamSource = new StreamSource(this);
+
+    mPlayerClient = new PlayerClient;
+
+    mPlayer = service->create(getpid(), mPlayerClient, 0);
+    CHECK(mPlayer != NULL);
+    CHECK_EQ(mPlayer->setDataSource(mStreamSource), (status_t)OK);
+
+    mPlayer->setVideoSurfaceTexture(
+            mSurfaceTex != NULL ? mSurfaceTex : mSurface->getSurfaceTexture());
+
+    mPlayer->start();
+}
+
+void TunnelRenderer::destroyPlayer() {
+    mStreamSource.clear();
+
+    mPlayer->stop();
+    mPlayer.clear();
+
+    if (mSurfaceTex == NULL) {
+        mSurface.clear();
+        mSurfaceControl.clear();
+
+        mComposerClient->dispose();
+        mComposerClient.clear();
+    }
+}
+
+}  // namespace android
+
diff --git a/media/libstagefright/wifi-display/sink/TunnelRenderer.h b/media/libstagefright/wifi-display/sink/TunnelRenderer.h
new file mode 100644
index 0000000..c9597e0
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/TunnelRenderer.h
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2012, 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.
+ */
+
+#ifndef TUNNEL_RENDERER_H_
+
+#define TUNNEL_RENDERER_H_
+
+#include <gui/Surface.h>
+#include <media/stagefright/foundation/AHandler.h>
+
+namespace android {
+
+struct ABuffer;
+struct SurfaceComposerClient;
+struct SurfaceControl;
+struct Surface;
+struct IMediaPlayer;
+struct IStreamListener;
+
+// This class reassembles incoming RTP packets into the correct order
+// and sends the resulting transport stream to a mediaplayer instance
+// for playback.
+struct TunnelRenderer : public AHandler {
+    TunnelRenderer(
+            const sp<AMessage> &notifyLost,
+            const sp<ISurfaceTexture> &surfaceTex);
+
+    sp<ABuffer> dequeueBuffer();
+
+    enum {
+        kWhatQueueBuffer,
+    };
+
+protected:
+    virtual void onMessageReceived(const sp<AMessage> &msg);
+    virtual ~TunnelRenderer();
+
+private:
+    struct PlayerClient;
+    struct StreamSource;
+
+    mutable Mutex mLock;
+
+    sp<AMessage> mNotifyLost;
+    sp<ISurfaceTexture> mSurfaceTex;
+
+    List<sp<ABuffer> > mPackets;
+    int64_t mTotalBytesQueued;
+
+    sp<SurfaceComposerClient> mComposerClient;
+    sp<SurfaceControl> mSurfaceControl;
+    sp<Surface> mSurface;
+    sp<PlayerClient> mPlayerClient;
+    sp<IMediaPlayer> mPlayer;
+    sp<StreamSource> mStreamSource;
+
+    int32_t mLastDequeuedExtSeqNo;
+    int64_t mFirstFailedAttemptUs;
+    bool mRequestedRetransmission;
+
+    void initPlayer();
+    void destroyPlayer();
+
+    void queueBuffer(const sp<ABuffer> &buffer);
+
+    DISALLOW_EVIL_CONSTRUCTORS(TunnelRenderer);
+};
+
+}  // namespace android
+
+#endif  // TUNNEL_RENDERER_H_
diff --git a/media/libstagefright/wifi-display/sink/WifiDisplaySink.cpp b/media/libstagefright/wifi-display/sink/WifiDisplaySink.cpp
new file mode 100644
index 0000000..fcd20d4
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/WifiDisplaySink.cpp
@@ -0,0 +1,644 @@
+/*
+ * Copyright 2012, 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_NDEBUG 0
+#define LOG_TAG "WifiDisplaySink"
+#include <utils/Log.h>
+
+#include "WifiDisplaySink.h"
+#include "ParsedMessage.h"
+#include "RTPSink.h"
+
+#include <media/stagefright/foundation/ABuffer.h>
+#include <media/stagefright/foundation/ADebug.h>
+#include <media/stagefright/foundation/AMessage.h>
+#include <media/stagefright/MediaErrors.h>
+
+namespace android {
+
+WifiDisplaySink::WifiDisplaySink(
+        const sp<ANetworkSession> &netSession,
+        const sp<ISurfaceTexture> &surfaceTex)
+    : mState(UNDEFINED),
+      mNetSession(netSession),
+      mSurfaceTex(surfaceTex),
+      mSessionID(0),
+      mNextCSeq(1) {
+}
+
+WifiDisplaySink::~WifiDisplaySink() {
+}
+
+void WifiDisplaySink::start(const char *sourceHost, int32_t sourcePort) {
+    sp<AMessage> msg = new AMessage(kWhatStart, id());
+    msg->setString("sourceHost", sourceHost);
+    msg->setInt32("sourcePort", sourcePort);
+    msg->post();
+}
+
+void WifiDisplaySink::start(const char *uri) {
+    sp<AMessage> msg = new AMessage(kWhatStart, id());
+    msg->setString("setupURI", uri);
+    msg->post();
+}
+
+// static
+bool WifiDisplaySink::ParseURL(
+        const char *url, AString *host, int32_t *port, AString *path,
+        AString *user, AString *pass) {
+    host->clear();
+    *port = 0;
+    path->clear();
+    user->clear();
+    pass->clear();
+
+    if (strncasecmp("rtsp://", url, 7)) {
+        return false;
+    }
+
+    const char *slashPos = strchr(&url[7], '/');
+
+    if (slashPos == NULL) {
+        host->setTo(&url[7]);
+        path->setTo("/");
+    } else {
+        host->setTo(&url[7], slashPos - &url[7]);
+        path->setTo(slashPos);
+    }
+
+    ssize_t atPos = host->find("@");
+
+    if (atPos >= 0) {
+        // Split of user:pass@ from hostname.
+
+        AString userPass(*host, 0, atPos);
+        host->erase(0, atPos + 1);
+
+        ssize_t colonPos = userPass.find(":");
+
+        if (colonPos < 0) {
+            *user = userPass;
+        } else {
+            user->setTo(userPass, 0, colonPos);
+            pass->setTo(userPass, colonPos + 1, userPass.size() - colonPos - 1);
+        }
+    }
+
+    const char *colonPos = strchr(host->c_str(), ':');
+
+    if (colonPos != NULL) {
+        char *end;
+        unsigned long x = strtoul(colonPos + 1, &end, 10);
+
+        if (end == colonPos + 1 || *end != '\0' || x >= 65536) {
+            return false;
+        }
+
+        *port = x;
+
+        size_t colonOffset = colonPos - host->c_str();
+        size_t trailing = host->size() - colonOffset;
+        host->erase(colonOffset, trailing);
+    } else {
+        *port = 554;
+    }
+
+    return true;
+}
+
+void WifiDisplaySink::onMessageReceived(const sp<AMessage> &msg) {
+    switch (msg->what()) {
+        case kWhatStart:
+        {
+            int32_t sourcePort;
+
+            if (msg->findString("setupURI", &mSetupURI)) {
+                AString path, user, pass;
+                CHECK(ParseURL(
+                            mSetupURI.c_str(),
+                            &mRTSPHost, &sourcePort, &path, &user, &pass)
+                        && user.empty() && pass.empty());
+            } else {
+                CHECK(msg->findString("sourceHost", &mRTSPHost));
+                CHECK(msg->findInt32("sourcePort", &sourcePort));
+            }
+
+            sp<AMessage> notify = new AMessage(kWhatRTSPNotify, id());
+
+            status_t err = mNetSession->createRTSPClient(
+                    mRTSPHost.c_str(), sourcePort, notify, &mSessionID);
+            CHECK_EQ(err, (status_t)OK);
+
+            mState = CONNECTING;
+            break;
+        }
+
+        case kWhatRTSPNotify:
+        {
+            int32_t reason;
+            CHECK(msg->findInt32("reason", &reason));
+
+            switch (reason) {
+                case ANetworkSession::kWhatError:
+                {
+                    int32_t sessionID;
+                    CHECK(msg->findInt32("sessionID", &sessionID));
+
+                    int32_t err;
+                    CHECK(msg->findInt32("err", &err));
+
+                    AString detail;
+                    CHECK(msg->findString("detail", &detail));
+
+                    ALOGE("An error occurred in session %d (%d, '%s/%s').",
+                          sessionID,
+                          err,
+                          detail.c_str(),
+                          strerror(-err));
+
+                    if (sessionID == mSessionID) {
+                        ALOGI("Lost control connection.");
+
+                        // The control connection is dead now.
+                        mNetSession->destroySession(mSessionID);
+                        mSessionID = 0;
+
+                        looper()->stop();
+                    }
+                    break;
+                }
+
+                case ANetworkSession::kWhatConnected:
+                {
+                    ALOGI("We're now connected.");
+                    mState = CONNECTED;
+
+                    if (!mSetupURI.empty()) {
+                        status_t err =
+                            sendDescribe(mSessionID, mSetupURI.c_str());
+
+                        CHECK_EQ(err, (status_t)OK);
+                    }
+                    break;
+                }
+
+                case ANetworkSession::kWhatData:
+                {
+                    onReceiveClientData(msg);
+                    break;
+                }
+
+                case ANetworkSession::kWhatBinaryData:
+                {
+                    CHECK(sUseTCPInterleaving);
+
+                    int32_t channel;
+                    CHECK(msg->findInt32("channel", &channel));
+
+                    sp<ABuffer> data;
+                    CHECK(msg->findBuffer("data", &data));
+
+                    mRTPSink->injectPacket(channel == 0 /* isRTP */, data);
+                    break;
+                }
+
+                default:
+                    TRESPASS();
+            }
+            break;
+        }
+
+        case kWhatStop:
+        {
+            looper()->stop();
+            break;
+        }
+
+        default:
+            TRESPASS();
+    }
+}
+
+void WifiDisplaySink::registerResponseHandler(
+        int32_t sessionID, int32_t cseq, HandleRTSPResponseFunc func) {
+    ResponseID id;
+    id.mSessionID = sessionID;
+    id.mCSeq = cseq;
+    mResponseHandlers.add(id, func);
+}
+
+status_t WifiDisplaySink::sendM2(int32_t sessionID) {
+    AString request = "OPTIONS * RTSP/1.0\r\n";
+    AppendCommonResponse(&request, mNextCSeq);
+
+    request.append(
+            "Require: org.wfa.wfd1.0\r\n"
+            "\r\n");
+
+    status_t err =
+        mNetSession->sendRequest(sessionID, request.c_str(), request.size());
+
+    if (err != OK) {
+        return err;
+    }
+
+    registerResponseHandler(
+            sessionID, mNextCSeq, &WifiDisplaySink::onReceiveM2Response);
+
+    ++mNextCSeq;
+
+    return OK;
+}
+
+status_t WifiDisplaySink::onReceiveM2Response(
+        int32_t sessionID, const sp<ParsedMessage> &msg) {
+    int32_t statusCode;
+    if (!msg->getStatusCode(&statusCode)) {
+        return ERROR_MALFORMED;
+    }
+
+    if (statusCode != 200) {
+        return ERROR_UNSUPPORTED;
+    }
+
+    return OK;
+}
+
+status_t WifiDisplaySink::onReceiveDescribeResponse(
+        int32_t sessionID, const sp<ParsedMessage> &msg) {
+    int32_t statusCode;
+    if (!msg->getStatusCode(&statusCode)) {
+        return ERROR_MALFORMED;
+    }
+
+    if (statusCode != 200) {
+        return ERROR_UNSUPPORTED;
+    }
+
+    return sendSetup(sessionID, mSetupURI.c_str());
+}
+
+status_t WifiDisplaySink::onReceiveSetupResponse(
+        int32_t sessionID, const sp<ParsedMessage> &msg) {
+    int32_t statusCode;
+    if (!msg->getStatusCode(&statusCode)) {
+        return ERROR_MALFORMED;
+    }
+
+    if (statusCode != 200) {
+        return ERROR_UNSUPPORTED;
+    }
+
+    if (!msg->findString("session", &mPlaybackSessionID)) {
+        return ERROR_MALFORMED;
+    }
+
+    if (!ParsedMessage::GetInt32Attribute(
+                mPlaybackSessionID.c_str(),
+                "timeout",
+                &mPlaybackSessionTimeoutSecs)) {
+        mPlaybackSessionTimeoutSecs = -1;
+    }
+
+    ssize_t colonPos = mPlaybackSessionID.find(";");
+    if (colonPos >= 0) {
+        // Strip any options from the returned session id.
+        mPlaybackSessionID.erase(
+                colonPos, mPlaybackSessionID.size() - colonPos);
+    }
+
+    status_t err = configureTransport(msg);
+
+    if (err != OK) {
+        return err;
+    }
+
+    mState = PAUSED;
+
+    return sendPlay(
+            sessionID,
+            !mSetupURI.empty()
+                ? mSetupURI.c_str() : "rtsp://x.x.x.x:x/wfd1.0/streamid=0");
+}
+
+status_t WifiDisplaySink::configureTransport(const sp<ParsedMessage> &msg) {
+    if (sUseTCPInterleaving) {
+        return OK;
+    }
+
+    AString transport;
+    if (!msg->findString("transport", &transport)) {
+        ALOGE("Missing 'transport' field in SETUP response.");
+        return ERROR_MALFORMED;
+    }
+
+    AString sourceHost;
+    if (!ParsedMessage::GetAttribute(
+                transport.c_str(), "source", &sourceHost)) {
+        sourceHost = mRTSPHost;
+    }
+
+    AString serverPortStr;
+    if (!ParsedMessage::GetAttribute(
+                transport.c_str(), "server_port", &serverPortStr)) {
+        ALOGE("Missing 'server_port' in Transport field.");
+        return ERROR_MALFORMED;
+    }
+
+    int rtpPort, rtcpPort;
+    if (sscanf(serverPortStr.c_str(), "%d-%d", &rtpPort, &rtcpPort) != 2
+            || rtpPort <= 0 || rtpPort > 65535
+            || rtcpPort <=0 || rtcpPort > 65535
+            || rtcpPort != rtpPort + 1) {
+        ALOGE("Invalid server_port description '%s'.",
+                serverPortStr.c_str());
+
+        return ERROR_MALFORMED;
+    }
+
+    if (rtpPort & 1) {
+        ALOGW("Server picked an odd numbered RTP port.");
+    }
+
+    return mRTPSink->connect(sourceHost.c_str(), rtpPort, rtcpPort);
+}
+
+status_t WifiDisplaySink::onReceivePlayResponse(
+        int32_t sessionID, const sp<ParsedMessage> &msg) {
+    int32_t statusCode;
+    if (!msg->getStatusCode(&statusCode)) {
+        return ERROR_MALFORMED;
+    }
+
+    if (statusCode != 200) {
+        return ERROR_UNSUPPORTED;
+    }
+
+    mState = PLAYING;
+
+    return OK;
+}
+
+void WifiDisplaySink::onReceiveClientData(const sp<AMessage> &msg) {
+    int32_t sessionID;
+    CHECK(msg->findInt32("sessionID", &sessionID));
+
+    sp<RefBase> obj;
+    CHECK(msg->findObject("data", &obj));
+
+    sp<ParsedMessage> data =
+        static_cast<ParsedMessage *>(obj.get());
+
+    ALOGV("session %d received '%s'",
+          sessionID, data->debugString().c_str());
+
+    AString method;
+    AString uri;
+    data->getRequestField(0, &method);
+
+    int32_t cseq;
+    if (!data->findInt32("cseq", &cseq)) {
+        sendErrorResponse(sessionID, "400 Bad Request", -1 /* cseq */);
+        return;
+    }
+
+    if (method.startsWith("RTSP/")) {
+        // This is a response.
+
+        ResponseID id;
+        id.mSessionID = sessionID;
+        id.mCSeq = cseq;
+
+        ssize_t index = mResponseHandlers.indexOfKey(id);
+
+        if (index < 0) {
+            ALOGW("Received unsolicited server response, cseq %d", cseq);
+            return;
+        }
+
+        HandleRTSPResponseFunc func = mResponseHandlers.valueAt(index);
+        mResponseHandlers.removeItemsAt(index);
+
+        status_t err = (this->*func)(sessionID, data);
+        CHECK_EQ(err, (status_t)OK);
+    } else {
+        AString version;
+        data->getRequestField(2, &version);
+        if (!(version == AString("RTSP/1.0"))) {
+            sendErrorResponse(sessionID, "505 RTSP Version not supported", cseq);
+            return;
+        }
+
+        if (method == "OPTIONS") {
+            onOptionsRequest(sessionID, cseq, data);
+        } else if (method == "GET_PARAMETER") {
+            onGetParameterRequest(sessionID, cseq, data);
+        } else if (method == "SET_PARAMETER") {
+            onSetParameterRequest(sessionID, cseq, data);
+        } else {
+            sendErrorResponse(sessionID, "405 Method Not Allowed", cseq);
+        }
+    }
+}
+
+void WifiDisplaySink::onOptionsRequest(
+        int32_t sessionID,
+        int32_t cseq,
+        const sp<ParsedMessage> &data) {
+    AString response = "RTSP/1.0 200 OK\r\n";
+    AppendCommonResponse(&response, cseq);
+    response.append("Public: org.wfa.wfd1.0, GET_PARAMETER, SET_PARAMETER\r\n");
+    response.append("\r\n");
+
+    status_t err = mNetSession->sendRequest(sessionID, response.c_str());
+    CHECK_EQ(err, (status_t)OK);
+
+    err = sendM2(sessionID);
+    CHECK_EQ(err, (status_t)OK);
+}
+
+void WifiDisplaySink::onGetParameterRequest(
+        int32_t sessionID,
+        int32_t cseq,
+        const sp<ParsedMessage> &data) {
+    AString body =
+        "wfd_video_formats: xxx\r\n"
+        "wfd_audio_codecs: xxx\r\n"
+        "wfd_client_rtp_ports: RTP/AVP/UDP;unicast xxx 0 mode=play\r\n";
+
+    AString response = "RTSP/1.0 200 OK\r\n";
+    AppendCommonResponse(&response, cseq);
+    response.append("Content-Type: text/parameters\r\n");
+    response.append(StringPrintf("Content-Length: %d\r\n", body.size()));
+    response.append("\r\n");
+    response.append(body);
+
+    status_t err = mNetSession->sendRequest(sessionID, response.c_str());
+    CHECK_EQ(err, (status_t)OK);
+}
+
+status_t WifiDisplaySink::sendDescribe(int32_t sessionID, const char *uri) {
+    uri = "rtsp://xwgntvx.is.livestream-api.com/livestreamiphone/wgntv";
+    uri = "rtsp://v2.cache6.c.youtube.com/video.3gp?cid=e101d4bf280055f9&fmt=18";
+
+    AString request = StringPrintf("DESCRIBE %s RTSP/1.0\r\n", uri);
+    AppendCommonResponse(&request, mNextCSeq);
+
+    request.append("Accept: application/sdp\r\n");
+    request.append("\r\n");
+
+    status_t err = mNetSession->sendRequest(
+            sessionID, request.c_str(), request.size());
+
+    if (err != OK) {
+        return err;
+    }
+
+    registerResponseHandler(
+            sessionID, mNextCSeq, &WifiDisplaySink::onReceiveDescribeResponse);
+
+    ++mNextCSeq;
+
+    return OK;
+}
+
+status_t WifiDisplaySink::sendSetup(int32_t sessionID, const char *uri) {
+    mRTPSink = new RTPSink(mNetSession, mSurfaceTex);
+    looper()->registerHandler(mRTPSink);
+
+    status_t err = mRTPSink->init(sUseTCPInterleaving);
+
+    if (err != OK) {
+        looper()->unregisterHandler(mRTPSink->id());
+        mRTPSink.clear();
+        return err;
+    }
+
+    AString request = StringPrintf("SETUP %s RTSP/1.0\r\n", uri);
+
+    AppendCommonResponse(&request, mNextCSeq);
+
+    if (sUseTCPInterleaving) {
+        request.append("Transport: RTP/AVP/TCP;interleaved=0-1\r\n");
+    } else {
+        int32_t rtpPort = mRTPSink->getRTPPort();
+
+        request.append(
+                StringPrintf(
+                    "Transport: RTP/AVP/UDP;unicast;client_port=%d-%d\r\n",
+                    rtpPort, rtpPort + 1));
+    }
+
+    request.append("\r\n");
+
+    ALOGV("request = '%s'", request.c_str());
+
+    err = mNetSession->sendRequest(sessionID, request.c_str(), request.size());
+
+    if (err != OK) {
+        return err;
+    }
+
+    registerResponseHandler(
+            sessionID, mNextCSeq, &WifiDisplaySink::onReceiveSetupResponse);
+
+    ++mNextCSeq;
+
+    return OK;
+}
+
+status_t WifiDisplaySink::sendPlay(int32_t sessionID, const char *uri) {
+    AString request = StringPrintf("PLAY %s RTSP/1.0\r\n", uri);
+
+    AppendCommonResponse(&request, mNextCSeq);
+
+    request.append(StringPrintf("Session: %s\r\n", mPlaybackSessionID.c_str()));
+    request.append("\r\n");
+
+    status_t err =
+        mNetSession->sendRequest(sessionID, request.c_str(), request.size());
+
+    if (err != OK) {
+        return err;
+    }
+
+    registerResponseHandler(
+            sessionID, mNextCSeq, &WifiDisplaySink::onReceivePlayResponse);
+
+    ++mNextCSeq;
+
+    return OK;
+}
+
+void WifiDisplaySink::onSetParameterRequest(
+        int32_t sessionID,
+        int32_t cseq,
+        const sp<ParsedMessage> &data) {
+    const char *content = data->getContent();
+
+    if (strstr(content, "wfd_trigger_method: SETUP\r\n") != NULL) {
+        status_t err =
+            sendSetup(
+                    sessionID,
+                    "rtsp://x.x.x.x:x/wfd1.0/streamid=0");
+
+        CHECK_EQ(err, (status_t)OK);
+    }
+
+    AString response = "RTSP/1.0 200 OK\r\n";
+    AppendCommonResponse(&response, cseq);
+    response.append("\r\n");
+
+    status_t err = mNetSession->sendRequest(sessionID, response.c_str());
+    CHECK_EQ(err, (status_t)OK);
+}
+
+void WifiDisplaySink::sendErrorResponse(
+        int32_t sessionID,
+        const char *errorDetail,
+        int32_t cseq) {
+    AString response;
+    response.append("RTSP/1.0 ");
+    response.append(errorDetail);
+    response.append("\r\n");
+
+    AppendCommonResponse(&response, cseq);
+
+    response.append("\r\n");
+
+    status_t err = mNetSession->sendRequest(sessionID, response.c_str());
+    CHECK_EQ(err, (status_t)OK);
+}
+
+// static
+void WifiDisplaySink::AppendCommonResponse(AString *response, int32_t cseq) {
+    time_t now = time(NULL);
+    struct tm *now2 = gmtime(&now);
+    char buf[128];
+    strftime(buf, sizeof(buf), "%a, %d %b %Y %H:%M:%S %z", now2);
+
+    response->append("Date: ");
+    response->append(buf);
+    response->append("\r\n");
+
+    response->append("User-Agent: stagefright/1.1 (Linux;Android 4.1)\r\n");
+
+    if (cseq >= 0) {
+        response->append(StringPrintf("CSeq: %d\r\n", cseq));
+    }
+}
+
+}  // namespace android
diff --git a/media/libstagefright/wifi-display/sink/WifiDisplaySink.h b/media/libstagefright/wifi-display/sink/WifiDisplaySink.h
new file mode 100644
index 0000000..f886ee5
--- /dev/null
+++ b/media/libstagefright/wifi-display/sink/WifiDisplaySink.h
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2012, 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.
+ */
+
+#ifndef WIFI_DISPLAY_SINK_H_
+
+#define WIFI_DISPLAY_SINK_H_
+
+#include "ANetworkSession.h"
+
+#include <gui/Surface.h>
+#include <media/stagefright/foundation/AHandler.h>
+
+namespace android {
+
+struct ParsedMessage;
+struct RTPSink;
+
+// Represents the RTSP client acting as a wifi display sink.
+// Connects to a wifi display source and renders the incoming
+// transport stream using a MediaPlayer instance.
+struct WifiDisplaySink : public AHandler {
+    WifiDisplaySink(
+            const sp<ANetworkSession> &netSession,
+            const sp<ISurfaceTexture> &surfaceTex = NULL);
+
+    void start(const char *sourceHost, int32_t sourcePort);
+    void start(const char *uri);
+
+protected:
+    virtual ~WifiDisplaySink();
+    virtual void onMessageReceived(const sp<AMessage> &msg);
+
+private:
+    enum State {
+        UNDEFINED,
+        CONNECTING,
+        CONNECTED,
+        PAUSED,
+        PLAYING,
+    };
+
+    enum {
+        kWhatStart,
+        kWhatRTSPNotify,
+        kWhatStop,
+    };
+
+    struct ResponseID {
+        int32_t mSessionID;
+        int32_t mCSeq;
+
+        bool operator<(const ResponseID &other) const {
+            return mSessionID < other.mSessionID
+                || (mSessionID == other.mSessionID
+                        && mCSeq < other.mCSeq);
+        }
+    };
+
+    typedef status_t (WifiDisplaySink::*HandleRTSPResponseFunc)(
+            int32_t sessionID, const sp<ParsedMessage> &msg);
+
+    static const bool sUseTCPInterleaving = false;
+
+    State mState;
+    sp<ANetworkSession> mNetSession;
+    sp<ISurfaceTexture> mSurfaceTex;
+    AString mSetupURI;
+    AString mRTSPHost;
+    int32_t mSessionID;
+
+    int32_t mNextCSeq;
+
+    KeyedVector<ResponseID, HandleRTSPResponseFunc> mResponseHandlers;
+
+    sp<RTPSink> mRTPSink;
+    AString mPlaybackSessionID;
+    int32_t mPlaybackSessionTimeoutSecs;
+
+    status_t sendM2(int32_t sessionID);
+    status_t sendDescribe(int32_t sessionID, const char *uri);
+    status_t sendSetup(int32_t sessionID, const char *uri);
+    status_t sendPlay(int32_t sessionID, const char *uri);
+
+    status_t onReceiveM2Response(
+            int32_t sessionID, const sp<ParsedMessage> &msg);
+
+    status_t onReceiveDescribeResponse(
+            int32_t sessionID, const sp<ParsedMessage> &msg);
+
+    status_t onReceiveSetupResponse(
+            int32_t sessionID, const sp<ParsedMessage> &msg);
+
+    status_t configureTransport(const sp<ParsedMessage> &msg);
+
+    status_t onReceivePlayResponse(
+            int32_t sessionID, const sp<ParsedMessage> &msg);
+
+    void registerResponseHandler(
+            int32_t sessionID, int32_t cseq, HandleRTSPResponseFunc func);
+
+    void onReceiveClientData(const sp<AMessage> &msg);
+
+    void onOptionsRequest(
+            int32_t sessionID,
+            int32_t cseq,
+            const sp<ParsedMessage> &data);
+
+    void onGetParameterRequest(
+            int32_t sessionID,
+            int32_t cseq,
+            const sp<ParsedMessage> &data);
+
+    void onSetParameterRequest(
+            int32_t sessionID,
+            int32_t cseq,
+            const sp<ParsedMessage> &data);
+
+    void sendErrorResponse(
+            int32_t sessionID,
+            const char *errorDetail,
+            int32_t cseq);
+
+    static void AppendCommonResponse(AString *response, int32_t cseq);
+
+    bool ParseURL(
+            const char *url, AString *host, int32_t *port, AString *path,
+            AString *user, AString *pass);
+
+    DISALLOW_EVIL_CONSTRUCTORS(WifiDisplaySink);
+};
+
+}  // namespace android
+
+#endif  // WIFI_DISPLAY_SINK_H_
diff --git a/media/libstagefright/wifi-display/wfd.cpp b/media/libstagefright/wifi-display/wfd.cpp
index 5e7d9fd..d886f14 100644
--- a/media/libstagefright/wifi-display/wfd.cpp
+++ b/media/libstagefright/wifi-display/wfd.cpp
@@ -18,11 +18,8 @@
 #define LOG_TAG "wfd"
 #include <utils/Log.h>
 
-#define SUPPORT_SINK    0
-
-#if SUPPORT_SINK
 #include "sink/WifiDisplaySink.h"
-#endif
+#include "source/WifiDisplaySource.h"
 
 #include <binder/ProcessState.h>
 #include <binder/IServiceManager.h>
@@ -49,10 +46,8 @@ static void enableDisableRemoteDisplay(const char *iface) {
 static void usage(const char *me) {
     fprintf(stderr,
             "usage:\n"
-#if SUPPORT_SINK
             "           %s -c host[:port]\tconnect to wifi source\n"
             "           -u uri        \tconnect to an rtsp uri\n"
-#endif
             "           -e ip[:port]       \tenable remote display\n"
             "           -d            \tdisable remote display\n",
             me);
@@ -72,7 +67,6 @@ int main(int argc, char **argv) {
     int res;
     while ((res = getopt(argc, argv, "hc:l:u:e:d")) >= 0) {
         switch (res) {
-#if SUPPORT_SINK
             case 'c':
             {
                 const char *colonPos = strrchr(optarg, ':');
@@ -100,7 +94,6 @@ int main(int argc, char **argv) {
                 uri = optarg;
                 break;
             }
-#endif
 
             case 'e':
             {
@@ -124,7 +117,6 @@ int main(int argc, char **argv) {
         }
     }
 
-#if SUPPORT_SINK
     if (connectToPort < 0 && uri.empty()) {
         fprintf(stderr,
                 "You need to select either source host or uri.\n");
@@ -154,7 +146,6 @@ int main(int argc, char **argv) {
     }
 
     looper->start(true /* runOnCallingThread */);
-#endif
 
     return 0;
 }