mh5streamer.c

Go to the documentation of this file.

/*******************************************************************************
 * Copyright © 2014 The DTVKit Open Software Foundation Ltd (www.dtvkit.org)
 * Copyright © 2010 Ocean Blue Software Ltd
 *
 * This file is part of a DTVKit Software Component
 * You are permitted to copy, modify or distribute this file subject to the terms
 * of the DTVKit 1.0 Licence which can be found in licence.txt or at www.dtvkit.org
 *
 * THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND,
 * EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
 *
 * If you or your organisation is not a member of DTVKit then you have access
 * to this source code outside of the terms of the licence agreement
 * and you are expected to delete this and any associated files immediately.
 * Further information on DTVKit, membership and terms can be found at www.dtvkit.org
 *******************************************************************************/
/*---includes for this file--------------------------------------------------*/
#include <string.h>

#include "stb_os.h"
#include "dvb_audio.h"
#include "dvb_video.h"
#include "httptype.h"
#include "mherrors.h"
#include "dvb_ics.h"

#ifdef INCLUDE_ICS

#include "mh5streamer.h"
#include "mh5profile.h"
#include "mh5memory.h"
#include "mh5gate.h"
#include "mh5object.h"
#include "mh5audio.h"
#include "mh5video.h"
#include "glue_queue.h"
#include "glue_timers.h"
#include "mh5application.h"
#include "mh5queue.h"
#include "mh5prgs.h"
#include "mh5control.h"
#include "mh5access.h"
#include "mh5tls.h"
#include "mh5fileorm.h"
#include "mh5parse.h"
#include "stmr_common.h"
#include "stmr_queue.h"
#include "stmr_dl.h"
#include "stmr_task.h"
#include "stmr_msp.h"
#include "glue_main.h"
#include "fs_types.h"

/*---constant definitions for this file--------------------------------------*/

#define STREAMER_STACK_SIZE  (4096)
#define MIN_BUFFERED_BYTES   (524288)
/*#define MIN_BUFFERED_BYTES   (3000000)*/
#define INVALID_MEASUREMENT    (-1)
#define MAX_HISTORY_SIZE     (262144)
#define MAX_REDIRECTS           (5)

#define DEBUG_DETAIL(x)
#define DBG(x)
#define PDBG(x) TRACE(TICS, x)

/* verbose debug */
#define VDBG(x)

#define REQUEST_STREAM_STATE_STR(r)                                     \
   ((r)->streamState == STATE_STREAM_STOPPED ? "STATE_STREAM_STOPPED" : \
    (r)->streamState == STATE_STREAM_PREPARED ? "STATE_STREAM_PREPARED" : \
    (r)->streamState == STATE_STREAM_RUNNING ? "STATE_STREAM_RUNNING" : \
    "UNKNOWN")

#define REQUEST_DOWNLOAD_STATE_STR(r)                                   \
   ((r)->downloadState == STATE_DOWNLOAD_STOPPED ? "STATE_DOWNLOAD_STOPPED" : \
    (r)->downloadState == STATE_DOWNLOAD_QUEUED ? "STATE_DOWNLOAD_QUEUED" : \
    (r)->downloadState == STATE_DOWNLOAD_CHECK_ACCESS ? "STATE_DOWNLOAD_CHECK_ACCESS" : \
    (r)->downloadState == STATE_DOWNLOAD_WAIT_ACCESS ? "STATE_DOWNLOAD_WAIT_ACCESS" : \
    (r)->downloadState == STATE_DOWNLOAD_HAS_ACCESS ? "STATE_DOWNLOAD_HAS_ACCESS" : \
    (r)->downloadState == STATE_DOWNLOAD_WAIT_CERT ? "STATE_DOWNLOAD_WAIT_CERT" : \
    (r)->downloadState == STATE_DOWNLOAD_HAS_CERT ? "STATE_DOWNLOAD_HAS_CERT" : \
    (r)->downloadState == STATE_DOWNLOAD_STARTED ? "STATE_DOWNLOAD_STARTED" : \
    (r)->downloadState == STATE_DOWNLOAD_RUNNING ? "STATE_DOWNLOAD_RUNNING" : \
    (r)->downloadState == STATE_DOWNLOAD_DONE ? "STATE_DOWNLOAD_DONE" : \
    (r)->downloadState == STATE_DOWNLOAD_FAILED ? "STATE_DOWNLOAD_FAILED" : \
    "UNKNOWN")

#define REQUEST_OUTPUT_STATE_STR(r)                                     \
   ((r)->outputState == STATE_OUTPUT_STOPPED ? "STATE_OUTPUT_STOPPED" : \
    (r)->outputState == STATE_OUTPUT_PENDING ? "STATE_OUTPUT_PENDING" : \
    (r)->outputState == STATE_OUTPUT_SETUP ? "STATE_OUTPUT_SETUP" :    \
    (r)->outputState == STATE_OUTPUT_READY ? "STATE_OUTPUT_READY" :    \
    (r)->outputState == STATE_OUTPUT_RUNNING ? "STATE_OUTPUT_RUNNING" : \
    (r)->outputState == STATE_OUTPUT_STARVED ? "STATE_OUTPUT_STARVED" : \
    (r)->outputState == STATE_OUTPUT_DONE ? "STATE_OUTPUT_DONE" : \
    "UNKNOWN")

#define REQUEST_FLOW_STATE_STR(r)                                       \
   ((r)->flowState == STATE_FLOW_STOPPED ? "STATE_FLOW_STOPPED" :      \
    (r)->flowState == STATE_FLOW_STARTED ? "STATE_FLOW_STARTED" :      \
    (r)->flowState == STATE_FLOW_RUNNING ? "STATE_FLOW_RUNNING" :      \
    (r)->flowState == STATE_FLOW_UNDERFLOW ? "STATE_FLOW_UNDERFLOW" :  \
    "UNKNOWN")

#define REQUEST_KEY_STATE_STR(r)                                        \
   ((r)->keyState == STATE_KEY_UNKNOWN ? "STATE_KEY_UNKNOWN" :         \
    (r)->keyState == STATE_KEY_VALID ? "STATE_KEY_VALID" :             \
    (r)->keyState == STATE_KEY_PENDING ? "STATE_KEY_PENDING" :         \
    (r)->keyState == STATE_KEY_ERROR ? "STATE_KEY_ERROR" :             \
    "UNKNOWN")


/*---local typedef structs for this file-------------------------------------*/

typedef enum
{
   ST_PES_PKT     = 0x06,
   ST_AAC_ADTS    = 0x0f,
   ST_HEAAC       = 0x11,
   ST_H264        = 0x1b,
   ST_DOLBY_BRAY  = 0x84,
   ST_DOLBY_ATSC  = 0x87,
} E_STREAM_TYPE;

typedef enum {
   ISO_LANG_DTAG = 0x0a,
   STREAMID_DTAG = 0x52,
   SUBTITLE_DTAG = 0x59,
} E_DTAGS;

typedef enum
{
   STATE_STREAM_STOPPED,
   STATE_STREAM_PREPARED,
   STATE_STREAM_RUNNING
} StreamState;

typedef enum
{
   STATE_DOWNLOAD_QUEUED,
   STATE_DOWNLOAD_STOPPED,
   STATE_DOWNLOAD_CHECK_ACCESS,
   STATE_DOWNLOAD_WAIT_ACCESS,
   STATE_DOWNLOAD_HAS_ACCESS,
   STATE_DOWNLOAD_WAIT_CERT,
   STATE_DOWNLOAD_HAS_CERT,
   STATE_DOWNLOAD_STARTED,
   STATE_DOWNLOAD_RUNNING,
   STATE_DOWNLOAD_DONE,
   STATE_DOWNLOAD_FAILED
} DownloadState;

typedef enum
{
   STATE_OUTPUT_STOPPED,
   STATE_OUTPUT_PENDING,
   STATE_OUTPUT_SETUP,
   STATE_OUTPUT_READY,
   STATE_OUTPUT_RUNNING,
   STATE_OUTPUT_STARVED,
   STATE_OUTPUT_DONE
} OutputState;

typedef enum
{
   STATE_FLOW_STOPPED,
   STATE_FLOW_STARTED,
   STATE_FLOW_RUNNING,
   STATE_FLOW_UNDERFLOW
} FlowState;

typedef enum
{
   STATE_KEY_UNKNOWN,
   STATE_KEY_VALID,
   STATE_KEY_PENDING,
   STATE_KEY_ERROR
} KeyState;

/* A/V component selection for IP streaming */
typedef struct
{
   BOOLEAN audio_active;
   BOOLEAN video_active;
   S32BIT audio_component_tag;
   S32BIT video_component_tag;
   E_VideoTermination video_termination;
} S_ICSComponents;

typedef struct sCounterTrigger
{
   S32BIT id;
   S32BIT position;
   struct sCounterTrigger *next;
} CounterTrigger;

typedef struct sStreamerRequest
{
   U32BIT requestId;
   U32BIT downloadId;

   MHEG5Root *root;

   BOOLEAN isMeasureType;
   BOOLEAN contentAvailableNotified;
   BOOLEAN contentAvailableGenerated;
   BOOLEAN streamPlayingGenerated;
   BOOLEAN paused;
   BOOLEAN restart;
   BOOLEAN isPartial;
   BOOLEAN pmtRetrieved;
   BOOLEAN terminateStreamSetup;

   S32BIT pauseRequestCount;
   S32BIT counterEndPosition;
   S32BIT counterMaxPosition;
   S32BIT looping;
   S32BIT currentLoop;

   U64BIT contentLength;
   U64BIT bytePosition;
   U64BIT currentBytePosition;
   U64BIT expectedBytes;
   U64BIT downloadedBytes;
   U64BIT presentedBytes;

   StreamState streamState;
   DownloadState downloadState;
   OutputState outputState;
   FlowState flowState;

   MHEG5String url;
   MHEG5String redirectUrl;

   U32BIT certCount;
   U32BIT redirectCount;

   U32BIT bitrate;
   U32BIT prepareNumber;
   U32BIT maxBytes;

   S_ICSKeys keys;
   MHEG5String keyLocation;
   KeyState keyState;
   U32BIT keyDownloadId;

   U8BIT pmt[STMR_MAX_PSI_SECTION];

   CounterTrigger *triggers;

   struct sStreamerRequest *next;
} StreamerRequest;

/*---local function definitions----------------------------------------------*/

/*---global function definitions---------------------------------------------*/

static StreamerRequest* CreateRequest(BOOLEAN isMeasure);
static void PrepareStreamingRequest(StreamerRequest *request,
   MHEG5Stream *stream);
static void SetCounterPosition(StreamerRequest *request, MHEG5Stream *stream);
static void AddRequestToList(StreamerRequest *request);
static void RemoveRequestFromList(StreamerRequest *request);
static void DestroyRequest(StreamerRequest *request);
static void CreateDownloadRequest(StreamerRequest *request);
static void ProgressDownloadRequest(StreamerRequest *request);
static void CheckTlsCertificates(StreamerRequest *request);
static void DestroyDownloadRequest(StreamerRequest *request);
static void DestroyMeasurementRequest(StreamerRequest *request);
static StreamerRequest* FindRequestById(U32BIT requestId);
static StreamerRequest* FindRequestByObj(MHEG5Root *root);
static StreamerRequest* FindRequestByDownloadId(U32BIT downloadId);
static void ProcessHttpHeaders(U32BIT downloadId,S32BIT code);
static void CalculateExpectedBytes(StreamerRequest *request);
static void HandleStreamKeys(StreamerRequest *request);
static void HandleDownloadFinished(U32BIT downloadId,E_HttpStatus status,S32BIT code);
static void HandleDownloadRedirection(StreamerRequest *request,
   BOOLEAN permanent);
static void HandleDownloadSuccess(StreamerRequest *request);
static void HandleDownloadPartial(StreamerRequest *request,U32BIT downloadId);
static void HandleMeasurementFinished(U32BIT downloadId);
static void HandleMeasurementRedirection(StreamerRequest *request);
static void HandleStreamPlaying(U32BIT requestId);
static void HandleStreamStopped(U32BIT requestId);
static void HandleCounterTrigger(U32BIT requestId, S32BIT triggerId);
static void HandleContentAvailable(U32BIT requestId);
static void GenerateStreamRefError(StreamerRequest *request);
#ifdef INCLUDE_FREESAT
static void GenerateIPStreamAccessError(StreamerRequest *request);
#endif  /* INCLUDE_FREESAT */
static void GenerateStreamUnderflow(void);
static void GenerateStreamUnderflowResume(void);
static void GenerateContentAvailable(StreamerRequest *request);
static void GenerateStreamPlaying(StreamerRequest *request);
static void GenerateStreamStopped(StreamerRequest *request);
static void GenerateCounterTrigger(StreamerRequest *request, S32BIT triggerId);
static void GenerateKeyFileError(void);
static void NotifyContentAvailable(U32BIT requestId);
static void NotifyStreamPlaying(U32BIT requestId);
static void NotifyStreamStopped(U32BIT requestId);
static void ItemInsertCallback(U32BIT downloadId, U64BIT base,
   U64BIT position, U32BIT len, BOOLEAN last);
static void ItemReleaseCallback(U32BIT requestId, U64BIT base,
   U64BIT position, U32BIT len, BOOLEAN last);
static void CheckCounterTriggers(StreamerRequest *request, U64BIT base,
   U64BIT position, U32BIT len);
