protocolcurl.cpp

/*
 * protocolcurl.c: IPTV plugin for the Video Disk Recorder
 *
 * See the README file for copyright information and how to reach the author.
 *
 */

#include "common.h"
#include "config.h"
#include "log.h"
#include "protocolcurl.h"

#ifdef CURLOPT_RTSPHEADER
#define USE_RTSP
#endif

#define iptv_curl_easy_setopt(X, Y, Z) \
  if ((res = curl_easy_setopt((X), (Y), (Z))) != CURLE_OK) { \
     error("curl_easy_setopt(%s, %s) failed: %s (%d)", #Y, #Z, curl_easy_strerror(res), res); \
     }

#define iptv_curl_easy_perform(X) \
  if ((res = curl_easy_perform((X))) != CURLE_OK) { \
     error("curl_easy_perform() failed: %s (%d)", curl_easy_strerror(res), res); \
     }

cIptvProtocolCurl::cIptvProtocolCurl()
    : streamUrlM(""),
      streamParamM(0),
      streamPortM(0),
      mutexM(),
      handleM(nullptr),
      multiM(nullptr),
      headerListM(nullptr),
      ringBufferM(new cRingBufferLinear(IPTV_BUFFER_SIZE, 7*TS_SIZE, false, "IPTV CURL")),
      rtspControlM(""),
      modeM(eModeUnknown),
      timeoutM(),
      connectedM(false),
      pausedM(false) {
    debug1("%s", __PRETTY_FUNCTION__);

    if (ringBufferM) {
        ringBufferM->SetTimeouts(100, 0);
        ringBufferM->SetIoThrottle();
    }

    Connect();
}

cIptvProtocolCurl::~cIptvProtocolCurl() {
    debug1("%s", __PRETTY_FUNCTION__);

    Disconnect();
    // Free allocated memory
    DELETE_POINTER(ringBufferM);
}

int cIptvProtocolCurl::DebugCallback(CURL *handleP, curl_infotype typeP, char *dataP, size_t sizeP, void *userPtrP) {
    cIptvProtocolCurl *obj = reinterpret_cast<cIptvProtocolCurl *>(dataP);

    if (obj) {
        switch (typeP) {
        case CURLINFO_TEXT:debug8("%s INFO %.*s", __PRETTY_FUNCTION__, (int) sizeP, dataP);
            break;

        case CURLINFO_HEADER_IN:debug8("%s HEAD <<< %.*s", __PRETTY_FUNCTION__, (int) sizeP, dataP);
            break;

        case CURLINFO_HEADER_OUT:debug8("%s HEAD >>>\n%.*s", __PRETTY_FUNCTION__, (int) sizeP, dataP);
            break;

        case CURLINFO_DATA_IN:debug8("%s DATA <<< %zu", __PRETTY_FUNCTION__, sizeP);
            break;

        case CURLINFO_DATA_OUT:debug8("%s DATA >>> %zu", __PRETTY_FUNCTION__, sizeP);
            break;

        default:
            break;
        }
    }

    return 0;
}

size_t cIptvProtocolCurl::WriteCallback(void *ptrP, size_t sizeP, size_t nmembP, void *dataP) {
    cIptvProtocolCurl *obj = reinterpret_cast<cIptvProtocolCurl *>(dataP);
    size_t len = sizeP*nmembP;

    debug16("%s (, %zu, %zu, ) len=%zu", __PRETTY_FUNCTION__, sizeP, nmembP, len);

    if (obj && !obj->PutData((unsigned char *) ptrP, (int) len)) {
        return CURL_WRITEFUNC_PAUSE;
    }

    return len;
}

size_t cIptvProtocolCurl::WriteRtspCallback(void *ptrP, size_t sizeP, size_t nmembP, void *dataP) {
    cIptvProtocolCurl *obj = reinterpret_cast<cIptvProtocolCurl *>(dataP);
    size_t len = sizeP*nmembP;
    unsigned char *p = (unsigned char *) ptrP;

    debug16("%s (, %zu, %zu, ) len=%zu", __PRETTY_FUNCTION__, sizeP, nmembP, len);

    // Validate packet header ('$') and channel (0)
    if (obj && (p[0] == 0x24) && (p[1] == 0)) {
        int length = (p[2] << 8) | p[3];
        if (length > 3) {
            // Skip interleave header
            p += 4;
            // http://tools.ietf.org/html/rfc3550
            // http://tools.ietf.org/html/rfc2250
            // Version
            unsigned int v = (p[0] >> 6) & 0x03;
            // Extension bit
            unsigned int x = (p[0] >> 4) & 0x01;
            // CSCR count
            unsigned int cc = p[0] & 0x0F;
            // Payload type: MPEG2 TS = 33
            //unsigned int pt = p[1] & 0x7F;
            // Header lenght
            unsigned int headerlen = (3 + cc) * (unsigned int) sizeof(uint32_t);

            // Check if extension
            if (x) {
                // Extension header length
                unsigned int ehl = (((p[headerlen + 2] & 0xFF) << 8) | (p[headerlen + 3] & 0xFF));
                // Update header length
                headerlen += (ehl + 1) * (unsigned int) sizeof(uint32_t);
            }

            // Check that rtp is version 2 and payload contains multiple of TS packet data
            if ((v == 2) && (((length - headerlen) % TS_SIZE)==0) && (p[headerlen] == TS_SYNC_BYTE)) {
                // Set argument point to payload in read buffer
                obj->PutData(&p[headerlen], (length - headerlen));
            }
        }
    }

    return len;
}

size_t cIptvProtocolCurl::DescribeCallback(void *ptrP, size_t sizeP, size_t nmembP, void *dataP) {
    cIptvProtocolCurl *obj = reinterpret_cast<cIptvProtocolCurl *>(dataP);
    size_t len = sizeP*nmembP;

    debug16("%s (, %zu, %zu, ) len=%zu", __PRETTY_FUNCTION__, sizeP, nmembP, len);

    bool found = false;
    cString control = "";
    char *p = (char *) ptrP;
    char *r = strtok(p, "\r\n");

    while (r) {
        debug16("%s (, %zu, %zu, ) len=%zu r=%s", __PRETTY_FUNCTION__, sizeP, nmembP, len, r);

        // Look for a media name: "video"
        if (strstr(r, "m=video")) {
            found = true;
        }

        // ... and find out its attribute
        if (found && strstr(r, "a=control")) {
            char *s = nullptr;
            if (sscanf(r, "a=control:%255ms", &s)==1)
                control = compactspace(s);
            free(s);

            break;
        }

        r = strtok(nullptr, "\r\n");
    }

    if (!isempty(*control) && obj) {
        obj->SetRtspControl(*control);
    }

    return len;
}

size_t cIptvProtocolCurl::HeaderCallback(void *ptrP, size_t sizeP, size_t nmembP, void *dataP) {
    //cIptvProtocolCurl *obj = reinterpret_cast<cIptvProtocolCurl *>(dataP);
    size_t len = sizeP*nmembP;

    debug16("%s (, %zu, %zu, ) len=%zu", __PRETTY_FUNCTION__, sizeP, nmembP, len);

    char *p = (char *) ptrP;
    char *r = strtok(p, "\r\n");

    while (r) {
        debug16("%s (, %zu, %zu, ) len=%zu r=%s", __PRETTY_FUNCTION__, sizeP, nmembP, len, r);

        r = strtok(nullptr, "\r\n");
    }

    return len;
}

void cIptvProtocolCurl::SetRtspControl(const char *controlP) {
    cMutexLock MutexLock(&mutexM);

    debug16("%s (%s)", __PRETTY_FUNCTION__, controlP);

    cString protocol = ChangeCase(controlP, false).Truncate(7);

    if (startswith(*protocol, "rtsp://")) {
        streamUrlM = controlP;
        rtspControlM = "";
    } else {
        rtspControlM = controlP;
    }
}

bool cIptvProtocolCurl::PutData(unsigned char *dataP, int lenP) {
    cMutexLock MutexLock(&mutexM);

    debug16("%s (, %d)", __PRETTY_FUNCTION__, lenP);

    if (pausedM) {
        return false;
    }

    if (ringBufferM && (lenP >= 0)) {
        // Should we pause the transfer ?
        if (ringBufferM->Free() < (2*CURL_MAX_WRITE_SIZE)) {
            debug1("%s Pause free=%d available=%d len=%d", __PRETTY_FUNCTION__, ringBufferM->Free(), ringBufferM->Available(), lenP);

            pausedM = true;
            return false;
        }

        int p = ringBufferM->Put(dataP, lenP);
        if (p != lenP) {
            ringBufferM->ReportOverflow(lenP - p);
        }
    }

    return true;
}

void cIptvProtocolCurl::DelData(int lenP) {
    cMutexLock MutexLock(&mutexM);

    debug16("%s", __PRETTY_FUNCTION__);

    if (ringBufferM && (lenP >= 0)) {
        ringBufferM->Del(lenP);
    }
}

void cIptvProtocolCurl::ClearData() {
    debug16("%s", __PRETTY_FUNCTION__);

    if (ringBufferM) {
        ringBufferM->Clear();
    }
}

unsigned char *cIptvProtocolCurl::GetData(int &lenP) {
    cMutexLock MutexLock(&mutexM);

    debug16("%s", __PRETTY_FUNCTION__);

    unsigned char *p = nullptr;
    lenP = 0;

    if (ringBufferM) {
        int count = 0;
        p = ringBufferM->Get(count);
#if 0
        if (p && count >= TS_SIZE) {
           if (*p != TS_SYNC_BYTE) {
              for (int i = 1; i < count; ++i) {
                  if (p[i] == TS_SYNC_BYTE) {
                     count = i;
                     break;
                     }
                  }
              error("IPTV skipped %d bytes to sync on TS packet", count);
              ringBufferM->Del(count);
              lenP = 0;
              return NULL;
              }
           }
#endif
        count -= (count % TS_SIZE);
        lenP = count;
    }

    return p;
}

bool cIptvProtocolCurl::Connect() {
    cMutexLock MutexLock(&mutexM);

    debug1("%s", __PRETTY_FUNCTION__);

    if (connectedM) {
        return true;
    }

    // Initialize the curl session
    if (!handleM) {
        handleM = curl_easy_init();
    }

    if (handleM && !isempty(*streamUrlM)) {
        CURLcode res = CURLE_OK;
        cString netrc = cString::sprintf("%s/netrc", IptvConfig.GetConfigDirectory());

        // Verbose output
        iptv_curl_easy_setopt(handleM, CURLOPT_VERBOSE, 1L);
        iptv_curl_easy_setopt(handleM, CURLOPT_DEBUGFUNCTION, cIptvProtocolCurl::DebugCallback);
        iptv_curl_easy_setopt(handleM, CURLOPT_DEBUGDATA, this);

        // Set callbacks
        iptv_curl_easy_setopt(handleM, CURLOPT_WRITEFUNCTION, cIptvProtocolCurl::WriteCallback);
        iptv_curl_easy_setopt(handleM, CURLOPT_WRITEDATA, this);
        iptv_curl_easy_setopt(handleM, CURLOPT_HEADERFUNCTION, cIptvProtocolCurl::HeaderCallback);
        iptv_curl_easy_setopt(handleM, CURLOPT_WRITEHEADER, this);

        // No progress meter and no signaling
        iptv_curl_easy_setopt(handleM, CURLOPT_NOPROGRESS, 1L);
        iptv_curl_easy_setopt(handleM, CURLOPT_NOSIGNAL, 1L);

        // Support netrc
        iptv_curl_easy_setopt(handleM, CURLOPT_NETRC, (long) CURL_NETRC_OPTIONAL);
        iptv_curl_easy_setopt(handleM, CURLOPT_NETRC_FILE, *netrc);

        // Set timeouts
        iptv_curl_easy_setopt(handleM, CURLOPT_CONNECTTIMEOUT, (long) eConnectTimeoutS);
        iptv_curl_easy_setopt(handleM, CURLOPT_LOW_SPEED_LIMIT, (long) eLowSpeedLimitBytes);
        iptv_curl_easy_setopt(handleM, CURLOPT_LOW_SPEED_TIME, (long) eLowSpeedTimeoutS);

        // Set user-agent
        iptv_curl_easy_setopt(handleM, CURLOPT_USERAGENT, *cString::sprintf("vdr-%s/%s", PLUGIN_NAME_I18N, VERSION));

        // Set URL
        char *p = curl_easy_unescape(handleM, *streamUrlM, 0, nullptr);
        streamUrlM = p;
        curl_free(p);
        iptv_curl_easy_setopt(handleM, CURLOPT_URL, *streamUrlM);

        // Protocol specific initializations
        switch (modeM) {
#ifdef USE_RTSP
        case eModeRtsp: {
            cString uri, control, transport, range;

            // Create the listening socket for UDP mode
            if (!streamParamM) {
                OpenSocket(streamPortM);
            }

            // Request server options
            uri = cString::sprintf("%s", *streamUrlM);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_STREAM_URI, *uri);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_OPTIONS);
            iptv_curl_easy_perform(handleM);

            // Request session description - SDP is delivered in message body and not in the header!
            iptv_curl_easy_setopt(handleM, CURLOPT_WRITEFUNCTION, cIptvProtocolCurl::DescribeCallback);
            iptv_curl_easy_setopt(handleM, CURLOPT_WRITEDATA, this);
            uri = cString::sprintf("%s", *streamUrlM);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_STREAM_URI, *uri);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_DESCRIBE);
            iptv_curl_easy_perform(handleM);
            iptv_curl_easy_setopt(handleM, CURLOPT_WRITEFUNCTION, NULL);
            iptv_curl_easy_setopt(handleM, CURLOPT_WRITEDATA, NULL);

            // Setup media stream
            if (isempty(*rtspControlM)) {
                uri = cString::sprintf("%s", *streamUrlM);
            } else {
                uri = cString::sprintf("%s/%s", *streamUrlM, *rtspControlM);
            }

            if (streamParamM) {
                transport = "RTP/AVP/TCP;unicast;interleaved=0-1";
            } else {
                transport = cString::sprintf("RTP/AVP;unicast;client_port=%d-%d", streamPortM, streamPortM + 1);
            }

            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_STREAM_URI, *uri);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_TRANSPORT, *transport);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_SETUP);
            iptv_curl_easy_perform(handleM);

            // Start playing
            uri = cString::sprintf("%s/", *streamUrlM);
            range = "0.000-";
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_STREAM_URI, *uri);
            //iptv_curl_easy_setopt(handleM, CURLOPT_RANGE, *range);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_PLAY);
            iptv_curl_easy_perform(handleM);

            // Start receiving
            if (streamParamM) {
                iptv_curl_easy_setopt(handleM, CURLOPT_INTERLEAVEFUNCTION, cIptvProtocolCurl::WriteRtspCallback);
                iptv_curl_easy_setopt(handleM, CURLOPT_INTERLEAVEDATA, this);
                iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_RECEIVE);
                iptv_curl_easy_perform(handleM);
            }

            // Don't add handle into multi set
            isActiveM = true;
        }
            break;
