rtmpsuck.c

/*  RTMP Proxy Server
 *  Copyright (C) 2009 Andrej Stepanchuk
 *  Copyright (C) 2009 Howard Chu
 *
 *  This Program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2, or (at your option)
 *  any later version.
 *
 *  This Program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with RTMPDump; see the file COPYING.  If not, write to
 *  the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
 *  Boston, MA  02110-1301, USA.
 *  http://www.gnu.org/copyleft/gpl.html
 *
 */

/* This is a Proxy Server that displays the connection parameters from a
 * client and then saves any data streamed to the client.
 */

#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <limits.h>

#include <signal.h>
#include <getopt.h>

#include <assert.h>

#include "librtmp/rtmp_sys.h"
#include "librtmp/log.h"

#include "thread.h"

#ifdef linux
#include <linux/netfilter_ipv4.h>
#endif

#define RD_SUCCESS 0
#define RD_FAILED 1
#define RD_INCOMPLETE 2

#define PACKET_SIZE 1024 * 1024

#ifdef WIN32
#define InitSockets()              \
  {                                \
    WORD version;                  \
    WSADATA wsaData;               \
                                   \
    version = MAKEWORD(1, 1);      \
    WSAStartup(version, &wsaData); \
  }

#define CleanupSockets() WSACleanup()
#else
#define InitSockets()
#define CleanupSockets()
#endif

enum
{
  STREAMING_ACCEPTING,
  STREAMING_IN_PROGRESS,
  STREAMING_STOPPING,
  STREAMING_STOPPED
};

typedef struct Flist
{
  struct Flist *f_next;
  FILE *f_file;
  AVal f_path;
} Flist;

typedef struct Plist
{
  struct Plist *p_next;
  RTMPPacket p_pkt;
} Plist;

typedef struct
{
  int socket;
  int state;
  uint32_t stamp;
  RTMP rs;
  RTMP rc;
  Plist *rs_pkt[2]; /* head, tail */
  Plist *rc_pkt[2]; /* head, tail */
  Flist *f_head, *f_tail;
  Flist *f_cur;

} STREAMING_SERVER;

STREAMING_SERVER *rtmpServer = 0; // server structure pointer

STREAMING_SERVER *startStreaming(const char *address, int port);
void stopStreaming(STREAMING_SERVER *server);

#define STR2AVAL(av, str) \
  av.av_val = str;        \
  av.av_len = strlen(av.av_val)

#ifdef _DEBUG
uint32_t debugTS = 0;

int pnum = 0;

FILE *netstackdump = NULL;
FILE *netstackdump_read = NULL;
#endif

#define BUFFERTIME (4 * 60 * 60 * 1000) /* 4 hours */

#define SAVC(x) static const AVal av_##x = AVC(#x)

SAVC(app);
SAVC(connect);
SAVC(flashVer);
SAVC(swfUrl);
SAVC(pageUrl);
SAVC(tcUrl);
SAVC(fpad);
SAVC(capabilities);
SAVC(audioCodecs);
SAVC(videoCodecs);
SAVC(videoFunction);
SAVC(objectEncoding);
SAVC(_result);
SAVC(createStream);
SAVC(play);
SAVC(closeStream);
SAVC(fmsVer);
SAVC(mode);
SAVC(level);
SAVC(code);
SAVC(secureToken);
SAVC(onStatus);
SAVC(close);
static const AVal av_NetStream_Failed = AVC("NetStream.Failed");
static const AVal av_NetStream_Play_Failed = AVC("NetStream.Play.Failed");
static const AVal av_NetStream_Play_StreamNotFound =
    AVC("NetStream.Play.StreamNotFound");
static const AVal av_NetConnection_Connect_InvalidApp =
    AVC("NetConnection.Connect.InvalidApp");
static const AVal av_NetStream_Play_Start = AVC("NetStream.Play.Start");
static const AVal av_NetStream_Play_Complete = AVC("NetStream.Play.Complete");
static const AVal av_NetStream_Play_Stop = AVC("NetStream.Play.Stop");

static const char *cst[] = {"client", "server"};