static void CheckRunCondition(StreamerRequest *request);
static BOOLEAN IsEnoughDataBuffered(StreamerRequest *request);
static U32BIT ConvertToUnits(StreamerRequest *request, U64BIT bytes);
static U64BIT ConvertToBytes(StreamerRequest *request, U32BIT units);
static BOOLEAN IsStreamRunning(void);
static void CheckDownloadCondition(void);
static void StopContentOutput(StreamerRequest *request, BOOLEAN flush);
static void ResumePlayback(StreamerRequest *request);
static BOOLEAN IsValidUrl(MHEG5String *url);
static void UnderflowCallback(U32BIT requestId,U32BIT playoutTime);
static void StreamKeyLoadedCallback(void *userData, S_CONTENT *content);
static void StreamKeyLoadFailCallback(void *userData);
static void ServerAccessCallback(void);
static void TlsCertCallback(void);
static BOOLEAN MHEG5StreamerProcess(void);


/*---local (static) variable declarations for this file----------------------*/
static StreamerRequest *requestList = NULL;

/* Start from 1, to detect invalid requests */
static U32BIT globalRequestId = 1;

/* Start from 1, to detect invalid values */
static U32BIT globalPrepareNumber = 1;

static void *streamerBuffer = NULL;
static U32BIT streamerBufferSize = 0;

static void *requestMutex = NULL;
static BOOLEAN isIcsSetup = FALSE;
static BOOLEAN processRequests = FALSE;

E_MhegErr MHEG5StreamerOpen(void *buffer, U32BIT size, U32BIT taskPriority)
{
   E_MhegErr result;
   streamerBuffer = NULL;
   streamerBufferSize = 0;

   if (buffer == NULL)
   {
   #ifdef ERROR_ON_SMALL_ICS_BUFFER
      TRACE(TERROR, ("Streaming buffer must not be NULL"));
      result = MHERR_BAD_ICS_BUFFER;
   #else
      DPL2(("Streaming buffer NULL - Streaming disabled"));
      result = MHERR_OK;
   #endif
   }
   else if (size < ICS_MINIMUM_BUFFER_SIZE)
   {
   #ifdef ERROR_ON_SMALL_ICS_BUFFER
      TRACE(TERROR, ("Minimum buffer size is %d, given %d", ICS_MINIMUM_BUFFER_SIZE, size));
      result = MHERR_BAD_ICS_BUFFER_SIZE;
   #else
      DPL2(("Streaming buffer too small - Streaming disabled"));
      result = MHERR_OK;
   #endif
   }
   else
   {
      streamerBuffer = buffer;
      streamerBufferSize = size;

      /* Initialise queue manager */
      result = MHEG5QueueInit(streamerBuffer, streamerBufferSize);
      if (result == MHERR_OK)
      {
         MHEG5QueueRegisterInsertCallback(ItemInsertCallback);
         MHEG5QueueRegisterReleaseCallback(ItemReleaseCallback);

         /* Initialise streamer task */
         MHEG5StreamerRegisterUnderflowCallback(UnderflowCallback);

         requestList = NULL;
         ASSERT(requestMutex == NULL)

         requestMutex = STB_OSCreateMutex();
         if (requestMutex == NULL)
         {
            TRACE(TERROR, ("Streaming requerst mutex failure"));
            result = MHERR_SYSTEM_RESOURCE_MUTEX;
         }
         else
         {
            MHEG5AddServerAccessCallback(ServerAccessCallback);
            MHEG5AddTlsCertificateCallback(TlsCertCallback);

            result = MHEG5StreamerStartTask(STREAMER_STACK_SIZE, (U8BIT)taskPriority);
         }
      }
   }
   return result;
}

void MHEG5StreamerReset(void)
{
   StreamerRequest *request;
   StreamerRequest *list;
   if (streamerBuffer != NULL)
   {
      MHEG5QueueReleaseAllItems();

      STB_OSMutexLock(requestMutex);
      list = requestList;
      requestList = NULL;
      STB_OSMutexUnlock(requestMutex);

      /* The request list should be empty, but clean up if not */
      while (list != NULL)
      {
         request = list;
         list = request->next;
         DestroyRequest(request);
      }
   }
}

void MHEG5StreamerPrepare(MHEG5Stream *stream)
{
   StreamerRequest *request;

   assert(((MHEG5Root *)stream)->clazz == MHEG5STREAM);
   request = FindRequestByObj(&stream->ingredient.root);
   if (request == NULL)
   {
      /* Create and initialise new request */
      request = CreateRequest(FALSE);
      if (request != NULL)
      {
         STB_OSMutexLock(requestMutex);
         AddRequestToList(request);
         STB_OSMutexUnlock(requestMutex);
      }
   }
    else
    {
        /* This could be SetData action - previous counter max posn is no longer valid */
        request->counterMaxPosition = -1;
    }

   if (request != NULL)
   {
      /* ContentAvailable will only be generated when preparing
       * the stream (also after SetData and Preload, if necessary). But
       * if the stream is stopped and then run again, ContentAvailable
       * will not be generated.
       */
      if (request->streamState != STATE_STREAM_RUNNING)
      {
         request->contentAvailableNotified = FALSE;
         request->contentAvailableGenerated = FALSE;
      }

      /* Update "prepare number" to push the request to the end
       * of the queue. If the request is primary, it doesn't matter.
       */
      request->prepareNumber = globalPrepareNumber;
      globalPrepareNumber = (globalPrepareNumber % 1073741824) + 1;

      PrepareStreamingRequest(request, stream);
   }
}

S32BIT MHEG5StreamerGetCounterPosition(MHEG5Stream *stream)
{
   StreamerRequest *request;
   S32BIT counterPosition;

   counterPosition = 0;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      STB_OSMutexLock(requestMutex);
      counterPosition = ConvertToUnits(request, request->currentBytePosition);
      STB_OSMutexUnlock(requestMutex);
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
      counterPosition = stream->counterPosition;
   }

   return counterPosition;
}

void MHEG5StreamerSetCounterPosition(MHEG5Stream *stream)
{
   StreamerRequest *request;
   S32BIT endPosition;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      /* Sanity check */
      if (stream->counterPosition >= 0)
      {
         if (((request->counterMaxPosition == -1) ||
              (request->counterMaxPosition > stream->counterPosition)) &&
             ((request->counterEndPosition == -1) ||
              (request->counterEndPosition > stream->counterPosition)))
         {
            SetCounterPosition(request, stream);
         }
         else
         {
            if (request->streamState == STATE_STREAM_RUNNING)
            {
               MHEG5QueueDisableStreaming();
            }

            /* Stop current download */
            DestroyDownloadRequest(request);

            /* Generate StreamStopped if required */
            if (request->outputState != STATE_OUTPUT_DONE)
            {
               GenerateStreamStopped(request);
            }

            /* Disable streaming and remove queue items from the queue */
            StopContentOutput(request, TRUE);

            /* Update counter position */
            endPosition = stream->counterPosition;
            if (stream->counterEndPosition >= 0 &&
                stream->counterEndPosition < endPosition)
            {
               endPosition = stream->counterEndPosition;
            }
            if (stream->counterMaxPosition >= 0 &&
                stream->counterMaxPosition < endPosition)
            {
               endPosition = stream->counterMaxPosition;
            }
            request->bytePosition = ConvertToBytes(request, endPosition);
            request->currentBytePosition = request->bytePosition;
            request->currentLoop = 0;
            request->restart = TRUE;

            if (request->streamState == STATE_STREAM_RUNNING && isIcsSetup)
            {
               isIcsSetup = FALSE;
               DVB_MhegICStreamRelease();
            }
         }
      }
   }
}

void MHEG5StreamerSetCounterEndPosition(MHEG5Stream *stream)
{
   StreamerRequest *request;
   //S32BIT contentLength = -1;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      request->counterEndPosition = stream->counterEndPosition;
      //contentLength = 0;
      //if (request->counterMaxPosition != -1)
      //{
      //    contentLength = request->counterMaxPosition + 1;
      //}
      CalculateExpectedBytes(request);
   }
}

S32BIT MHEG5StreamerGetCounterMaxPosition(MHEG5Stream *stream)
{
   StreamerRequest *request;
   S32BIT counterMaxPosition;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      STB_OSMutexLock(requestMutex);
      counterMaxPosition = request->counterMaxPosition;
      STB_OSMutexUnlock(requestMutex);
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
      counterMaxPosition = stream->counterMaxPosition;
   }

   return counterMaxPosition;
}

void MHEG5StreamerSetCounterTrigger(MHEG5Stream *stream, S32BIT triggerId,
   S32BIT newValue)
{
   StreamerRequest *request;
   CounterTrigger **pTrigger, *trigger;
   BOOLEAN found;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      STB_OSMutexLock(requestMutex);

      found = FALSE;
      pTrigger = &request->triggers;
      while (*pTrigger != NULL)
      {
         trigger = *pTrigger;
         if (trigger->id == triggerId)
         {
            if (newValue >= 0)
            {
               /* Replace value */
               trigger->position = newValue;
            }
            else
            {
               /* Remove trigger */
               *pTrigger = trigger->next;
               MHEG5freeMem(trigger);
            }
            found = TRUE;
            break;
         }
         pTrigger = &trigger->next;
      }

      if (!found && newValue >= 0)
      {
         /* Add new trigger */
         trigger = MHEG5getMem(sizeof *trigger);
         if (trigger != NULL)
         {
            trigger->id = triggerId;
            trigger->position = newValue;
            trigger->next = NULL;
            *pTrigger = trigger;
         }
      }
      STB_OSMutexUnlock(requestMutex);
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
   }
}

void MHEG5StreamerRun(MHEG5Stream *stream)
{
   StreamerRequest *request;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         /* Shouldn't be here - already running */
         TRACE(TICS,("already running"));
         return;
      }

      if (IsStreamRunning())
      {
         /* Another stream is already running, do nothing */
         TRACE(TICS,("another running"));
         return;
      }

      /* If the stream is at the end (including looping), it should
       * play as if it were activated for the first time.
       */
      if (request->restart)
      {
         /* Restart stream */
         request->currentLoop = 0;
         ULL_Set32(request->bytePosition, 0);
         ULL_Set32(request->currentBytePosition, 0);
      }

        #ifdef TO_REMOVE
      /* The download may be stopped in the following scenario:
       *   prepare -> run -> stop -> run
       * In that case the stream is already "prepared", but the
       * download is stopped. In this case we need to start
       * the download again (everything else is ready to run).
       */
      if (request->downloadState == STATE_DOWNLOAD_STOPPED)
      {
         /* Already stopped, need to re-prepare it */
         CreateDownloadRequest(request);
      }
        #endif

      STB_OSMutexLock(requestMutex);

      /* StreamPlaying will be generated when the stream starts
       * playing
       */
      request->streamPlayingGenerated = FALSE;
      request->streamState = STATE_STREAM_RUNNING;

      CheckDownloadCondition();

      assert(request->outputState == STATE_OUTPUT_STOPPED);
      CheckRunCondition(request);

      STB_OSMutexUnlock(requestMutex);
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
   }
}

void MHEG5StreamerStop(MHEG5Stream *stream)
{
   StreamerRequest *request;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      if (request->streamState == STATE_STREAM_STOPPED)
      {
         /* Nothing to do. This can happen if the stream was :Run, but
          * another stream was already running. In that case the streamer
          * state is incosistent with the object state, so :Stop can
          * still be used on the Stream object, but the request is
          * actually stopped.
          */
         return;
      }

      if (request->streamState == STATE_STREAM_RUNNING)
      {
         MHEG5QueueDisableStreaming();
      }

      /* Stop and destroy the HTTP request before clearing the queue,
       * otherwise there's a race condition.
       */
      DestroyDownloadRequest(request);

      /* Generate StreamStopped if required */
      if (request->outputState != STATE_OUTPUT_DONE)
      {
         GenerateStreamStopped(request);
      }

      /* Disable streaming and remove queue items from the queue */
      StopContentOutput(request, TRUE);

      /* Update counter position */
      stream->counterPosition = ConvertToUnits(request, request->currentBytePosition);
      request->bytePosition = request->currentBytePosition;
      request->pauseRequestCount = 0;

      MHEG5stringDestruct(&request->redirectUrl);
      request->redirectCount = 0;

      if (request->streamState == STATE_STREAM_RUNNING && isIcsSetup)
      {
         isIcsSetup = FALSE;
         DVB_MhegICStreamRelease();
      }

      request->streamState = STATE_STREAM_STOPPED;
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
   }
}

void MHEG5StreamerPause(MHEG5Stream *stream)
{
   StreamerRequest *request;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      if ((request->streamState == STATE_STREAM_RUNNING) &&
          (!request->paused))
      {
         switch (request->outputState)
         {
            case STATE_OUTPUT_READY:
            case STATE_OUTPUT_RUNNING:
            case STATE_OUTPUT_STARVED:
               ++request->pauseRequestCount;
               if (request->pauseRequestCount == 1)
               {
                  DVB_MhegICStreamPause();
                  MHEG5QueueDisableStreaming();
               }
               break;

            default:
               /* Do nothing - no output */
               ;
         }

         /* Update counter position */
         stream->counterPosition = ConvertToUnits(request, request->currentBytePosition);
      }
      request->paused = TRUE;
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
   }
}