#endif
        case eModeHttp:
        case eModeHttps: {
            // Limit download speed (bytes/s)
            iptv_curl_easy_setopt(handleM, CURLOPT_MAX_RECV_SPEED_LARGE, eMaxDownloadSpeedMBits*131072L);

            // Follow location
            iptv_curl_easy_setopt(handleM, CURLOPT_FOLLOWLOCATION, 1L);

            // Fail if HTTP return code is >= 400
            iptv_curl_easy_setopt(handleM, CURLOPT_FAILONERROR, 1L);

            // Set additional headers to prevent caching
            headerListM = curl_slist_append(headerListM, "Cache-Control: no-store, no-cache, must-revalidate");
            headerListM = curl_slist_append(headerListM, "Cache-Control: post-check=0, pre-check=0");
            headerListM = curl_slist_append(headerListM, "Pragma: no-cache");
            headerListM = curl_slist_append(headerListM, "Expires: Mon, 26 Jul 1997 05:00:00 GMT");
            iptv_curl_easy_setopt(handleM, CURLOPT_HTTPHEADER, headerListM);

            // Initialize multi set and add handle into it
            if (!multiM)
                multiM = curl_multi_init();
            if (multiM)
                curl_multi_add_handle(multiM, handleM);
        }
            break;

        case eModeFile: {
            // Set timeout
            iptv_curl_easy_setopt(handleM, CURLOPT_TIMEOUT_MS, 10L);

            // Initialize multi set and add handle into it
            if (!multiM) {
                multiM = curl_multi_init();
            }

            if (multiM) {
                curl_multi_add_handle(multiM, handleM);
            }
        }
            break;

        case eModeUnknown:
        default:
            break;
        }

        timeoutM.Set(eKeepAliveIntervalMs);
        connectedM = true;
        return true;
    }

    return false;
}