// Returns 0 for OK/Failed/error, 1 for 'Stop or Complete'
int ServeInvoke(STREAMING_SERVER *server, int which, RTMPPacket *pack, const char *body)
{
  int ret = 0, nRes;
  int nBodySize = pack->m_nBodySize;

  if (body > pack->m_body)
    nBodySize--;

  if (body[0] != 0x02) // make sure it is a string method name we start with
  {
    RTMP_Log(RTMP_LOGWARNING, "%s, Sanity failed. no string method in invoke packet",
             __FUNCTION__);
    return 0;
  }

  AMFObject obj;
  nRes = AMF_Decode(&obj, body, nBodySize, FALSE);
  if (nRes < 0)
  {
    RTMP_Log(RTMP_LOGERROR, "%s, error decoding invoke packet", __FUNCTION__);
    return 0;
  }

  AMF_Dump(&obj);
  AVal method;
  AMFProp_GetString(AMF_GetProp(&obj, NULL, 0), &method);
  RTMP_Log(RTMP_LOGDEBUG, "%s, %s invoking <%s>", __FUNCTION__, cst[which], method.av_val);

  if (AVMATCH(&method, &av_connect))
  {
    AMFObject cobj;
    AVal pname, pval;
    int i;
    AMFProp_GetObject(AMF_GetProp(&obj, NULL, 2), &cobj);
    RTMP_LogPrintf("Processing connect\n");
    for (i = 0; i < cobj.o_num; i++)
    {
      pname = cobj.o_props[i].p_name;
      pval.av_val = NULL;
      pval.av_len = 0;
      if (cobj.o_props[i].p_type == AMF_STRING)
      {
        pval = cobj.o_props[i].p_vu.p_aval;
        RTMP_LogPrintf("%.*s: %.*s\n", pname.av_len, pname.av_val, pval.av_len, pval.av_val);
      }
      if (AVMATCH(&pname, &av_app))
      {
        server->rc.Link.app = pval;
        pval.av_val = NULL;
      }
      else if (AVMATCH(&pname, &av_flashVer))
      {
        server->rc.Link.flashVer = pval;
        pval.av_val = NULL;
      }
      else if (AVMATCH(&pname, &av_swfUrl))
      {
#ifdef CRYPTO
        if (pval.av_val)
          RTMP_HashSWF(pval.av_val, &server->rc.Link.SWFSize,
                       (unsigned char *)server->rc.Link.SWFHash, 30);
#endif
        server->rc.Link.swfUrl = pval;
        pval.av_val = NULL;
      }
      else if (AVMATCH(&pname, &av_tcUrl))
      {
        char *r1 = NULL, *r2;
        int len;

        server->rc.Link.tcUrl = pval;
        if ((pval.av_val[0] | 0x40) == 'r' &&
            (pval.av_val[1] | 0x40) == 't' &&
            (pval.av_val[2] | 0x40) == 'm' &&
            (pval.av_val[3] | 0x40) == 'p')
        {
          if (pval.av_val[4] == ':')
          {
            server->rc.Link.protocol = RTMP_PROTOCOL_RTMP;
            r1 = pval.av_val + 7;
          }
          else if ((pval.av_val[4] | 0x40) == 'e' && pval.av_val[5] == ':')
          {
            server->rc.Link.protocol = RTMP_PROTOCOL_RTMPE;
            r1 = pval.av_val + 8;
          }
          r2 = strchr(r1, '/');
          if (r2)
            len = r2 - r1;
          else
            len = pval.av_len - (r1 - pval.av_val);
          r2 = malloc(len + 1);
          memcpy(r2, r1, len);
          r2[len] = '\0';
          server->rc.Link.hostname.av_val = r2;
          r1 = strrchr(r2, ':');
          if (r1)
          {
            server->rc.Link.hostname.av_len = r1 - r2;
            *r1++ = '\0';
            server->rc.Link.port = atoi(r1);
          }
          else
          {
            server->rc.Link.hostname.av_len = len;
            server->rc.Link.port = 1935;
          }
        }
        pval.av_val = NULL;
      }
      else if (AVMATCH(&pname, &av_pageUrl))
      {
        server->rc.Link.pageUrl = pval;
        pval.av_val = NULL;
      }
      else if (AVMATCH(&pname, &av_audioCodecs))
      {
        server->rc.m_fAudioCodecs = cobj.o_props[i].p_vu.p_number;
      }
      else if (AVMATCH(&pname, &av_videoCodecs))
      {
        server->rc.m_fVideoCodecs = cobj.o_props[i].p_vu.p_number;
      }
      else if (AVMATCH(&pname, &av_objectEncoding))
      {
        server->rc.m_fEncoding = cobj.o_props[i].p_vu.p_number;
        server->rc.m_bSendEncoding = TRUE;
      }
      /* Dup'd a string we didn't recognize? */
      if (pval.av_val)
        free(pval.av_val);
    }
    if (obj.o_num > 3)
    {
      if (AMFProp_GetBoolean(&obj.o_props[3]))
        server->rc.Link.lFlags |= RTMP_LF_AUTH;
      if (obj.o_num > 4)
      {
        AMFProp_GetString(&obj.o_props[4], &server->rc.Link.auth);
      }
    }

    if (!RTMP_Connect(&server->rc, pack))
    {
      /* failed */
      return 1;
    }
    server->rc.m_bSendCounter = FALSE;
  }
  else if (AVMATCH(&method, &av_play))
  {
    Flist *fl;
    AVal av;
    FILE *out;
    char *file, *p, *q;
    char flvHeader[] = {
        'F', 'L', 'V', 0x01,
        0x05, // video + audio, we finalize later if the value is different
        0x00, 0x00, 0x00, 0x09,
        0x00, 0x00, 0x00, 0x00 // first prevTagSize=0
    };
    int count = 0, flen;

    server->rc.m_stream_id = pack->m_nInfoField2;
    AMFProp_GetString(AMF_GetProp(&obj, NULL, 3), &av);
    server->rc.Link.playpath = av;
    if (!av.av_val)
      goto out;

    /* check for duplicates */
    for (fl = server->f_head; fl; fl = fl->f_next)
    {
      if (AVMATCH(&av, &fl->f_path))
        count++;
    }
    /* strip trailing URL parameters */
    q = memchr(av.av_val, '?', av.av_len);
    if (q)
    {
      if (q == av.av_val)
      {
        av.av_val++;
        av.av_len--;
      }
      else
      {
        av.av_len = q - av.av_val;
      }
    }
    /* strip leading slash components */
    for (p = av.av_val + av.av_len - 1; p >= av.av_val; p--)
      if (*p == '/')
      {
        p++;
        av.av_len -= p - av.av_val;
        av.av_val = p;
        break;
      }
    /* skip leading dot */
    if (av.av_val[0] == '.')
    {
      av.av_val++;
      av.av_len--;
    }
    flen = av.av_len;
    /* hope there aren't more than 255 dups */
    if (count)
      flen += 2;
    file = malloc(flen + 1);

    memcpy(file, av.av_val, av.av_len);
    if (count)
      sprintf(file + av.av_len, "%02x", count);
    else
      file[av.av_len] = '\0';
    for (p = file; *p; p++)
      if (*p == ':')
        *p = '_';
    RTMP_LogPrintf("Playpath: %.*s\nSaving as: %s\n",
                   server->rc.Link.playpath.av_len, server->rc.Link.playpath.av_val,
                   file);
    out = fopen(file, "wb");
    free(file);
    if (!out)
      ret = 1;
    else
    {
      fwrite(flvHeader, 1, sizeof(flvHeader), out);
      av = server->rc.Link.playpath;
      fl = malloc(sizeof(Flist) + av.av_len + 1);
      fl->f_file = out;
      fl->f_path.av_len = av.av_len;
      fl->f_path.av_val = (char *)(fl + 1);
      memcpy(fl->f_path.av_val, av.av_val, av.av_len);
      fl->f_path.av_val[av.av_len] = '\0';
      fl->f_next = NULL;
      if (server->f_tail)
        server->f_tail->f_next = fl;
      else
        server->f_head = fl;
      server->f_tail = fl;
    }
  }
  else if (AVMATCH(&method, &av_onStatus))
  {
    AMFObject obj2;
    AVal code, level;
    AMFProp_GetObject(AMF_GetProp(&obj, NULL, 3), &obj2);
    AMFProp_GetString(AMF_GetProp(&obj2, &av_code, -1), &code);
    AMFProp_GetString(AMF_GetProp(&obj2, &av_level, -1), &level);

    RTMP_Log(RTMP_LOGDEBUG, "%s, onStatus: %s", __FUNCTION__, code.av_val);
    if (AVMATCH(&code, &av_NetStream_Failed) || AVMATCH(&code, &av_NetStream_Play_Failed) || AVMATCH(&code, &av_NetStream_Play_StreamNotFound) || AVMATCH(&code, &av_NetConnection_Connect_InvalidApp))
    {
      ret = 1;
    }

    if (AVMATCH(&code, &av_NetStream_Play_Start))
    {
      /* set up the next stream */
      if (server->f_cur)
      {
        if (server->f_cur->f_next)
          server->f_cur = server->f_cur->f_next;
      }
      else
      {
        for (server->f_cur = server->f_head; server->f_cur &&
                                             !server->f_cur->f_file;
             server->f_cur = server->f_cur->f_next)
          ;
      }
      server->rc.m_bPlaying = TRUE;
    }

    // Return 1 if this is a Play.Complete or Play.Stop
    if (AVMATCH(&code, &av_NetStream_Play_Complete) || AVMATCH(&code, &av_NetStream_Play_Stop))
    {
      ret = 1;
    }
  }
  else if (AVMATCH(&method, &av_closeStream))
  {
    ret = 1;
  }
  else if (AVMATCH(&method, &av_close))
  {
    RTMP_Close(&server->rc);
    ret = 1;
  }
out:
  AMF_Reset(&obj);
  return ret;
}