void MHEG5StreamerResume(MHEG5Stream *stream)
{
   StreamerRequest *request;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      if ((request->streamState == STATE_STREAM_RUNNING) &&
          (request->paused))
      {
         switch (request->outputState)
         {
            case STATE_OUTPUT_READY:
            case STATE_OUTPUT_RUNNING:
            case STATE_OUTPUT_STARVED:
            case STATE_OUTPUT_STOPPED:
               ResumePlayback(request);
               break;
            default:
               ;
         }
      }
      request->paused = FALSE;
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
   }
}

BOOLEAN MHEG5StreamerIsRunable(MHEG5Stream *stream)
{
   return (FindRequestByObj(&stream->ingredient.root))? TRUE : FALSE;
}

void MHEG5StreamerRemove(MHEG5Stream *stream)
{
   StreamerRequest *request;

   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         MHEG5QueueDisableStreaming();
      }

      /* Stop and destroy the HTTP request before clearing the queue,
       * otherwise there's a race condition.
       */
      DestroyDownloadRequest(request);

      /* Disable streaming and remove queue items from the queue */
      StopContentOutput(request, TRUE);

      STB_OSMutexLock(requestMutex);
      RemoveRequestFromList(request);
      STB_OSMutexUnlock(requestMutex);

      DestroyRequest(request);
   }
}

void MHEG5StreamerHandle(MHEG5StreamerEventParams_t *params)
{
    switch (params->eventType)
   {
      case MHEG5_STREAMER_EVENT_MSP_DOWNLOAD_DONE:
         PDBG(("MHEG5_STREAMER_EVENT_MSP_DOWNLOAD_DONE"));
         HandleMeasurementFinished(params->requestId);
         break;
      case MHEG5_STREAMER_EVENT_HTTP_HEADERS_DONE:
         PDBG(("MHEG5_STREAMER_HTTP_HEADERS_DONE"));
         ProcessHttpHeaders(params->requestId,params->code);
         break;
      case MHEG5_STREAMER_EVENT_HTTP_DOWNLOAD_DONE:
         PDBG(("MHEG5_STREAMER_EVENT_HTTP_DOWNLOAD_DONE"));
         HandleDownloadFinished(params->requestId,params->status,params->code);
         break;
      case MHEG5_STREAMER_EVENT_TASK_STREAM_PLAYING:
         PDBG(("MHEG5_STREAMER_EVENT_TASK_STREAM_PLAYING"));
         HandleStreamPlaying(params->requestId);
         break;
      case MHEG5_STREAMER_EVENT_TASK_STREAM_STOPPED:
         PDBG(("MHEG5_STREAMER_EVENT_TASK_STREAM_STOPPED"));
         HandleStreamStopped(params->requestId);
         break;
      case MHEG5_STREAMER_EVENT_COUNTER_TRIGGER:
         PDBG(("MHEG5_STREAMER_EVENT_COUNTER_TRIGGER"));
         HandleCounterTrigger(params->requestId, params->triggerId);
         break;
      case MHEG5_STREAMER_EVENT_CONTENT_AVAILABLE:
         DBG(("MHEG5_STREAMER_EVENT_CONTENT_AVAILABLE"));
         HandleContentAvailable(params->requestId);
         break;
      default:
         PDBG(("Unhandled event %d!", params->eventType));
         break;
   }
   MH5GlueAddPostProcessFunc( MHEG5StreamerProcess );
}

void MHEG5StreamerNotifyStreamStarted(MHEG5Stream *stream)
{
   StreamerRequest *request;

   assert(((MHEG5Root *)stream)->clazz == MHEG5STREAM);
   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      switch (request->flowState)
      {
         case STATE_FLOW_STARTED:
            request->flowState = STATE_FLOW_RUNNING;
            GenerateStreamPlaying(request);
            break;
         case STATE_FLOW_UNDERFLOW:
            request->flowState = STATE_FLOW_RUNNING;
            GenerateStreamUnderflowResume();
            break;
         case STATE_FLOW_RUNNING:
            /* Multiple components, playback resumed or confused decoders */
            DBG(("Ignoring StreamStarted in STATE_FLOW_RUNNING"));
            break;
         case STATE_FLOW_STOPPED:
            PDBG(("%d ~ WARNING: Stream started when flow is stopped",
                  request->requestId));
            break;
         default:
            TRACE(TERROR, ("Unhandled flow state"));
            assert(0);
            break;
      }
   }
}

void MHEG5StreamerNotifyStreamStopped(MHEG5Stream *stream)
{
   StreamerRequest *request;

   assert(((MHEG5Root *)stream)->clazz == MHEG5STREAM);
   request = FindRequestByObj(&stream->ingredient.root);
   if (request != NULL)
   {
      PDBG(("%d ~ request->flowState = %s", request->requestId,
            REQUEST_FLOW_STATE_STR(request)));
      switch (request->flowState)
      {
         case STATE_FLOW_RUNNING:
            PDBG(("%d ~ request->downloadState = %s", request->requestId,
                  REQUEST_DOWNLOAD_STATE_STR(request)));
            if (request->downloadState == STATE_DOWNLOAD_RUNNING)
            {
               request->flowState = STATE_FLOW_UNDERFLOW;
               GenerateStreamUnderflow();
            }
            else if (request->downloadState == STATE_DOWNLOAD_DONE)
            {
               request->flowState = STATE_FLOW_STOPPED;
               GenerateStreamStopped(request);
            }
            else
            {
               TRACE(TERROR, ("Unhandled download state"));
               assert(0);
            }
            break;
         case STATE_FLOW_STOPPED:
            /* Expected underflow, ignore */
            DBG(("Expected underflow in %s", REQUEST_FLOW_STATE_STR(request)));
            break;
         case STATE_FLOW_UNDERFLOW:
            /* Spurious notification */
            break;
         case STATE_FLOW_STARTED:
            /* This can happen if the flow is starting for a new stream
             * (or a new position in the same stream), without notifying
             * that the old stream is stopped.
             * TODO: Need to think if this is valid
             */
            break;
         default:
            TRACE(TERROR, ("Unhandled flow state"));
            assert(0);
      }
   }
}

static BOOLEAN MHEG5StreamerProcess(void)
{
   StreamerRequest *request;

   if (processRequests)
   {
      /* Set flag to FALSE, but while processing this flag may be
       * turned on again.
       */
      processRequests = FALSE;

      request = requestList;
      while (request != NULL)
      {
         switch (request->downloadState)
         {
            case STATE_DOWNLOAD_CHECK_ACCESS:
               ProgressDownloadRequest(request);
               break;

            default:
               ;
         }

         request = request->next;
      }
   }
   return processRequests;
}

void MHEG5StreamerMeasurePerformance(MHEG5String *url, MHEG5Int maxBytes,
   MHEG5Program *program)
{
   StreamerRequest *request;

   request = CreateRequest(TRUE);
   if (request != NULL)
   {
      /* Convert using name mapping rules */
      MHEG5stringDestruct(&request->url);
      request->url = MHEG5convertGID(url);
      if (IsValidUrl(&request->url))
      {
         DBG(("Measurement URL: %s", request->url.data));

         request->root = &program->ingredient.root;
         request->maxBytes = maxBytes;

         STB_OSMutexLock(requestMutex);
         AddRequestToList(request);
         STB_OSMutexUnlock(requestMutex);

         request->downloadState = STATE_DOWNLOAD_QUEUED;
         ProgressDownloadRequest(request);
      }
      else
      {
         /* Invalid source, cannot download */
         DBG(("MHEG5StreamerMeasurePerformance: Invalid URL"));
         request->downloadState = STATE_DOWNLOAD_FAILED;
      }
   }
}

void MHEG5StreamerAbortMeasurement(MHEG5Program *program)
{
   StreamerRequest *request;

   assert(((MHEG5Root *)program)->clazz == MHEG5RESIDENTPROGRAM);
   request = FindRequestByObj(&program->ingredient.root);
   if (request != NULL)
   {
      MHEG5StreamerNotifyStreamingPerformance(request->requestId, -1);
   }
   else
   {
      /* Ignore quietly */
      TRACE(TERROR, ("Cannot find request"));
   }
}

void MHEG5StreamerSetActiveState(BOOLEAN activeState)
{
   StreamerRequest *request;

   /* Find the running request (if any) */
   request = requestList;
   while (request != NULL)
   {
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         break;
      }
      request = request->next;
   }

   /* This only affects streaming output */
   if (request != NULL)
   {
      if (activeState && isIcsSetup)
      {
         PDBG(("rid=%d",request->requestId));
         MHEG5QueueEnableStreaming(request->requestId);
         if (request->isPartial)
         {
            request->downloadState = STATE_DOWNLOAD_QUEUED;
            ProgressDownloadRequest(request);
         }
      }
      else
      {
         MHEG5QueueDisableStreaming();
      }
   }
}

void MHEG5StreamerClose(void)
{
   if (streamerBuffer != NULL)
   {
      /* Stop and delete every request */
      MHEG5StreamerReset();

      /* Stop the task */
      MHEG5StreamerStopTask();

      if (requestMutex != NULL)
      {
         STB_OSDeleteMutex( requestMutex );
         requestMutex = NULL;
      }

      /* Terminate queue */
      MHEG5QueueTerm();
   }
}

void MHEG5StreamerRefresh(void)
{
   StreamerRequest *request;
   if (streamerBuffer != NULL)
   {
      STB_OSMutexLock(requestMutex);
      request = requestList;
      while (request != NULL)
      {
         if (request->streamState == STATE_STREAM_RUNNING)
         {
            MHEG5QueueDisableStreaming();
            request->flowState = STATE_FLOW_STOPPED;
            request->outputState = STATE_OUTPUT_STOPPED;
            CheckRunCondition(request);
         }
         request = request->next;
      }
      STB_OSMutexUnlock(requestMutex);
   }
}

/*****************************************************************************/
/* NOTIFICATIONS FROM MSP MODULE                                             */
/*****************************************************************************/

void MHEG5StreamerNotifyStreamingPerformance(U32BIT requestId, U32BIT speed)
{
   StreamerRequest *request;

   request = FindRequestById(requestId);
   if (request != NULL)
   {
      /* Stop and destroy the measurement request */
      MHEG5StopMeasureRequest(requestId);
      MHEG5DestroyMeasureRequest(requestId);

      /* Report measurement to resident program */
      MHEG5notifyStreamPerformanceMeasurement(request->root, speed);

      /* Remove request from list and destroy it */
      STB_OSMutexLock(requestMutex);
      RemoveRequestFromList(request);
      STB_OSMutexUnlock(requestMutex);

      DestroyRequest(request);
   }
}

/*****************************************************************************/
/*                                                                           */
/*****************************************************************************/

static StreamerRequest* CreateRequest(BOOLEAN isMeasure)
{
   StreamerRequest *request;

#ifndef ERROR_ON_SMALL_ICS_BUFFER
   if (streamerBuffer == NULL)
   {
      TRACE(TERROR, ("Streaming buffer must not be NULL"));
      request = NULL;
   }
   else
#endif
   {
      request = MHEG5getMem(sizeof *request);
      if (request != NULL)
      {
         /* Request ID is a unique ID */
         request->requestId = globalRequestId;
         globalRequestId = (globalRequestId % 1073741824) + 1;

         /* Download ID is unknown */
         request->downloadId = 0;

         /* Set type */
         request->isMeasureType = isMeasure;

         /* Initialise request */
         request->paused = TRUE;
         request->isPartial = FALSE;
         request->pauseRequestCount = 0;
         request->contentAvailableNotified = FALSE;
         request->contentAvailableGenerated = FALSE;
         request->streamPlayingGenerated = FALSE;
         request->counterEndPosition = -1;
         request->counterMaxPosition = -1;
         request->certCount = 0;
         request->streamState = STATE_STREAM_STOPPED;
         request->downloadState = STATE_DOWNLOAD_STOPPED;
         request->outputState = STATE_OUTPUT_STOPPED;
         request->flowState = STATE_FLOW_STOPPED;

         ULL_SetInvalid(request->contentLength);
         ULL_Set32(request->bytePosition, 0);
         ULL_Set32(request->currentBytePosition, 0);
         ULL_SetInvalid(request->expectedBytes);
         ULL_Set32(request->downloadedBytes, 0);
         ULL_Set32(request->presentedBytes, 0);

         request->url.len = 0;
         request->url.data = NULL;

         request->redirectUrl.len = 0;
         request->redirectUrl.data = NULL;
         request->redirectCount = 0;

         /* 0 = unknown, X-BytesPerSecond must be positive */
         request->bitrate = 0;

         request->keys.num_pids = 0;
         request->keys.pid_info = NULL;
         request->keyLocation.len = 0;
         request->keyLocation.data = NULL;
         request->keyState = STATE_KEY_UNKNOWN;
         request->keyDownloadId = 0;

         request->pmtRetrieved = FALSE;
         request->triggers = NULL;
         request->prepareNumber = globalPrepareNumber;
         globalPrepareNumber = (globalPrepareNumber % 1073741824) + 1;
      }
   }
   return request;
}