bool cIptvProtocolCurl::Disconnect() {
    cMutexLock MutexLock(&mutexM);
    debug1("%s", __PRETTY_FUNCTION__);

    if (!connectedM) {
        return true;
    }

    // Terminate curl session
    if (handleM) {
        // Remove handle from multi set
        if (multiM) {
            curl_multi_remove_handle(multiM, handleM);
            curl_multi_cleanup(multiM);
            multiM = nullptr;
        }

        // Mode specific tricks
        switch (modeM) {
#ifdef USE_RTSP
        case eModeRtsp: {
            CURLcode res = CURLE_OK;
            // Teardown rtsp session
            cString uri = cString::sprintf("%s/", *streamUrlM);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_STREAM_URI, *uri);
            iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_TEARDOWN);
            iptv_curl_easy_perform(handleM);
            rtspControlM = "";
            isActiveM = false;
            // Close the listening socket
            CloseSocket();
        }
            break;
#endif
        case eModeHttp:
        case eModeHttps:
        case eModeFile:
        case eModeUnknown:
        default:break;
        }

        // Cleanup curl stuff
        if (headerListM) {
            curl_slist_free_all(headerListM);
            headerListM = nullptr;
        }

        curl_easy_cleanup(handleM);
        handleM = nullptr;
    }

    ClearData();
    connectedM = false;

    return true;
}