int ServePacket(STREAMING_SERVER *server, int which, RTMPPacket *packet)
{
  int ret = 0;

  RTMP_Log(RTMP_LOGDEBUG, "%s, %s sent packet type %02X, size %u bytes", __FUNCTION__,
           cst[which], packet->m_packetType, packet->m_nBodySize);

  switch (packet->m_packetType)
  {
  case RTMP_PACKET_TYPE_CHUNK_SIZE:
    // chunk size
    //      HandleChangeChunkSize(r, packet);
    break;

  case RTMP_PACKET_TYPE_BYTES_READ_REPORT:
    // bytes read report
    break;

  case RTMP_PACKET_TYPE_CONTROL:
    // ctrl
    //      HandleCtrl(r, packet);
    break;

  case RTMP_PACKET_TYPE_SERVER_BW:
    // server bw
    //      HandleServerBW(r, packet);
    break;

  case RTMP_PACKET_TYPE_CLIENT_BW:
    // client bw
    //     HandleClientBW(r, packet);
    break;

  case RTMP_PACKET_TYPE_AUDIO:
    // audio data
    //RTMP_Log(RTMP_LOGDEBUG, "%s, received: audio %lu bytes", __FUNCTION__, packet.m_nBodySize);
    break;

  case RTMP_PACKET_TYPE_VIDEO:
    // video data
    //RTMP_Log(RTMP_LOGDEBUG, "%s, received: video %lu bytes", __FUNCTION__, packet.m_nBodySize);
    break;

  case RTMP_PACKET_TYPE_FLEX_STREAM_SEND:
    // flex stream send
    break;

  case RTMP_PACKET_TYPE_FLEX_SHARED_OBJECT:
    // flex shared object
    break;

  case RTMP_PACKET_TYPE_FLEX_MESSAGE:
    // flex message
    {
      ret = ServeInvoke(server, which, packet, packet->m_body + 1);
      break;
    }
  case RTMP_PACKET_TYPE_INFO:
    // metadata (notify)
    break;

  case RTMP_PACKET_TYPE_SHARED_OBJECT:
    /* shared object */
    break;

  case RTMP_PACKET_TYPE_INVOKE:
    // invoke
    ret = ServeInvoke(server, which, packet, packet->m_body);
    break;

  case RTMP_PACKET_TYPE_FLASH_VIDEO:
    /* flv */
    break;

  default:
    RTMP_Log(RTMP_LOGDEBUG, "%s, unknown packet type received: 0x%02x", __FUNCTION__,
             packet->m_packetType);
#ifdef _DEBUG
    RTMP_LogHex(RTMP_LOGDEBUG, packet->m_body, packet->m_nBodySize);
#endif
  }
  return ret;
}