static void PrepareStreamingRequest(StreamerRequest *request,
   MHEG5Stream *stream)
{
   int len;
   MHEG5String *ref;
   MHEG5String newUrl;
   MHEG5Bool sameUrl;

   ref = &stream->ingredient.content.ref.referenced.reference;

   /* Store stream object reference */
   request->root = &stream->ingredient.root;

   /* Convert using name mapping rules */
   newUrl = MHEG5convertGID(ref);
   sameUrl = MHEG5stringEqual(&newUrl, &request->url);
   MHEG5stringDestruct( &request->url );
   request->url = newUrl;

   /* Clear any previous redirections */
   MHEG5stringDestruct(&request->redirectUrl);
   request->redirectCount = 0;

   len = request->url.len;
   if (len == 0)
   {
      PDBG(("%d ~ Cannot convert GID", request->requestId));
      request->downloadState = STATE_DOWNLOAD_FAILED;
      return;
   }

   if (IsValidUrl(&request->url))
   {
      PDBG(("Stream URL: %s, counterPosition %d", request->url.data, stream->counterPosition));

      if (request->streamState == STATE_STREAM_RUNNING)
      {
         MHEG5QueueDisableStreaming();
      }

      /* Stop current download */
      DestroyDownloadRequest(request);

      /* Disable streaming and remove queue items from the queue */
      StopContentOutput(request, TRUE);

      /* Counter position is taken from the stream object */
      request->bytePosition = ConvertToBytes(request, stream->counterPosition);
      request->currentBytePosition = request->bytePosition;
      request->counterEndPosition = stream->counterEndPosition;

      request->looping = stream->looping;
      request->currentLoop = 0;
      request->restart = FALSE;

      if (stream->speed[0] == 0)
      {
         request->paused = TRUE;
      }
      else
      {
         request->paused = FALSE;
      }

      switch (request->streamState)
      {
         case STATE_STREAM_RUNNING:
            request->flowState = STATE_FLOW_STOPPED;
            request->outputState = STATE_OUTPUT_STOPPED;
            break;
         case STATE_STREAM_STOPPED:
            request->streamState = STATE_STREAM_PREPARED;
            /*fall thro*/
         case STATE_STREAM_PREPARED:
            request->contentAvailableNotified = FALSE;
            request->contentAvailableGenerated = FALSE;
            break;
      }

      /* If it's the same URL, we assume that the PMT and stream
       * keys are the same. We also expect another ContentAvailable event
       */
      if (!sameUrl)
      {
         request->contentAvailableNotified = FALSE;
         request->contentAvailableGenerated = FALSE;

         request->pmtRetrieved = FALSE;
         if (request->keys.num_pids > 0)
         {
            MHEG5freeMem(request->keys.pid_info);
            request->keys.num_pids = 0;
            request->keys.pid_info = NULL;
         }
         MHEG5stringDestruct(&request->keyLocation);
         request->keyState = STATE_KEY_UNKNOWN;
         request->keyDownloadId = 0;
      }

      CheckDownloadCondition();
   }
   else
   {
      /* Invalid source, cannot download */
      DBG(("InitStreamRequest: Invalid source"));
      request->downloadState = STATE_DOWNLOAD_FAILED;
   }
}

static void SetCounterPosition(StreamerRequest *request, MHEG5Stream *stream)
{
   U64BIT oldpos;
   U64BIT newpos;
   U64BIT downpos;

   oldpos = request->currentBytePosition;
   downpos = request->bytePosition;
   ULL_Add(downpos,request->downloadedBytes);

   /* Counter position is taken from the stream object */
   newpos = ConvertToBytes(request, stream->counterPosition);
   TRACE(TICS,("requestId=%d newBytePos=%d oldBytePos=%d",request->requestId,
      ULL_Get32(newpos), ULL_Get32(oldpos) ));

   if (ULL_Compare(newpos,oldpos) > 0 &&
       ULL_Compare(newpos,downpos) <= 0)
   {
      /* flush downloaded and queued data that is no longer required */
      TRACE(TICS,("flushing data that is no longer required"));
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         MHEG5QueueDisableStreaming();
         request->flowState = STATE_FLOW_STOPPED;
         request->outputState = STATE_OUTPUT_STOPPED;
      }
      MHEG5QueueReleaseItems(request->requestId,newpos);
      request->currentBytePosition = newpos;

      STB_OSMutexLock(requestMutex);
      CheckRunCondition(request);
      STB_OSMutexUnlock(requestMutex);
   }
   else if (stream->ingredient.root.runningStatus)
   {
      TRACE(TICS,("SCP outside of buffered data, restarting download SCOut"));
      /* New counter position maps to place outside of the currently buffered data */
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         MHEG5QueueDisableStreaming();
      }
      /* remove queue items from the queue */
      MHEG5QueueReleaseRequestItems(request->requestId);

      /* Stop current download */
      DestroyDownloadRequest(request);

      /* new download request will be at new counter position */
      request->bytePosition = newpos;
      request->currentBytePosition = newpos;

      request->currentLoop = 0;
      request->restart = FALSE;

      PDBG(("streamState %u", request->streamState));
      switch (request->streamState)
      {
         case STATE_STREAM_RUNNING:
            request->flowState = STATE_FLOW_STOPPED;
            request->outputState = STATE_OUTPUT_STOPPED;
            break;

         case STATE_STREAM_STOPPED:
            request->streamState = STATE_STREAM_PREPARED;
            /*fall thro*/
         case STATE_STREAM_PREPARED:
            request->contentAvailableNotified = FALSE;
            request->contentAvailableGenerated = FALSE;
            break;
      }

      CheckDownloadCondition();
   }
}

static void AddRequestToList(StreamerRequest *request)
{
   StreamerRequest **ppRequest;

   /* Always append, as the order is important for buffering model */
   ppRequest = &requestList;
   while (*ppRequest != NULL)
   {
      ppRequest = &(*ppRequest)->next;
   }

   *ppRequest = request;
   request->next = NULL;
}

static void RemoveRequestFromList(StreamerRequest *request)
{
   StreamerRequest **ppRequest;

   ppRequest = &requestList;
   while (*ppRequest != NULL && *ppRequest != request)
   {
      ppRequest = &(*ppRequest)->next;
   }

   if (*ppRequest != NULL)
   {
      /* Found the request in the list, remove it */
      *ppRequest = (*ppRequest)->next;
   }
}

static void CreateDownloadRequest(StreamerRequest *request)
{
   assert(request->downloadId == 0);
   assert(request->downloadState != STATE_DOWNLOAD_STARTED &&
      request->downloadState != STATE_DOWNLOAD_RUNNING);

   request->downloadState = STATE_DOWNLOAD_QUEUED;
   ProgressDownloadRequest(request);
}

static void ProgressDownloadRequest(StreamerRequest *request)
{
   MHEG5ServerAccess_t access;
   BOOLEAN lastRequest;
   char *url;
   U64BIT rangeFrom, rangeTo;

   if (request->downloadState == STATE_DOWNLOAD_QUEUED)
   {
      /* Start from scratch */
      PDBG(("%d ~ Progress: STATE_DOWNLOAD_QUEUED", request->requestId));
      ULL_Set32(request->downloadedBytes, 0);
      ULL_Set32(request->presentedBytes, 0);
      request->downloadState = STATE_DOWNLOAD_CHECK_ACCESS;
   }

   if (request->downloadState == STATE_DOWNLOAD_CHECK_ACCESS)
   {
      /* Check server access */
      PDBG(("%d ~ Progress: STATE_DOWNLOAD_CHECK_ACCESS", request->requestId));
      if (request->redirectCount == 0)
      {
         access = MHEG5CheckServerPermitted(request->url.data);
         if (access == SERVER_ACCESS_BLOCKED)
         {
            request->downloadState = STATE_DOWNLOAD_FAILED;
            GenerateStreamRefError(request);
            PDBG(("%d ~ Progress: STATE_DOWNLOAD_FAILED", request->requestId));
         }
         else if (access == SERVER_ACCESS_PENDING)
         {
            request->downloadState = STATE_DOWNLOAD_WAIT_ACCESS;
            PDBG(("%d ~ Progress: STATE_DOWNLOAD_WAIT_ACCESS", request->requestId));
         }
         else
         {
            request->downloadState = STATE_DOWNLOAD_HAS_ACCESS;
         }
      }
      else
      {
         /* URL obtained by redirection does not need to be checked */
         request->downloadState = STATE_DOWNLOAD_HAS_ACCESS;
      }
   }

   if (request->downloadState == STATE_DOWNLOAD_HAS_ACCESS)
   {
      /* Check TLS certificates */
      PDBG(("%d ~ Progress: STATE_DOWNLOAD_HAS_ACCESS", request->requestId));
      CheckTlsCertificates(request);
   }

   if (request->downloadState == STATE_DOWNLOAD_HAS_CERT)
   {
      /* Check redirection */
      PDBG(("%d ~ Progress: STATE_DOWNLOAD_HAS_CERT", request->requestId));
      if (request->redirectCount > 0)
      {
         url = (char *)request->redirectUrl.data;
      }
      else
      {
         url = (char *)request->url.data;
      }

      if (request->isPartial)
      {
         assert(request->downloadId != 0);
         MHEG5RenewDownloadRequest(request->downloadId,url);
         request->isPartial = FALSE;
      }
      else
      {
         /* Check last request -- depends on looping attribute */
         if (request->currentLoop == request->looping - 1)
         {
            lastRequest = TRUE;
         }
         else
         {
            lastRequest = FALSE;
         }

         rangeFrom = request->bytePosition;
         ULL_SetInvalid(rangeTo);
         if (!request->isMeasureType)
         {
            request->downloadId = MHEG5CreateDownloadRequest(
                  request->requestId, url, rangeFrom, rangeTo, lastRequest);
         }
         else
         {
            request->downloadId = MHEG5CreateMeasureRequest(
                  request->requestId, url, request->maxBytes);
         }
      }

      if (request->downloadId != 0)
      {
         request->downloadState = STATE_DOWNLOAD_STARTED;
         if (!request->isMeasureType)
         {
            MHEG5StartDownloadRequest(request->downloadId);
         }
         else
         {
            MHEG5StartMeasureRequest(request->downloadId);
         }
      }
      else
      {
         request->downloadState = STATE_DOWNLOAD_FAILED;
      }
   }
}

static void CheckTlsCertificates(StreamerRequest *request)
{
   U32BIT certCount;
   MHEG5TlsCertRequestStatus status;
   U8BIT *url;

   if (request->redirectCount == 0)
   {
      url = request->url.data;
   }
   else
   {
      url = request->redirectUrl.data;
   }

   if (MHEG5strncmp(url, "https://", 8) == 0)
   {
      /* Check TLS certificates */
      request->downloadState = STATE_DOWNLOAD_WAIT_CERT;

      certCount = MHEG5GetTlsCertStoreCount();
      if (certCount == 0 || request->certCount == certCount)
      {
         /* Either no certificates, or no new certificates - request
          * next certificate
          */
         status = MHEG5GetNextTlsCertificate();
         if (status == MHEG5_TLS_CERT_REQUEST_SUCCESS)
         {
            /* New certificates obtained (synchronously) */
            request->certCount = MHEG5GetTlsCertStoreCount();
            request->downloadState = STATE_DOWNLOAD_HAS_CERT;
         }
         else if (status == MHEG5_TLS_CERT_REQUEST_FAILURE)
         {
            request->downloadState = STATE_DOWNLOAD_FAILED;
            GenerateStreamRefError(request);
         }
      }
      else if (request->certCount < certCount)
      {
         /* No need to request new certificates */
         request->certCount = certCount;
         request->downloadState = STATE_DOWNLOAD_HAS_CERT;
      }
   }
   else
   {
      /* HTTP requests don't need to check certificates */
      request->downloadState = STATE_DOWNLOAD_HAS_CERT;
   }
}

static void DestroyDownloadRequest(StreamerRequest *request)
{
   MHEG5StopDownloadRequest(request->downloadId);
   MHEG5DestroyDownloadRequest(request->downloadId);

   request->downloadState = STATE_DOWNLOAD_STOPPED;
   request->downloadId = 0;
   request->certCount = 0;
}

static void DestroyMeasurementRequest(StreamerRequest *request)
{
   MHEG5StopMeasureRequest(request->downloadId);
   MHEG5DestroyMeasureRequest(request->downloadId);

   request->downloadState = STATE_DOWNLOAD_STOPPED;
   request->downloadId = 0;
   request->certCount = 0;
}

static void DestroyRequest(StreamerRequest *request)
{
   CounterTrigger *trigger, *next;

   DestroyDownloadRequest(request);
   MHEG5stringDestruct(&request->url);
   MHEG5stringDestruct(&request->redirectUrl);

   trigger = request->triggers;
   while (trigger != NULL)
   {
      next = trigger->next;
      MHEG5freeMem(trigger);
      trigger = next;
   }
   request->triggers = NULL;

   if (request->keys.pid_info != NULL)
   {
      MHEG5freeMem(request->keys.pid_info);
   }
   MHEG5stringDestruct(&request->keyLocation);

   MHEG5freeMem(request);
}

static StreamerRequest* FindRequestById(U32BIT requestId)
{
   StreamerRequest *request;

   request = requestList;

   while (request != NULL && request->requestId != requestId)
   {
      request = request->next;
   }

   return request;
}

static StreamerRequest* FindRequestByObj(MHEG5Root *root)
{
   StreamerRequest *request;

   request = requestList;
   while ((request != NULL) && (request->root != root))
   {
      request = request->next;
   }

   return request;
}

static StreamerRequest* FindRequestByDownloadId(U32BIT downloadId)
{
   StreamerRequest *request;

   request = requestList;

   while (request != NULL && request->downloadId != downloadId)
   {
      request = request->next;
   }

   return request;
}