bool cIptvProtocolCurl::Open() {
    debug1("%s", __PRETTY_FUNCTION__);

    return Connect();
}

bool cIptvProtocolCurl::Close() {
    debug1("%s", __PRETTY_FUNCTION__);

    Disconnect();

    return true;
}

int cIptvProtocolCurl::Read(unsigned char *bufferAddrP, unsigned int bufferLenP) {
    debug16("%s (, %u)", __PRETTY_FUNCTION__, bufferLenP);

    int len = 0;
    if (ringBufferM) {
        // Fill up the buffer
        if (handleM) {
            switch (modeM) {
#ifdef USE_RTSP
            case eModeRtsp: {
                cMutexLock MutexLock(&mutexM);
                CURLcode res = CURLE_OK;

                // Remember the heart beat
                if (timeoutM.TimedOut()) {
                    debug1("%s KeepAlive", __PRETTY_FUNCTION__);
                    cString uri = cString::sprintf("%s", *streamUrlM);
                    iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_STREAM_URI, *uri);
                    iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_OPTIONS);
                    iptv_curl_easy_perform(handleM);
                    timeoutM.Set(eKeepAliveIntervalMs);
                }

                // Check whether UDP or TCP mode used
                if (streamParamM) {
                    iptv_curl_easy_setopt(handleM, CURLOPT_RTSP_REQUEST, (long) CURL_RTSPREQ_RECEIVE);
                    iptv_curl_easy_perform(handleM);
                } else {
                    return cIptvUdpSocket::Read(bufferAddrP, bufferLenP);
                }
            }
                break;
#endif
            case eModeFile:
            case eModeHttp:
            case eModeHttps:
                if (multiM) {
                    CURLMcode res;
                    int running_handles;

                    do {
                        cMutexLock MutexLock(&mutexM);
                        res = curl_multi_perform(multiM, &running_handles);
                    } while (res==CURLM_CALL_MULTI_PERFORM);

                    // Use 20% threshold before continuing to filling up the buffer.
                    mutexM.Lock();
                    if (pausedM && (ringBufferM->Available() < (IPTV_BUFFER_SIZE/5))) {
                        debug1("%s Continue free=%d available=%d", __PRETTY_FUNCTION__, ringBufferM->Free(), ringBufferM->Available());

                        pausedM = false;
                        curl_easy_pause(handleM, CURLPAUSE_CONT);
                    }
                    mutexM.Unlock();

                    // Check if end of file
                    if (running_handles==0) {
                        int msgcount;
                        mutexM.Lock();
                        CURLMsg *msg = curl_multi_info_read(multiM, &msgcount);
                        mutexM.Unlock();
                        if (msg && (msg->msg==CURLMSG_DONE)) {
                            debug1("%s Done %s (%d)", __PRETTY_FUNCTION__, curl_easy_strerror(msg->data.result), msg->data.result);
                            Disconnect();
                            Connect();
                        }
                    }
                }
                break;

            case eModeUnknown:
            default:
                break;
            }
        }

        // ... and try to empty it
        unsigned char *p = GetData(len);
        if (p && (len > 0)) {
            len = min(len, (int) bufferLenP);
            memcpy(bufferAddrP, p, len);
            DelData(len);

            debug16("%s Get %d bytes", __PRETTY_FUNCTION__, len);
        }
    }

    return len;
}