int WriteStream(char **buf,         // target pointer, maybe preallocated
                unsigned int *plen, // length of buffer if preallocated
                uint32_t *nTimeStamp,
                RTMPPacket *packet)
{
  uint32_t prevTagSize = 0;
  int ret = -1, len = *plen;

  while (1)
  {
    char *packetBody = packet->m_body;
    unsigned int nPacketLen = packet->m_nBodySize;

    // skip video info/command packets
    if (packet->m_packetType == RTMP_PACKET_TYPE_VIDEO &&
        nPacketLen == 2 && ((*packetBody & 0xf0) == 0x50))
    {
      ret = 0;
      break;
    }

    if (packet->m_packetType == RTMP_PACKET_TYPE_VIDEO && nPacketLen <= 5)
    {
      RTMP_Log(RTMP_LOGWARNING, "ignoring too small video packet: size: %d",
               nPacketLen);
      ret = 0;
      break;
    }
    if (packet->m_packetType == RTMP_PACKET_TYPE_AUDIO && nPacketLen <= 1)
    {
      RTMP_Log(RTMP_LOGWARNING, "ignoring too small audio packet: size: %d",
               nPacketLen);
      ret = 0;
      break;
    }
#ifdef _DEBUG
    RTMP_Log(RTMP_LOGDEBUG, "type: %02X, size: %d, TS: %d ms", packet->m_packetType,
             nPacketLen, packet->m_nTimeStamp);
    if (packet->m_packetType == RTMP_PACKET_TYPE_VIDEO)
      RTMP_Log(RTMP_LOGDEBUG, "frametype: %02X", (*packetBody & 0xf0));
#endif

    // calculate packet size and reallocate buffer if necessary
    unsigned int size = nPacketLen +
                        ((packet->m_packetType == RTMP_PACKET_TYPE_AUDIO || packet->m_packetType == RTMP_PACKET_TYPE_VIDEO || packet->m_packetType == RTMP_PACKET_TYPE_INFO) ? 11 : 0) + (packet->m_packetType != 0x16 ? 4 : 0);

    if (size + 4 > len)
    {
      /* The extra 4 is for the case of an FLV stream without a last
           * prevTagSize (we need extra 4 bytes to append it).  */
      *buf = (char *)realloc(*buf, size + 4);
      if (*buf == 0)
      {
        RTMP_Log(RTMP_LOGERROR, "Couldn't reallocate memory!");
        ret = -1; // fatal error
        break;
      }
    }
    char *ptr = *buf, *pend = ptr + size + 4;

    /* audio (RTMP_PACKET_TYPE_AUDIO), video (RTMP_PACKET_TYPE_VIDEO)
       * or metadata (RTMP_PACKET_TYPE_INFO) packets: construct 11 byte
       * header then add rtmp packet's data.  */
    if (packet->m_packetType == RTMP_PACKET_TYPE_AUDIO || packet->m_packetType == RTMP_PACKET_TYPE_VIDEO || packet->m_packetType == RTMP_PACKET_TYPE_INFO)
    {
      // set data type
      //*dataType |= (((packet->m_packetType == RTMP_PACKET_TYPE_AUDIO)<<2)|(packet->m_packetType == RTMP_PACKET_TYPE_VIDEO));

      (*nTimeStamp) = packet->m_nTimeStamp;
      prevTagSize = 11 + nPacketLen;

      *ptr++ = packet->m_packetType;
      ptr = AMF_EncodeInt24(ptr, pend, nPacketLen);
      ptr = AMF_EncodeInt24(ptr, pend, *nTimeStamp);
      *ptr = (char)(((*nTimeStamp) & 0xFF000000) >> 24);
      ptr++;

      // stream id
      ptr = AMF_EncodeInt24(ptr, pend, 0);
    }

    memcpy(ptr, packetBody, nPacketLen);
    unsigned int len = nPacketLen;

    // correct tagSize and obtain timestamp if we have an FLV stream
    if (packet->m_packetType == RTMP_PACKET_TYPE_FLASH_VIDEO)
    {
      unsigned int pos = 0;

      while (pos + 11 < nPacketLen)
      {
        uint32_t dataSize = AMF_DecodeInt24(packetBody + pos + 1); // size without header (11) and without prevTagSize (4)
        *nTimeStamp = AMF_DecodeInt24(packetBody + pos + 4);
        *nTimeStamp |= (packetBody[pos + 7] << 24);

#if 0
	      /* set data type */
	      *dataType |= (((*(packetBody+pos) == RTMP_PACKET_TYPE_AUDIO) << 2)
                            | (*(packetBody+pos) == RTMP_PACKET_TYPE_VIDEO));
#endif

        if (pos + 11 + dataSize + 4 > nPacketLen)
        {
          if (pos + 11 + dataSize > nPacketLen)
          {
            RTMP_Log(RTMP_LOGERROR,
                     "Wrong data size (%u), stream corrupted, aborting!",
                     dataSize);
            ret = -2;
            break;
          }
          RTMP_Log(RTMP_LOGWARNING, "No tagSize found, appending!");

          // we have to append a last tagSize!
          prevTagSize = dataSize + 11;
          AMF_EncodeInt32(ptr + pos + 11 + dataSize, pend, prevTagSize);
          size += 4;
          len += 4;
        }
        else
        {
          prevTagSize =
              AMF_DecodeInt32(packetBody + pos + 11 + dataSize);

#ifdef _DEBUG
          RTMP_Log(RTMP_LOGDEBUG,
                   "FLV Packet: type %02X, dataSize: %lu, tagSize: %lu, timeStamp: %lu ms",
                   (unsigned char)packetBody[pos], dataSize, prevTagSize,
                   *nTimeStamp);
#endif

          if (prevTagSize != (dataSize + 11))
          {
#ifdef _DEBUG
            RTMP_Log(RTMP_LOGWARNING,
                     "Tag and data size are not consitent, writing tag size according to dataSize+11: %d",
                     dataSize + 11);
#endif

            prevTagSize = dataSize + 11;
            AMF_EncodeInt32(ptr + pos + 11 + dataSize, pend, prevTagSize);
          }
        }

        pos += prevTagSize + 4; //(11+dataSize+4);
      }
    }
    ptr += len;

    if (packet->m_packetType != RTMP_PACKET_TYPE_FLASH_VIDEO)
    { // FLV tag packets contain their own prevTagSize
      AMF_EncodeInt32(ptr, pend, prevTagSize);
      //ptr += 4;
    }

    ret = size;
    break;
  }

  if (len > *plen)
    *plen = len;

  return ret; // no more media packets
}