static void ProcessHttpHeaders(U32BIT downloadId,S32BIT code)
{
   StreamerRequest *request;
   U32BIT units;

   STB_OSMutexLock(requestMutex);

   request = FindRequestByDownloadId(downloadId);
   if (request != NULL)
   {
      PDBG(("%d ~ response code: %d", request->requestId, code));

      if (code == 200 || code == 206)
      {
         if (code == 200)
         {
            /* Range header is not supported, data is returned from the
             * start of the stream. This means that CounterPosition is 0.
             */
            ULL_Set32(request->currentBytePosition, 0);
         }

         DBG(("request %d download state %s", request->requestId,
              REQUEST_DOWNLOAD_STATE_STR(request)));

         /*assert(request->downloadState == STATE_DOWNLOAD_STARTED);*/
         if (request->downloadState != STATE_DOWNLOAD_STARTED)
         {
            PDBG(("%d ~ WARNING: Incorrect download state, overridden",
                  request->requestId));
         }
         request->downloadState = STATE_DOWNLOAD_RUNNING;

         /* Copy bitrate and index file (if any) */
         MHEG5CopyDownloadBitrate(request->downloadId, &request->bitrate);

         /* Calculate expected number of units (if possible) */
         request->contentLength =
            MHEG5GetDownloadContentLength(request->downloadId,
               request->contentLength);
         PDBG(("%d ~ contentLength:", request->requestId));
         ULL_Print(request->contentLength);
         if (ULL_IsValid(request->contentLength))
         {
            /* Convert bytes to units */
            units = ConvertToUnits(request, request->contentLength);
            request->counterMaxPosition = units;
            PDBG(("%d ~ counterMaxPosition = units = %d",
                  request->requestId, units));
            CalculateExpectedBytes(request);
         }

         /* Copy stream keys and locations (if any) */
         MHEG5CopyDownloadStreamKeyInfo(request->downloadId, &request->keys,
            &request->keyLocation);
         HandleStreamKeys(request);

         /* It may be possible to start streaming, if enough data
          * has already been buffered.
          */
         CheckRunCondition(request);
      }
      else
      {
         /* Anything else is handled when the download request is done */
      }
   }

   STB_OSMutexUnlock(requestMutex);
}

static void CalculateExpectedBytes(StreamerRequest *request)
{
   U64BIT endPosition, counterEndPosition;

   ULL_SetInvalid(request->expectedBytes);

   /* Check end position of stream, if known */
   ULL_SetInvalid(endPosition);
   if (ULL_IsValid(request->contentLength))
   {
      /* Initialise to content length */
      PDBG(("%d ~ contentLength is valid", request->requestId));
      endPosition = request->contentLength;
      if (request->counterEndPosition > 0)
      {
         PDBG(("%d ~ counterEndPosition > 0", request->requestId));
         counterEndPosition = ConvertToBytes(request, request->counterEndPosition);
         if (ULL_Compare(counterEndPosition, endPosition) < 0)
         {
            /* CounterEndPosition is before the end of the stream */
            PDBG(("%d ~ counterEndPosition < counterMaxPosition",
                  request->requestId));
            endPosition = counterEndPosition;
         }
      }
   }

   PDBG(("%d ~ endPosition is valid: %s\n", request->requestId,
         ULL_IsValid(endPosition) ? "TRUE" : "FALSE"));
   if (ULL_IsValid(endPosition))
   {
      PDBG(("endPosition(%u) - bytePosition(%u)", ULL_Get32(endPosition), ULL_Get32(request->bytePosition)));
      request->expectedBytes = endPosition;
      ULL_Sub(request->expectedBytes, request->bytePosition);
      PDBG(("%d ~ request->expectedBytes = ", request->requestId));
      ULL_Print(request->expectedBytes);
   }
}

static void HandleStreamKeys(StreamerRequest *request)
{
   /* Only load keys file if they are not available directly from the
    * headers.
    */
   if ((request->keyState == STATE_KEY_UNKNOWN) &&
       (request->keys.num_pids == 0) &&
       (request->keyLocation.len > 0))
   {
      request->keyState = STATE_KEY_PENDING;

      /* donwloadId is used to make sure the file is for the right download.
       * A key is associated with a specific download (but should remain
       * valid for future downloads unless the stream is prepared again).
       */
      request->keyDownloadId = request->downloadId;
      (void)MHEG5FileOrmGet(request->keyLocation,
         FRP_HASH | FRP_CACHE_DEFAULT,
         (void *)request->root,
         StreamKeyLoadedCallback,
         StreamKeyLoadFailCallback);
   }
   else
   {
      request->keyState = STATE_KEY_VALID;
   }
}

static void HandleDownloadFinished(U32BIT downloadId,E_HttpStatus status,S32BIT code)
{
   StreamerRequest *request;

   request = FindRequestByDownloadId(downloadId);
   if (request == NULL)
   {
      PDBG(("Cannot find request from download id %d - already gone",downloadId));
   }
   else
   {
      if (status == HTTP_STATUS_OK)
      {
         if (code == 301)
         {
            /* Permanent redirection */
            HandleDownloadRedirection(request, TRUE);
         }
         else if ((code > 301) && (code <= 307) && (code != 306))
         {
            /* Temporary redirection */
            HandleDownloadRedirection(request, FALSE);
         }
         else if ((code == 200) || (code == 206))
         {
            /* Success */
            HandleDownloadSuccess(request);
         }
         else
         {
            PDBG(("%d ~ HTTP_STATUS_OK code=%d", request->requestId,code));
         #ifdef INCLUDE_FREESAT
            if ((code == 401) || (code == 403) || (code == 407))
            {
               GenerateIPStreamAccessError(request);
            }
            else
            {
               GenerateStreamRefError(request);
            }
         #else  /* INCLUDE_FREESAT */
            GenerateStreamRefError(request);
         #endif  /* INCLUDE_FREESAT */
         }
      }
      else if (status == HTTP_STATUS_PARTIAL)
      {
         PDBG(("%d ~ HTTP_STATUS_PARTIAL", request->requestId));
         HandleDownloadPartial(request,downloadId);
      }
      else if (status == HTTP_STATUS_SSL_ERROR)
      {
         PDBG(("%d ~ HTTP_STATUS_SSL_ERROR", request->requestId));
         request->downloadState = STATE_DOWNLOAD_QUEUED;
         ProgressDownloadRequest(request);
      }
      else
      {
         PDBG(("%d ~ Network error, timeout or other", request->requestId));
         /* Network error, timeout or other - failure */
         /*request->state = STATE_FAILED;*/
         request->downloadState = STATE_DOWNLOAD_FAILED;
         GenerateStreamRefError(request);
      }
   }

   CheckDownloadCondition();
}

static void HandleDownloadRedirection(StreamerRequest *request,
   BOOLEAN permanent)
{
   MHEG5String redirectUrl;

   if (request->redirectCount < MAX_REDIRECTS)
   {
      redirectUrl.data = MHEG5GetDownloadRedirect(request->downloadId);
      if (redirectUrl.data != NULL)
      {
         MHEG5CopyDownloadBitrate(request->downloadId, &request->bitrate);
         MHEG5CopyDownloadStreamKeyInfo(request->downloadId, &request->keys,
            &request->keyLocation);

         MHEG5stringDestruct(&request->redirectUrl);
         redirectUrl.len = strlen((char *)redirectUrl.data);
         if (permanent)
         {
            MHEG5stringDestruct(&request->url);
            request->url = MHEG5convertGID(&redirectUrl);
         }

         request->redirectUrl = MHEG5convertGID(&redirectUrl);
         ++request->redirectCount;

         DestroyDownloadRequest(request);

         request->downloadState = STATE_DOWNLOAD_HAS_ACCESS;
         ProgressDownloadRequest(request);
      }
      else
      {
         PDBG(("%d ~ Cannot find redirection URL", request->requestId));
         request->downloadState = STATE_DOWNLOAD_FAILED;
         GenerateStreamRefError(request);
      }
   }
   else
   {
      PDBG(("%d ~ Too many redirections!", request->requestId));
      request->downloadState = STATE_DOWNLOAD_FAILED;
      GenerateStreamRefError(request);
   }
}

static void HandleDownloadSuccess(StreamerRequest *request)
{
   DestroyDownloadRequest(request);

   if (request->looping == 0 || request->currentLoop < request->looping - 1)
   {
      PDBG(("%d ~ Looping: current loop %d out of %d",
            request->requestId, request->currentLoop, request->looping));
      ++request->currentLoop;
      assert(request->downloadState == STATE_DOWNLOAD_STOPPED);
      CreateDownloadRequest(request);
   }
   else
   {
      request->downloadState = STATE_DOWNLOAD_DONE;
   }

   /* If the request is waiting for a "DOWNLOAD_DONE", it needs to
    * be released from this state
    */
   STB_OSMutexLock(requestMutex);
   CheckRunCondition(request);
   STB_OSMutexUnlock(requestMutex);
}

static void HandleDownloadPartial(StreamerRequest *request,U32BIT downloadId)
{
   U64BIT position;
   BOOLEAN needsRestart;
   needsRestart = MHEG5DownloadPositionPartial(downloadId,&position);
   if (needsRestart)
   {
      PDBG(("%d ~ restart download",request->requestId));

      MHEG5ClearDownloadRequest(downloadId);
      request->bytePosition = position;
      request->isPartial = TRUE;
      request->certCount = 0;
      if (MHEG5QueueIsStreamingEnabled())
      {
         request->downloadState = STATE_DOWNLOAD_QUEUED;
         ProgressDownloadRequest(request);
      }
   }
   else
   {
      DestroyDownloadRequest(request);
      request->downloadState = STATE_DOWNLOAD_DONE;
   }
   STB_OSMutexLock(requestMutex);
   CheckRunCondition(request);
   STB_OSMutexUnlock(requestMutex);
}

static void HandleMeasurementFinished(U32BIT downloadId)
{
   StreamerRequest *request;
   S32BIT status;
   S32BIT code;

   request = FindRequestByDownloadId(downloadId);
   if (request != NULL)
   {
      status = MHEG5GetMeasureStatus(downloadId);
      if (status == HTTP_STATUS_OK)
      {
         code = MHEG5GetMeasureResponseCode(downloadId);
         if ((code >= 301) && (code <= 307) && (code != 306))
         {
            /* Redirection */
            HandleMeasurementRedirection(request);
         }
         else
         {
            /* Success */
            MHEG5ProcessMeasureRequest(downloadId);
         }
      }
      else if (status == HTTP_STATUS_SSL_ERROR)
      {
         PDBG(("%d ~ HTTP_STATUS_SSL_ERROR", request->requestId));
         request->downloadState = STATE_DOWNLOAD_QUEUED;
         ProgressDownloadRequest(request);
      }
      else
      {
         PDBG(("%d ~ Network error, timeout or other", request->requestId));
         /* Network error, timeout or other - failure */
         request->downloadState = STATE_DOWNLOAD_FAILED;
         MHEG5ProcessMeasureRequest(downloadId);
      }
   }
   else
   {
      /* Cannot find request, measurement request must already be gone */
   }
}

static void HandleMeasurementRedirection(StreamerRequest *request)
{
   MHEG5String redirectUrl;

   if (request->redirectCount < MAX_REDIRECTS)
   {
      redirectUrl.data = MHEG5GetMeasureRedirect(request->downloadId);
      if (redirectUrl.data != NULL)
      {
         MHEG5stringDestruct(&request->redirectUrl);
         redirectUrl.len = strlen((char *)redirectUrl.data);
         request->redirectUrl = MHEG5convertGID(&redirectUrl);
         ++request->redirectCount;

         DestroyMeasurementRequest(request);

         request->downloadState = STATE_DOWNLOAD_HAS_ACCESS;
         ProgressDownloadRequest(request);
      }
      else
      {
         PDBG(("%d ~ Cannot find redirection URL", request->requestId));
         request->downloadState = STATE_DOWNLOAD_FAILED;
         MHEG5ProcessMeasureRequest(request->downloadId);
      }
   }
   else
   {
      PDBG(("%d ~ Too many redirections!", request->requestId));
      request->downloadState = STATE_DOWNLOAD_FAILED;
      MHEG5ProcessMeasureRequest(request->downloadId);
   }
}

void MHEG5streamerNotifyStreamingReady(void)
{
   StreamerRequest *request;

   STB_OSMutexLock(requestMutex);

   request = requestList;
   while (request != NULL)
   {
      if (request->outputState == STATE_OUTPUT_PENDING)
      {
         /* This must be done before output state is modified,
          * otherwise the run condition may be checked by other
          * tasks before the components are setup.
          */
         //SetupComponents(request);

         request->outputState = STATE_OUTPUT_SETUP;
         CheckRunCondition(request);
         break;
      }

      request = request->next;
   }

   STB_OSMutexUnlock(requestMutex);
}