bool
cIptvProtocolCurl::SetSource(SourceParameter  parameter) {
    debug1("%s (%s, %d, %d)", __PRETTY_FUNCTION__, parameter.locationP, parameter.parameterP, parameter.indexP);

    if (!isempty(parameter.locationP)) {
        // Disconnect
        Disconnect();
        // Update stream URL
        streamUrlM = parameter.locationP;
        cString protocol = ChangeCase(streamUrlM, false).Truncate(5);

        if (startswith(*protocol, "rtsp")) {
            modeM = eModeRtsp;
        } else if (startswith(*protocol, "https")) {
            modeM = eModeHttps;
        } else if (startswith(*protocol, "http")) {
            modeM = eModeHttp;
        } else if (startswith(*protocol, "file")) {
            modeM = eModeFile;
        } else {
            modeM = eModeUnknown;
        }

        debug1("%s (%s, %d, %d) protocol=%s mode=%d", __PRETTY_FUNCTION__, parameter.locationP, parameter.parameterP, parameter.indexP, *protocol, modeM);

        // Update stream parameter - force UDP mode for RTSP
        streamParamM = (modeM==eModeRtsp) ? 0 : parameter.parameterP;

        // Update listen port
        streamPortM = IptvConfig.GetProtocolBasePort() + parameter.indexP*2;

        // Reconnect
        Connect();
    }
    return true;
}

bool cIptvProtocolCurl::SetPid(int pidP, int typeP, bool onP) {
    debug16("%s (%d, %d, %d)", __PRETTY_FUNCTION__, pidP, typeP, onP);

    return true;
}

cString cIptvProtocolCurl::GetInformation() {
    debug16("%s", __PRETTY_FUNCTION__);

    return cString::sprintf("%s [%d]", *streamUrlM, streamParamM);
}