TFTYPE
controlServerThread(void *unused)
{
  char ich;
  while (1)
  {
    ich = getchar();
    switch (ich)
    {
    case 'q':
      RTMP_LogPrintf("Exiting\n");
      stopStreaming(rtmpServer);
      free(rtmpServer);
      exit(0);
      break;
    default:
      RTMP_LogPrintf("Unknown command \'%c\', ignoring\n", ich);
    }
  }
  TFRET();
}

TFTYPE doServe(void *arg) // server socket and state (our listening socket)
{
  STREAMING_SERVER *server = arg;
  RTMPPacket pc = {0}, ps = {0};
  RTMPChunk rk = {0};
  char *buf = NULL;
  unsigned int buflen = 131072;
  int paused = FALSE;
  int sockfd = server->socket;

  // timeout for http requests
  fd_set rfds;
  struct timeval tv;

  server->state = STREAMING_IN_PROGRESS;

  memset(&tv, 0, sizeof(struct timeval));
  tv.tv_sec = 5;

  FD_ZERO(&rfds);
  FD_SET(sockfd, &rfds);

  if (select(sockfd + 1, &rfds, NULL, NULL, &tv) <= 0)
  {
    RTMP_Log(RTMP_LOGERROR, "Request timeout/select failed, ignoring request");
    goto quit;
  }
  else
  {
    RTMP_Init(&server->rs);
    RTMP_Init(&server->rc);
    server->rs.m_sb.sb_socket = sockfd;
    if (!RTMP_Serve(&server->rs))
    {
      RTMP_Log(RTMP_LOGERROR, "Handshake failed");
      goto cleanup;
    }
  }

  buf = malloc(buflen);

  /* Just process the Connect request */
  while (RTMP_IsConnected(&server->rs) && RTMP_ReadPacket(&server->rs, &ps))
  {
    if (!RTMPPacket_IsReady(&ps))
      continue;
    ServePacket(server, 0, &ps);
    RTMPPacket_Free(&ps);
    if (RTMP_IsConnected(&server->rc))
      break;
  }

  pc.m_chunk = &rk;

  /* We have our own timeout in select() */
  server->rc.Link.timeout = 10;
  server->rs.Link.timeout = 10;
  while (RTMP_IsConnected(&server->rs) || RTMP_IsConnected(&server->rc))
  {
    int n;
    int sr, cr;

    cr = server->rc.m_sb.sb_size;
    sr = server->rs.m_sb.sb_size;

    if (cr || sr)
    {
    }
    else
    {
      n = server->rs.m_sb.sb_socket;
      if (server->rc.m_sb.sb_socket > n)
        n = server->rc.m_sb.sb_socket;
      FD_ZERO(&rfds);
      if (RTMP_IsConnected(&server->rs))
        FD_SET(sockfd, &rfds);
      if (RTMP_IsConnected(&server->rc))
        FD_SET(server->rc.m_sb.sb_socket, &rfds);

      /* give more time to start up if we're not playing yet */
      tv.tv_sec = server->f_cur ? 30 : 60;
      tv.tv_usec = 0;

      if (select(n + 1, &rfds, NULL, NULL, &tv) <= 0)
      {
        if (server->f_cur && server->rc.m_mediaChannel && !paused)
        {
          server->rc.m_pauseStamp = server->rc.m_channelTimestamp[server->rc.m_mediaChannel];
          if (RTMP_ToggleStream(&server->rc))
          {
            paused = TRUE;
            continue;
          }
        }
        RTMP_Log(RTMP_LOGERROR, "Request timeout/select failed, ignoring request");
        goto cleanup;
      }
      if (server->rs.m_sb.sb_socket > 0 &&
          FD_ISSET(server->rs.m_sb.sb_socket, &rfds))
        sr = 1;
      if (server->rc.m_sb.sb_socket > 0 &&
          FD_ISSET(server->rc.m_sb.sb_socket, &rfds))
        cr = 1;
    }
    if (sr)
    {
      while (RTMP_ReadPacket(&server->rs, &ps))
        if (RTMPPacket_IsReady(&ps))
        {
          /* change chunk size */
          if (ps.m_packetType == RTMP_PACKET_TYPE_CHUNK_SIZE)
          {
            if (ps.m_nBodySize >= 4)
            {
              server->rs.m_inChunkSize = AMF_DecodeInt32(ps.m_body);
              RTMP_Log(RTMP_LOGDEBUG, "%s, client: chunk size change to %d", __FUNCTION__,
                       server->rs.m_inChunkSize);
              server->rc.m_outChunkSize = server->rs.m_inChunkSize;
            }
          }
          /* bytes received */
          else if (ps.m_packetType == RTMP_PACKET_TYPE_BYTES_READ_REPORT)
          {
            if (ps.m_nBodySize >= 4)
            {
              int count = AMF_DecodeInt32(ps.m_body);
              RTMP_Log(RTMP_LOGDEBUG, "%s, client: bytes received = %d", __FUNCTION__,
                       count);
            }
          }
          /* ctrl */
          else if (ps.m_packetType == RTMP_PACKET_TYPE_CONTROL)
          {
            short nType = AMF_DecodeInt16(ps.m_body);
            /* UpdateBufferMS */
            if (nType == 0x03)
            {
              char *ptr = ps.m_body + 2;
              int id;
              int len;
              id = AMF_DecodeInt32(ptr);
              /* Assume the interesting media is on a non-zero stream */
              if (id)
              {
                len = AMF_DecodeInt32(ptr + 4);
#if 1
                /* request a big buffer */
                if (len < BUFFERTIME)
                {
                  AMF_EncodeInt32(ptr + 4, ptr + 8, BUFFERTIME);
                }
#endif
                RTMP_Log(RTMP_LOGDEBUG, "%s, client: BufferTime change in stream %d to %d", __FUNCTION__,
                         id, len);
              }
            }
          }
          else if (ps.m_packetType == RTMP_PACKET_TYPE_FLEX_MESSAGE || ps.m_packetType == RTMP_PACKET_TYPE_INVOKE)
          {
            if (ServePacket(server, 0, &ps) && server->f_cur)
            {
              fclose(server->f_cur->f_file);
              server->f_cur->f_file = NULL;
              server->f_cur = NULL;
            }
          }
          RTMP_SendPacket(&server->rc, &ps, FALSE);
          RTMPPacket_Free(&ps);
          break;
        }
    }
    if (cr)
    {
      while (RTMP_ReadPacket(&server->rc, &pc))
      {
        int sendit = 1;
        if (RTMPPacket_IsReady(&pc))
        {
          if (paused)
          {
            if (pc.m_nTimeStamp <= server->rc.m_mediaStamp)
              continue;
            paused = 0;
            server->rc.m_pausing = 0;
          }
          /* change chunk size */
          if (pc.m_packetType == RTMP_PACKET_TYPE_CHUNK_SIZE)
          {
            if (pc.m_nBodySize >= 4)
            {
              server->rc.m_inChunkSize = AMF_DecodeInt32(pc.m_body);
              RTMP_Log(RTMP_LOGDEBUG, "%s, server: chunk size change to %d", __FUNCTION__,
                       server->rc.m_inChunkSize);
              server->rs.m_outChunkSize = server->rc.m_inChunkSize;
            }
          }
          else if (pc.m_packetType == RTMP_PACKET_TYPE_CONTROL)
          {
            short nType = AMF_DecodeInt16(pc.m_body);
            /* SWFverification */
            if (nType == 0x1a)
#ifdef CRYPTO
              if (server->rc.Link.SWFSize)
              {
                RTMP_SendCtrl(&server->rc, 0x1b, 0, 0);
                sendit = 0;
              }
#else
              /* The session will certainly fail right after this */
              RTMP_Log(RTMP_LOGERROR, "%s, server requested SWF verification, need CRYPTO support! ", __FUNCTION__);
#endif
          }
          else if (server->f_cur && (pc.m_packetType == RTMP_PACKET_TYPE_AUDIO || pc.m_packetType == RTMP_PACKET_TYPE_VIDEO || pc.m_packetType == RTMP_PACKET_TYPE_INFO || pc.m_packetType == RTMP_PACKET_TYPE_FLASH_VIDEO) &&
                   RTMP_ClientPacket(&server->rc, &pc))
          {
            int len = WriteStream(&buf, &buflen, &server->stamp, &pc);
            if (len > 0 && fwrite(buf, 1, len, server->f_cur->f_file) != len)
              goto cleanup;
          }
          else if (pc.m_packetType == RTMP_PACKET_TYPE_FLEX_MESSAGE ||
                   pc.m_packetType == RTMP_PACKET_TYPE_INVOKE)
          {
            if (ServePacket(server, 1, &pc) && server->f_cur)
            {
              fclose(server->f_cur->f_file);
              server->f_cur->f_file = NULL;
              server->f_cur = NULL;
            }
          }
        }
        if (sendit && RTMP_IsConnected(&server->rs))
          RTMP_SendChunk(&server->rs, &rk);
        if (RTMPPacket_IsReady(&pc))
          RTMPPacket_Free(&pc);
        break;
      }
    }
    if (!RTMP_IsConnected(&server->rs) && RTMP_IsConnected(&server->rc) && !server->f_cur)
      RTMP_Close(&server->rc);
  }

cleanup:
  RTMP_LogPrintf("Closing connection... ");
  RTMP_Close(&server->rs);
  RTMP_Close(&server->rc);
  while (server->f_head)
  {
    Flist *fl = server->f_head;
    server->f_head = fl->f_next;
    if (fl->f_file)
      fclose(fl->f_file);
    free(fl);
  }
  server->f_tail = NULL;
  server->f_cur = NULL;
  free(buf);
  /* Should probably be done by RTMP_Close() ... */
  server->rc.Link.hostname.av_val = NULL;
  server->rc.Link.tcUrl.av_val = NULL;
  server->rc.Link.swfUrl.av_val = NULL;
  server->rc.Link.pageUrl.av_val = NULL;
  server->rc.Link.app.av_val = NULL;
  server->rc.Link.auth.av_val = NULL;
  server->rc.Link.flashVer.av_val = NULL;
  RTMP_LogPrintf("done!\n\n");

quit:
  if (server->state == STREAMING_IN_PROGRESS)
    server->state = STREAMING_ACCEPTING;

  TFRET();
}

TFTYPE