static void HandleStreamPlaying(U32BIT requestId)
{
   StreamerRequest *request;

   request = FindRequestById(requestId);
   if (request != NULL)
   {
      switch (request->flowState)
      {
         case STATE_FLOW_STARTED:
            request->flowState = STATE_FLOW_RUNNING;
            GenerateStreamPlaying(request);
            break;
         case STATE_FLOW_UNDERFLOW:
            request->flowState = STATE_FLOW_RUNNING;
            GenerateStreamUnderflowResume();
            break;
         case STATE_FLOW_RUNNING:
            /* Multiple components, playback resumed or confused decoders */
            DBG(("Ignoring StreamStarted in STATE_FLOW_RUNNING"));
            break;
         case STATE_FLOW_STOPPED:
            PDBG(("%d ~ WARNING: Stream started when flow is stopped",
                  request->requestId));
            break;
         default:
            TRACE(TERROR, ("Unhandled flow state"));
            assert(0);
            break;
      }
   }
}

static void HandleStreamStopped(U32BIT requestId)
{
   StreamerRequest *request;

   request = FindRequestById(requestId);
   if (request != NULL)
   {
      PDBG(("%d ~ request->flowState = %s", request->requestId,
            REQUEST_FLOW_STATE_STR(request)));
      switch (request->flowState)
      {
         case STATE_FLOW_RUNNING:
            PDBG(("%d ~ request->downloadState = %s", request->requestId,
                  REQUEST_DOWNLOAD_STATE_STR(request)));
            if (request->downloadState == STATE_DOWNLOAD_RUNNING)
            {
               request->flowState = STATE_FLOW_UNDERFLOW;
               GenerateStreamUnderflow();
            }
            else if (request->downloadState == STATE_DOWNLOAD_DONE)
            {
               request->flowState = STATE_FLOW_STOPPED;
               GenerateStreamStopped(request);
            }
            else if (request->downloadState == STATE_DOWNLOAD_STARTED)
            {
               /* Ignore, it's not the same stream */
            }
            else
            {
               TRACE(TERROR, ("Unhandled download state"));
               assert(0);
            }
            break;
         case STATE_FLOW_STOPPED:
            /* Expected underflow, ignore */
            DBG(("Expected underflow in %s", REQUEST_FLOW_STATE_STR(request)));
            break;
         case STATE_FLOW_UNDERFLOW:
            /* Spurious notification */
            break;
         case STATE_FLOW_STARTED:
            /* This can happen if the flow is starting for a new stream
             * (or a new position in the same stream), without notifying
             * that the old stream is stopped.
             * TODO: Need to think if this is valid
             */
            break;
         default:
            TRACE(TERROR, ("Unhandled flow state"));
            assert(0);
      }
   }
}

static void HandleCounterTrigger(U32BIT requestId, S32BIT triggerId)
{
   StreamerRequest *request;

   request = FindRequestById(requestId);
   if (request != NULL)
   {
      GenerateCounterTrigger(request, triggerId);
   }
}

static void HandleContentAvailable(U32BIT requestId)
{
   StreamerRequest *request;

   request = FindRequestById(requestId);
   if (request != NULL)
   {
      GenerateContentAvailable(request);
   }
}

static void GenerateStreamRefError(StreamerRequest *request)
{
   MHEG5Root *app;

   if (request->root->runningStatus)
   {
      app = (MHEG5Root *)MHEG5getCurrentApplication();
      if (app != NULL)
      {
         PDBG(("%d ~ Generating StreamRefError EngineEvent",
               request->requestId));
         MHEG5sendEvent(app, MHEG5ENGINEEVENT, EE_IC_STREAM_REF_ERROR);
      }
   }
}

#ifdef INCLUDE_FREESAT

static void GenerateIPStreamAccessError(StreamerRequest *request)
{
   MHEG5Root *app;

   if (request->root->runningStatus)
   {
      app = (MHEG5Root *)MHEG5getCurrentApplication();
      if (app != NULL)
      {
         PDBG(("%d ~ Generating IPStreamAccessError EngineEvent",
               request->requestId));
         MHEG5sendEvent(app, MHEG5ENGINEEVENT, EE_IP_STREAM_ACCESS_ERROR);
      }
   }
}

#endif  /* INCLUDE_FREESAT */


static void GenerateStreamUnderflow(void)
{
   MHEG5Root *app;

   app = (MHEG5Root *)MHEG5getCurrentApplication();
   if (app != NULL)
   {
      PDBG(("Generating StreamUnderflow EngineEvent"));
      MHEG5sendEvent(app, MHEG5ENGINEEVENT, EE_IC_STREAM_UNDERFLOW);
   }
}

static void GenerateStreamUnderflowResume(void)
{
   MHEG5Root *app;

   app = (MHEG5Root *)MHEG5getCurrentApplication();
   if (app != NULL)
   {
      PDBG(("Generating StreamUnderflowResume EngineEvent"));
      MHEG5sendEvent(app, MHEG5ENGINEEVENT, EE_IC_STREAM_UNDERFLOW_RESUME);
   }
}

static void GenerateContentAvailable(StreamerRequest *request)
{
   if (!request->contentAvailableGenerated)
   {
      /* Generate event */
      PDBG(("%d ~ Generating ContentAvailable for (%.*s %ld) [%d]",
            request->requestId,
            (int)request->root->grp->groupName.len,
            request->root->grp->groupName.data, request->root->id,
            request->requestId));
      request->contentAvailableGenerated = TRUE;
      MHEG5sendEvent(request->root, MHEG5CONTENTAVAILABLE, 0);
   }
}

static void GenerateStreamPlaying(StreamerRequest *request)
{
   if (!request->streamPlayingGenerated)
   {
      /* Generate event */
      PDBG(("%d ~ Generating StreamPlaying for (%.*s %ld)",
            request->requestId,
            (int)request->root->grp->groupName.len,
            request->root->grp->groupName.data, request->root->id));
      request->streamPlayingGenerated = TRUE;
      MHEG5sendEvent(request->root, MHEG5STREAMPLAYING, 0);
   }
}

static void GenerateStreamStopped(StreamerRequest *request)
{
   if (request->streamPlayingGenerated)
   {
      /* Generate event */
      PDBG(("%d ~ Generating StreamStopped for (%.*s %ld)",
            request->requestId, (int)request->root->grp->groupName.len,
            request->root->grp->groupName.data, request->root->id));
      if (request->terminateStreamSetup && isIcsSetup)
      {
         DVB_MhegICStreamRelease();
         isIcsSetup = FALSE;
      }
      request->streamPlayingGenerated = FALSE;
      MHEG5sendEvent(request->root, MHEG5STREAMSTOPPED, 0);
   }
}

static void GenerateCounterTrigger(StreamerRequest *request, S32BIT triggerId)
{
   /* Generate event */
   PDBG(("%d ~ Generating CounterTrigger for (%.*s %ld), triggerId = %d",
         request->requestId, (int)request->root->grp->groupName.len,
         request->root->grp->groupName.data, request->root->id, triggerId));
   MHEG5sendEvent(request->root, MHEG5COUNTERTRIGGER, triggerId);
}

static void GenerateKeyFileError(void)
{
   MHEG5Root *app;

   app = (MHEG5Root *)MHEG5getCurrentApplication();
   if (app != NULL)
   {
      PDBG(("Generating KeyFileError EngineEvent"));
      MHEG5sendEvent(app, MHEG5ENGINEEVENT, EE_IC_KEY_FILE_ERROR);
   }
}

static void NotifyContentAvailable(U32BIT requestId)
{
   MHEG5eventMessage_t event;
   E_MhegErr cqu_err;

   event.proc_msg_func = (F_MSG_PROCESS)MHEG5StreamerHandle;
   event.data_type = DT_VALUE;
   event.data.streamer.requestId = requestId;
   event.data.streamer.eventType = MHEG5_STREAMER_EVENT_CONTENT_AVAILABLE;

   PDBG(("%d ~ Sending MHEG5_STREAMER_EVENT_CONTENT_AVAILABLE", requestId));
   cqu_err = VQ_PutMsg(&event, PRTY_NORMAL);
   if (cqu_err != MHERR_OK)
   {
      TRACE(TERROR, ("Cannot generate event %d", cqu_err));
   }
}

static void NotifyStreamPlaying(U32BIT requestId)
{
   MHEG5eventMessage_t event;
   E_MhegErr cqu_err;

   event.proc_msg_func = (F_MSG_PROCESS)MHEG5StreamerHandle;
   event.data_type = DT_VALUE;
   event.data.streamer.requestId = requestId;
   event.data.streamer.eventType = MHEG5_STREAMER_EVENT_TASK_STREAM_PLAYING;

   PDBG(("%d ~ Sending MHEG5_STREAMER_EVENT_TASK_STREAM_PLAYING", requestId));
   cqu_err = VQ_PutMsg(&event, PRTY_NORMAL);
   if (cqu_err != MHERR_OK)
   {
      TRACE(TERROR, ("Cannot generate event %d", cqu_err));
   }
}

static void NotifyStreamStopped(U32BIT requestId)
{
   MHEG5eventMessage_t event;
   E_MhegErr cqu_err;

   event.proc_msg_func = (F_MSG_PROCESS)MHEG5StreamerHandle;
   event.data_type = DT_VALUE;
   event.data.streamer.requestId = requestId;
   event.data.streamer.eventType = MHEG5_STREAMER_EVENT_TASK_STREAM_STOPPED;

   PDBG(("%d ~ Sending MHEG5_STREAMER_EVENT_TASK_STREAM_STOPPED", requestId));
   cqu_err = VQ_PutMsg(&event, PRTY_NORMAL);
   if (cqu_err != MHERR_OK)
   {
      TRACE(TERROR, ("Cannot generate event %d", cqu_err));
   }
}

static void ItemInsertCallback(U32BIT downloadId, U64BIT base,
   U64BIT position, U32BIT len, BOOLEAN last)
{
   StreamerRequest *request;

   STB_OSMutexLock(requestMutex);

   request = FindRequestByDownloadId(downloadId);
   if (request != NULL)
   {
      assert(ULL_IsEqual(request->bytePosition, base));

      request->downloadedBytes = position;
      ULL_Add32(request->downloadedBytes, len);

      if (!request->contentAvailableNotified)
      {
         /* Item buffered, ContentAvailable not generated yet */
         request->contentAvailableNotified = TRUE;
         NotifyContentAvailable(request->requestId);
      }

      /* Check if the PMT is available */
      if (!request->pmtRetrieved)
      {
         request->pmtRetrieved = MHEG5GetDownloadPmt(request->downloadId,
               request->pmt);
      }

      /* Check termination (max/end position) */
      if ((ULL_IsValid(request->expectedBytes)) &&
          (ULL_Compare(request->downloadedBytes,
              request->expectedBytes) >= 0))
      {
         MHEG5StopDownloadRequestAsync(request->downloadId);
      }

      /* It may be possible to start streaming, if enough data
       * has already been buffered.
       */
      CheckRunCondition(request);
   }

   STB_OSMutexUnlock(requestMutex);
}

static void ItemReleaseCallback(U32BIT requestId, U64BIT base,
   U64BIT position, U32BIT len, BOOLEAN last)
{
   StreamerRequest *request;

   STB_OSMutexLock(requestMutex);

   request = FindRequestById(requestId);
   if (request != NULL)
   {
      DBG(("%d ~ request->outputState = %s", request->requestId,
           REQUEST_OUTPUT_STATE_STR(request)));
      if (request->outputState == STATE_OUTPUT_READY)
      {
         /* Stream started playing */
         request->outputState = STATE_OUTPUT_RUNNING;
         NotifyStreamPlaying(request->requestId);
      }
      else if (request->outputState == STATE_OUTPUT_STARVED)
      {
         /* Stream started playing */
         request->outputState = STATE_OUTPUT_RUNNING;
         NotifyStreamPlaying(request->requestId);
      }
      if (request->outputState == STATE_OUTPUT_RUNNING)
      {
         request->currentBytePosition = base;
         ULL_Add(request->currentBytePosition, position);
         ULL_Add32(request->currentBytePosition, len);

         request->presentedBytes = position;
         ULL_Add32(request->presentedBytes, len);

         /* Check triggers in for any position covered by this block */
         CheckCounterTriggers(request, base, position, len);
      }
   }
   STB_OSMutexUnlock(requestMutex);
}

static void CheckCounterTriggers(StreamerRequest *request, U64BIT base,
   U64BIT position, U32BIT len)
{
   CounterTrigger *trigger;
   MHEG5eventMessage_t event;
   E_MhegErr cqu_err;
   U64BIT itemPosition;
   U64BIT itemLimit;
   U64BIT triggerPosition;

   itemPosition = base;
   ULL_Add(itemPosition, position);

   itemLimit = itemPosition;
   ULL_Add32(itemLimit, len);

   trigger = request->triggers;
   while (trigger != NULL)
   {
      triggerPosition = ConvertToBytes(request, trigger->position);
      if ((ULL_Compare(triggerPosition, itemPosition) >= 0) &&
          (ULL_Compare(triggerPosition, itemLimit) < 0))
      {
         /* Generate trigger event */

         event.proc_msg_func = (F_MSG_PROCESS)MHEG5StreamerHandle;
         event.data_type = DT_VALUE;
         event.data.streamer.requestId = request->requestId;
         event.data.streamer.eventType = MHEG5_STREAMER_EVENT_COUNTER_TRIGGER;
         event.data.streamer.triggerId = trigger->id;

         PDBG(("%d ~ Sending MHEG5_STREAMER_EVENT_COUNTER_TRIGGER",
               request->requestId));
         cqu_err = VQ_PutMsg(&event, PRTY_NORMAL);
         if (cqu_err != MHERR_OK)
         {
            TRACE(TERROR, ("Cannot generate event %d", cqu_err));
         }
      }
      trigger = trigger->next;
   }
}

static void SetupComponents(StreamerRequest *request,S_ICSComponents *components)
{
   MHEG5Stream *stream;
   MHEG5Ingredient *item;
   MHEG5Video *video;
   MHEG5Audio *audio;
   MHEG5Int index;

   components->audio_active = FALSE;
   components->audio_component_tag = -1;
   components->video_active = FALSE;
   components->video_component_tag = -1;
   components->video_termination = MHEG5_VIDEO_TERMINATION_DISAPPEAR;

   stream = (MHEG5Stream *)request->root;
   for (index = 0; index != MAX_MLTPLX; index++)
   {
      item = stream->multiplex[index];
      if (item == NULL)
      {
         break;
      }
      if (!item->root.runningStatus)
      {
         continue;
      }

      if (item->root.clazz == MHEG5AUDIO && !components->audio_active)
      {
         audio = (MHEG5Audio *)item;
         components->audio_active = TRUE;
         components->audio_component_tag = audio->componentTag;
         DBGTRACE(TSTRM|TICS,"Found active audio component: tag = %d",
              components->audio_component_tag);
         continue;
      }
      if (item->root.clazz == MHEG5VIDEO && !components->video_active)
      {
         video = (MHEG5Video *)item;
         components->video_active = TRUE;
         components->video_component_tag = video->componentTag;
         if (video->terminationFreeze)
         {
            components->video_termination = MHEG5_VIDEO_TERMINATION_FREEZE;
         }
         DBGTRACE(TSTRM|TICS,"Found active video component: tag = %d, freeze = %s",
              components->video_component_tag,
              components->video_termination == MHEG5_VIDEO_TERMINATION_FREEZE ? "TRUE" : "FALSE");
         continue;
      }
   }
   request->terminateStreamSetup = (components->video_termination == MHEG5_VIDEO_TERMINATION_DISAPPEAR)? TRUE : FALSE;
}

static BOOLEAN ParseAudio( U8BIT *dptr, U8BIT *dend, S_ICSLangInfo *info_ptr )
{
   U8BIT dlen,dtag;
   BOOLEAN result = FALSE;
   while ( dptr < dend )
   {
      dtag = *dptr++;
      dlen = *dptr++;
      DBGTRACE(TSTRM,"dtag=%d dlen=%d",dtag,dlen)
      switch ( dtag )
      {
         default:
            break;
         case ISO_LANG_DTAG:
            DBGTRACE(TSTRM,"lang=%c%c%c atyp=%x",dptr[0],dptr[1],dptr[2],dptr[3])
            info_ptr->lang_code = dptr[0] << 16 | dptr[1] << 8 | dptr[2] | 0x202020; // lower case
            info_ptr->a_type = dptr[3];
            result = TRUE;
            break;
         case STREAMID_DTAG:
            DBGTRACE(TSTRM,"stag=%x",dptr[0])
            break;
         case SUBTITLE_DTAG:
            DBGTRACE(TERROR,"lang=%c%c%c styp=%x",dptr[0],dptr[1],dptr[2],dptr[3])
            break;
      }
      dptr += dlen;
   }
   return result;
}

static BOOLEAN ParseSubts( U8BIT *dptr, U8BIT *dend, S_ICSLangInfo *info_ptr )
{
   U8BIT dlen,dtag;
   BOOLEAN result = FALSE;
   while ( dptr < dend )
   {
      dtag = *dptr++;
      dlen = *dptr++;
      DBGTRACE(TSTRM,"dtag=%d dlen=%d",dtag,dlen)
      switch ( dtag )
      {
         default:
            break;
         case ISO_LANG_DTAG:
            DBGTRACE(TERROR,"lang=%c%c%c atyp=%x",dptr[0],dptr[1],dptr[2],dptr[3])
            break;
         case STREAMID_DTAG:
            DBGTRACE(TSTRM,"stag=%x",dptr[0])
            break;
         case SUBTITLE_DTAG:
            DBGTRACE(TSTRM,"lang=%c%c%c styp=%x",dptr[0],dptr[1],dptr[2],dptr[3])
            info_ptr->lang_code =  dptr[0] << 16 | dptr[1] << 8 | dptr[2] | 0x202020; // lower case
            info_ptr->s_type = dptr[3];
            info_ptr->u.page.composition = (dptr[4] << 8) | dptr[5];
            info_ptr->u.page.ancillary = (dptr[6] << 8) | dptr[7];
            result = TRUE;
            break;
      }
      dptr += dlen;
   }
   return result;
}

static BOOLEAN HasStreamId( U8BIT *dptr, U8BIT *dend, U8BIT id )
{
   BOOLEAN found = FALSE;
   U8BIT dlen,dtag;
   while ( dptr < dend )
   {
      dtag = *dptr++;
      dlen = *dptr++;
      if (dtag == STREAMID_DTAG && *dptr == id)
      {
         found = TRUE;
         break;
      }
      dptr += dlen;
   }
   return found;
}

static S_ICSPidKeys* SetupKeys(S_ICSPidInfo *pid_info, U16BIT num_pids, U16BIT pid, S_ICSPidKeys* pkeys)
{
   U16BIT i;
   //DBGTRACE((TICS|TSTRM),"num_pids=%u pid=%x",num_pids,pid)
   for (i = 0; i != num_pids; i++)
   {
      //DBGTRACE((TICS|TSTRM),"[%d]PID=%x",i,pid_info[i].PID)
      if (pid_info[i].PID == pid)
      {
         memcpy(pkeys->even, pid_info[i].even_key, 16);
         memcpy(pkeys->odd, pid_info[i].odd_key, 16);
         memcpy(pkeys->even + 16, pid_info[i].iv, 16);
         memcpy(pkeys->odd + 16, pid_info[i].iv, 16);
         break;
      }
   }
   return (i != num_pids)? pkeys : NULL;
}

static void StartStreaming(StreamerRequest *request)
{
   U32BIT pref_lang;
   U32BIT subt_lang;
   U16BIT dlen, pid;
   U8BIT type;
   S_StreamPids pids;
   S_ICSLangInfo *info_ptr;
   U8BIT *sptr;
   U8BIT *eptr;
   U8BIT *data = request->pmt;
   S_ICSComponents components;
   S_ICSPidKeys video_keys;
   S_ICSPidKeys audio_keys;
   E_ICSAudioCodec audcodec = ICS_AC_NONE;
   E_ICSVideoCodec vidcodec = ICS_VC_NONE;

   SetupComponents(request,&components);
   memset(&pids,0,sizeof(S_StreamPids));
   DVB_MhegPrefAudioLangs(&pref_lang,1);
   DBGTRACE((TICS|TSTRM),"PREF lang=%x",pref_lang)
   if (!DVB_MhegPrefSubtitleLangs(&subt_lang,1))
   {
      subt_lang = UND_LANGUAGE_CODE;
   }

   dlen = (data[1] << 8 | data[2]) & 0xfff;
   eptr = data + dlen - 4; // -4 for crc
   pids.pcr_pid = (data[8] << 8 | data[9]) & 0x1fff;
   dlen = (data[10] << 8 | data[11]) & 0xfff;
   data += 12 + dlen;
   sptr = data;
   while (sptr < eptr)
   {
      type = *sptr;
      dlen = ((sptr[3] & 0x0f) << 8) | sptr[4];
      switch( *sptr )
      {
         default:
            break;
         case ST_PES_PKT:
            if (components.video_active)
            {
               pids.info_num++;
            }
            break;
         case ST_HEAAC:
         case ST_AAC_ADTS:
         case ST_DOLBY_BRAY:
         case ST_DOLBY_ATSC:
            if (components.audio_active)
            {
               pids.info_num++;
            }
            break;
      }
      sptr += 5 + dlen;
   }
   if (pids.info_num)
   {
      pids.info_ptr = (S_ICSLangInfo*)STB_MemAlloc(sizeof(S_ICSLangInfo)*pids.info_num);
      if (pids.info_ptr == NULL)
      {
         pids.info_num = 0;
      }
      else
      {
         memset(pids.info_ptr,0,sizeof(S_ICSLangInfo)*pids.info_num);
      }
   }
   info_ptr = pids.info_ptr;
   while (data < eptr)
   {
      type = *data;
      pid = ((data[1] & 0x1f) << 8) | data[2];
      dlen = ((data[3] & 0x0f) << 8) | data[4];
      data += 5;
      switch( type )
      {
         default:
            DBGTRACE(TERROR,"Unsupported stream type=0x%x",type)
            break;
         case ST_PES_PKT:
            DBGTRACE(TSTRM,"PES type pid=%d len=%d",pid,dlen)
            if (components.video_active && info_ptr)
            {
               if (ParseSubts(data,data+dlen,info_ptr))
               {
                  info_ptr->pid = pid;
                  if (pids.subtitle_pid == 0 || info_ptr->lang_code == subt_lang)
                  {
                     pids.subtitle_pid = pid;
                  }
                  info_ptr++;
               }
               else
               {
                  pids.info_num--;
               }
            }
            else
            {
               DBGTRACE(TERROR,"cactive=%u iptr=%p",components.video_active,info_ptr)
            }
            break;
         case ST_H264:
            DBGTRACE(TSTRM,"H264 type pid=%d len=%d",pid,dlen)
            if (components.video_active)
            {
               if (components.video_component_tag == -1)
               {
                  vidcodec = ICS_VC_H264;
                  pids.video_pid = pid;
               }
               else if (HasStreamId(data,data+dlen,(U8BIT)components.video_component_tag))
               {
                  vidcodec = ICS_VC_H264;
                  pids.video_pid = pid;
               }
            }
            else
            {
               DBGTRACE(TERROR,"cactive=%u iptr=%p",components.video_active,info_ptr)
            }
            break;
         case ST_AAC_ADTS:
         case ST_HEAAC:
         case ST_DOLBY_BRAY:
         case ST_DOLBY_ATSC:
            DBGTRACE(TSTRM,"type=%d pid=%d len=%d",type,pid,dlen)
            if (components.audio_active && info_ptr)
            {
               info_ptr->pid = pid;
               switch( type )
               {
                  case ST_AAC_ADTS:
                  info_ptr->u.codec = ICS_AC_AAC_ADTS;
                  break;
                  case ST_HEAAC:
                  info_ptr->u.codec = ICS_AC_HE_AAC_LATM;
                  break;
                  case ST_DOLBY_BRAY:
                  case ST_DOLBY_ATSC:
                  info_ptr->u.codec = ICS_AC_EAC3;
                  break;
               }
               if (components.audio_component_tag == -1)
               {
                  if ((ParseAudio(data,data+dlen,info_ptr) && info_ptr->lang_code == pref_lang) || pids.audio_pid == 0)
                  {
                     DBGTRACE((TICS|TSTRM),"INFO lang=%x Aud-PID=%x",info_ptr->lang_code,pids.audio_pid)
                     audcodec = info_ptr->u.codec;
                     pids.audio_pid = pid;
                  }
               }
               else if (HasStreamId(data,data+dlen,(U8BIT)components.audio_component_tag))
               {
                  audcodec = info_ptr->u.codec;
                  pids.audio_pid = pid;
               }
               info_ptr++;
            }
            else
            {
               DBGTRACE(TERROR,"cactive=%u iptr=%p",components.audio_active,info_ptr)
            }
            break;
      }
      data += dlen;
   }

   DVB_MhegICStreamSetup( pids,
      SetupKeys(request->keys.pid_info, request->keys.num_pids, pids.audio_pid, &audio_keys),
      SetupKeys(request->keys.pid_info, request->keys.num_pids, pids.video_pid, &video_keys),
      components.video_termination, audcodec, vidcodec );
}

static void CheckRunCondition(StreamerRequest *request)
{
   DBG(("%d ~ CheckRunCondition: streamState = %s", request->requestId,
        REQUEST_STREAM_STATE_STR(request)));
   if (request->streamState == STATE_STREAM_RUNNING)
   {
      DBG(("%d ~ request->outputState = %s", request->requestId,
           REQUEST_OUTPUT_STATE_STR(request)));
      DBG(("%d ~ request->downloadState = %s", request->requestId,
           REQUEST_DOWNLOAD_STATE_STR(request)));
      DBG(("%d ~ request->keyState = %s", request->requestId,
           REQUEST_KEY_STATE_STR(request)));
      DBG(("%d ~ request->pmtRetrieved = %s", request->requestId,
           request->pmtRetrieved ? "TRUE" : "FALSE"));

      /* If output is stopped but we can setup ICS, go ahead */
      if ((request->outputState == STATE_OUTPUT_STOPPED) &&
          (request->downloadState != STATE_DOWNLOAD_STARTED) &&
          (request->keyState == STATE_KEY_VALID) &&
          (request->pmtRetrieved))
      {
         PDBG(("%d ~ Output stopped, ICS ready", request->requestId));
         if (isIcsSetup)
         {
            DVB_MhegICStreamRelease();
         }
         isIcsSetup = TRUE;
         StartStreaming(request);
         request->outputState = STATE_OUTPUT_SETUP;
      }

      if (request->outputState == STATE_OUTPUT_SETUP)
      {
         /* Check if there's enough data to start streaming */
         PDBG(("%d ~ Output is setup, check buffer / download",
               request->requestId));
         if ((IsEnoughDataBuffered(request)) ||
             (request->downloadState == STATE_DOWNLOAD_DONE))
         {
            PDBG(("%d ~ Start streaming", request->requestId));
            request->outputState = STATE_OUTPUT_READY;
            request->flowState = STATE_FLOW_STARTED;

            if (!request->paused)
            {
               MHEG5QueueEnableStreaming(request->requestId);
               MHEG5StreamerSendSignalToTask();
               if (request->isPartial)
               {
                  request->downloadState = STATE_DOWNLOAD_QUEUED;
                  ProgressDownloadRequest(request);
               }
            }
         }
      }
      else if (request->outputState == STATE_OUTPUT_STARVED)
      {
         /* Check if there's any data to continue streaming */
         PDBG(("%d ~ Output is starved, check buffer / download",
               request->requestId));
         if ((IsEnoughDataBuffered(request)) ||
             (request->downloadState == STATE_DOWNLOAD_DONE))
         {
            /*request->outputState = STATE_OUTPUT_RUNNING;*/
            PDBG(("%d ~ Starved output resumed", request->requestId));
            ResumePlayback(request);
            MHEG5StreamerSendSignalToTask();
         }
      }
   }
}

static BOOLEAN IsEnoughDataBuffered(StreamerRequest *request)
{
#if 0
   U64BIT expectedBytes;
   char buffer[24];
#endif
   U32BIT bufferedBytes;
   BOOLEAN enough = FALSE;

   bufferedBytes = MHEG5QueueGetBufferedBytes();
#if 0
   if (ULL_IsValid(request->expectedBytes))
   {
      ULL_Itoa(request->expectedBytes, buffer);
      PDBG(("%d ~ request->expectedBytes = %s", request->requestId, buffer));
      expectedBytes = request->expectedBytes;
      ULL_Sub(expectedBytes, request->downloadedBytes);
      ULL_Itoa(expectedBytes, buffer);
      PDBG(("%d ~ expectedBytes = %s", request->requestId, buffer));
   }
   else
   {
      /* Unknown - always allow */
      PDBG(("%d ~ Unknown expected: allow"));
      ULL_Set32(expectedBytes, 0);
   }

   PDBG(("%d ~ bufferedBytes = %d, expectedBytes %s bufferedBytes",
         request->requestId, bufferedBytes,
         bufferedBytes >= MIN_BUFFERED_BYTES ? "?" :
         ULL_Compare32(expectedBytes, bufferedBytes) <= 0 ? "<=" : ">"));
   if ((bufferedBytes >= MIN_BUFFERED_BYTES) ||
       (ULL_Compare32(expectedBytes, bufferedBytes) <= 0))
   {
      enough = TRUE;
   }
#else
   if (bufferedBytes >= MIN_BUFFERED_BYTES)
   {
      enough = TRUE;
   }
#endif

   return enough;
}

static U32BIT ConvertToUnits(StreamerRequest *request, U64BIT bytes)
{
   U64BIT tmp;
   U32BIT units;

   if (request->bitrate > 0)
   {
      /* Known bitrate: t_current = b_current * 1000 / bytesPerSecond */
      tmp = bytes;
      ULL_Mul32(tmp, 1000);
      ULL_Div32(tmp, request->bitrate);

      units = ULL_Get32(tmp);
   }
   else  /* TODO: Index file needs to be supported */
   {
      /* Unknown bitrate */
      tmp = bytes;
      ULL_Div32(tmp, STMR_TS_PACKET_SIZE);

      units = ULL_Get32(tmp);
   }

   return units;
}

static U64BIT ConvertToBytes(StreamerRequest *request, U32BIT units)
{
   U64BIT bytes;
   U32BIT mod;

   if (request->bitrate > 0)
   {
      /* Known bitrate: b_required = t_required * bytesPerSecond / 1000 */
      ULL_Set32(bytes, units);
      ULL_Mul32(bytes, request->bitrate);
      ULL_Div32(bytes, 1000);

      /* Round to nearest multiple of TS packet size */
      mod = ULL_Mod32(bytes, STMR_TS_PACKET_SIZE);
      if (mod < 94)
      {
         ULL_Sub32(bytes, mod);
      }
      else
      {
         ULL_Add32(bytes, STMR_TS_PACKET_SIZE - mod);
      }
   }
   else  /* TODO: Index file needs to be supported */
   {
      /* Unknown bitrate */
      ULL_Set32(bytes, units);
      ULL_Mul32(bytes, STMR_TS_PACKET_SIZE);
   }
   return bytes;
}

static BOOLEAN IsStreamRunning(void)
{
   StreamerRequest *request;
   BOOLEAN running;

   running = FALSE;
   request = requestList;
   while (request != NULL)
   {
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         /* Found a running stream */
         running = TRUE;
         break;
      }
      request = request->next;
   }

   return running;
}

static void CheckDownloadCondition(void)
{
   StreamerRequest *request;
   StreamerRequest *primaryRequest = NULL;
   BOOLEAN secondaryDownloading = FALSE;
   U32BIT minPrepareNumber = 0;
   StreamerRequest *nextPrepare = NULL;

   /* Gather some information about primary and secondary streams */
   request = requestList;
   while (request != NULL)
   {
      if (request->streamState == STATE_STREAM_RUNNING)
      {
         /* Running stream is always primary */
         primaryRequest = request;
      }
      else
      {
         if ((request->downloadState != STATE_DOWNLOAD_STOPPED) &&
             (request->downloadState != STATE_DOWNLOAD_DONE))
         {
            /* If any secondary stream is downloading, others should
             * not be downloaded at the same time.
             */
            secondaryDownloading = TRUE;
         }

         if ((minPrepareNumber == 0) ||
             (request->prepareNumber < minPrepareNumber))
         {
            if ((request->streamState == STATE_STREAM_PREPARED) &&
                (request->downloadState == STATE_DOWNLOAD_STOPPED))
            {
               minPrepareNumber = request->prepareNumber;
               nextPrepare = request;
            }
         }
      }
      request = request->next;
   }

   PDBG(("primaryRequest = %p, nextPrepare = %p", primaryRequest,
         nextPrepare));

   if ((primaryRequest != NULL) &&
       (primaryRequest->downloadState == STATE_DOWNLOAD_STOPPED))
   {
      /* Primary request should be downloading, any other download should
       * stop.
       */
      PDBG(("Downloading primary, stopping any secondary"));
      request = requestList;
      while (request != NULL)
      {
         if (request != primaryRequest)
         {
            PDBG(("Destroying download request for %d (secondary)",
                  request->requestId));
            DestroyDownloadRequest(request);
            MHEG5QueueReleaseRequestItems(request->requestId);
         }
         request = request->next;
      }

      PDBG(("Creating download request for %d (primary)",
            primaryRequest->requestId));

      primaryRequest->currentLoop = 0;
      CreateDownloadRequest(primaryRequest);
   }
   else if ((primaryRequest == NULL) ||
            (primaryRequest->downloadState == STATE_DOWNLOAD_DONE) ||
            (primaryRequest->downloadState == STATE_DOWNLOAD_FAILED))
   {
      /* Primary request download is done. Next prepared stream can
       * start downloading, if other secondary requests are not
       * currently download (only one download at a time is permitted).
       */
      if (nextPrepare != NULL && !secondaryDownloading)
      {
         PDBG(("Creating download request for %d (secondary)",
               nextPrepare->requestId));
         nextPrepare->currentLoop = 0;
         CreateDownloadRequest(nextPrepare);
      }
   }
}

static void StopContentOutput(StreamerRequest *request, BOOLEAN flush)
{
   if (flush)
   {
      MHEG5QueueReleaseRequestItems(request->requestId);
   }

   /* If the stream is running */
   if (request->streamState == STATE_STREAM_RUNNING)
   {
      request->flowState = STATE_FLOW_STOPPED;
      request->outputState = STATE_OUTPUT_STOPPED;
   }
}

static void ResumePlayback(StreamerRequest *request)
{
   /* This function may be called more than once when underflow is
    * resumed.
    */
   if (request->pauseRequestCount > 0)
   {
      --request->pauseRequestCount;
      if (request->pauseRequestCount == 0)
      {
         DVB_MhegICStreamResume();
      }
   }

   if (isIcsSetup)
   {
      MHEG5QueueEnableStreaming(request->requestId);
      if (request->isPartial)
      {
         request->downloadState = STATE_DOWNLOAD_QUEUED;
         ProgressDownloadRequest(request);
      }
   }
}

static BOOLEAN IsValidUrl(MHEG5String *url)
{
   if (((url->len >= 7) && (MHEG5strncmp(url->data, "http://", 7) == 0)) ||
       ((url->len >= 8) && (MHEG5strncmp(url->data, "https://", 8) == 0)))
   {
      return TRUE;
   }
   return FALSE;
}

static void TimerCallback(BOOLEAN triggered, void *callerRef, H_Timer timerHandle)
{
   U_PARAM request;
   request.ptr = callerRef;
   PDBG(("%d",STB_OSGetClockMilliseconds()));
   NotifyStreamStopped(request.u32);
}

static void UnderflowCallback(U32BIT requestId,U32BIT playoutTime)
{
   StreamerRequest *request;
   H_Timer timerHandle;

   STB_OSMutexLock(requestMutex);
   request = FindRequestById(requestId);
   if (request != NULL)
   {
      PDBG(("UnderflowCallback: request->outputState = %s",
            REQUEST_OUTPUT_STATE_STR(request)));
      if (request->outputState == STATE_OUTPUT_RUNNING)
      {
         if ((request->downloadState == STATE_DOWNLOAD_DONE) &&
             (ULL_IsValid(request->expectedBytes)) &&
             (ULL_Compare(request->presentedBytes,request->expectedBytes) >= 0) &&
             (request->currentLoop == request->looping - 1))
         {
            U_PARAM rqstId;
            rqstId.u32 = requestId;
            PDBG(("%d ~ STATE_OUTPUT_DONE", request->requestId));
            request->outputState = STATE_OUTPUT_DONE;
            request->restart = TRUE;

            if (playoutTime == 0 ||
               VT_TimerCreate(playoutTime,(F_TimerCallback)TimerCallback,rqstId.ptr,&timerHandle) != MHERR_OK)
            {
               NotifyStreamStopped(requestId);
            }
         }
         else
         {
            request->outputState = STATE_OUTPUT_STARVED;
            if (request->flowState != STATE_FLOW_UNDERFLOW)
            {
               request->pauseRequestCount++;
               if (request->pauseRequestCount == 1)
               {
                  DVB_MhegICStreamPause();
                  MHEG5QueueDisableStreaming();
               }

               MHEG5ClearDownloadThrottling();
               NotifyStreamStopped(request->requestId);
            }
         }
      }
      else
      {
         /* Not running, cannot be starved */
         PDBG(("%d ~ UnderflowCallback: output state = %s",
               request->requestId, REQUEST_OUTPUT_STATE_STR(request)));
      }
   }
   STB_OSMutexUnlock(requestMutex);
}

static void StreamKeyLoadedCallback(void *userData, S_CONTENT *content)
{
   StreamerRequest *request;
   U8BIT *data;
   U32BIT size;

   PDBG((" "));

   STB_OSMutexLock(requestMutex);
   request = FindRequestByObj(userData);
   if ((request != NULL) &&
       ((request->downloadId == request->keyDownloadId) ||
        (request->downloadState == STATE_DOWNLOAD_DONE)))
   {
      assert(request->keys.pid_info == NULL);

      /* HACK! This is a hack to pass ICS AES tests, but actually the
       * file format is wrong and this needs to be removed.
       */
      data = content->data;
      size = content->size;
      if (content->size > 8 && memcmp(content->data, "X-Keys: ", 8) == 0)
      {
         data = content->data + 8;
         content->size -= 8;
      }

      MHEG5parseStreamKeys(data, (U16BIT)size, &request->keys);
      if (request->keys.num_pids == 0)
      {
         GenerateKeyFileError();
         request->keyState = STATE_KEY_ERROR;
      }
      else
      {
         request->keyState = STATE_KEY_VALID;
         CheckRunCondition(request);
      }
   }
   STB_OSMutexUnlock(requestMutex);
}

static void StreamKeyLoadFailCallback(void *userData)
{
   StreamerRequest *request;

   PDBG((" "));

   request = FindRequestByObj(userData);
   if ((request != NULL) &&
       ((request->downloadId == request->keyDownloadId) ||
        (request->downloadState == STATE_DOWNLOAD_DONE)))
   {
      GenerateKeyFileError();
      request->keyState = STATE_KEY_ERROR;
   }
}

static void ServerAccessCallback(void)
{
   StreamerRequest *request;

   PDBG((" "));

   request = requestList;
   while (request != NULL)
   {
      if (request->downloadState == STATE_DOWNLOAD_WAIT_ACCESS)
      {
         request->downloadState = STATE_DOWNLOAD_CHECK_ACCESS;
         processRequests = TRUE;
         MH5GlueAddPostProcessFunc( MHEG5StreamerProcess );
      }
      request = request->next;
   }
}

static void TlsCertCallback(void)
{
   StreamerRequest *request;

   PDBG((" "));

   request = requestList;
   while (request != NULL)
   {
      if (request->downloadState == STATE_DOWNLOAD_WAIT_CERT)
      {
         request->downloadState = STATE_DOWNLOAD_QUEUED;
         processRequests = TRUE;
         MH5GlueAddPostProcessFunc( MHEG5StreamerProcess );
      }
      request = request->next;
   }
}

#endif /*INCLUDE_ICS*/