StreamClientPull.cpp

#include "Headers.h"
#include "StreamClientPull.h"
#include "Rand.h"
#include "Debug.h"
#include "Console.h"
#include "PacketStreamFrame.h"

CStreamClientPull::CStreamClientPull(
    CSimHandler*                                                        sim,
    CInfoPull*                                                          info,
    CAddress                                                            address,
    IDelegate5<void, __uint16, __uint16, CAddress, __byte, CPacket*>*   delegateSend,
    IDelegate2<void, CAddress, __uint32>*                               delegateIgmpJoin,
    IDelegate1<void, CAddress>*                                         delegateIgmpLeave,
    IDelegate0<CLink*>*                                                 delegateLink,
    __uint32                                                            entry
    ) : CStreamClient(sim)
{
    assert(info);
    
    assert(delegateSend);
    assert(delegateIgmpJoin);
    assert(delegateIgmpLeave);
    assert(delegateLink);

    // Set the initial state
    this->stateClient = STOPPED;
    this->stateRegistration = NOT_REGISTERED;
    this->stateSchedule = SCHEDULE_IDLE;

    // Set simulator
    this->info = info;
    this->address = address;

    // Set the channel
    this->channel = NULL;

    // Create the buffers
    this->bufferMcast = new CStreamBuffer(
        this->info->StreamBufferMcastSize(),
        this->info->StreamBufferMcastSizeHistory()
        );
    this->bufferUcast = new CStreamBufferSegment(
        this->info->StreamBufferUcastSize(),
        this->info->StreamBufferUcastSizeHistory(),
        this->info->StreamSegmentSize()
        );

    // Multicast session
    this->entry = entry;

    // Create the timers
    this->timerPlayMcast = new CTimer<CStreamClientPull>(
        this->sim,
        this,
        &CStreamClientPull::TimerPlayMcast
        );
    this->timerPlayUcast = new CTimer<CStreamClientPull>(
        this->sim,
        this,
        &CStreamClientPull::TimerPlayUcast
        );
    this->timerSchedule = new CTimer<CStreamClientPull>(
        this->sim,
        this,
        &CStreamClientPull::TimerSchedule
        );
    this->timerRegister = new CTimer<CStreamClientPull>(
        this->sim,
        this,
        &CStreamClientPull::TimerRegister
        );

    // Set the lower layer delegates
    this->delegateIgmpJoin = delegateIgmpJoin;
    this->delegateIgmpLeave = delegateIgmpLeave;
    this->delegateSend = delegateSend;
    this->delegateLink = delegateLink;

    // Delegates
    this->delegateRecvFrameMcast = new Delegate2<CStreamClientPull, void, CAddress, CStreamFrame>(this, &CStreamClientPull::RecvFrameMcast);
    this->delegateConnectionRecv = new Delegate2<CStreamClientPull, void, CConnectionReceiver*, CPacket*>(this, &CStreamClientPull::ConnectionRecv);
    this->delegateConnectionAccept = new Delegate2<CStreamClientPull, bool, CAddress, CPacket*>(this, &CStreamClientPull::ConnectionAccept);
    this->delegateConnectionAccepted = new Delegate1<CStreamClientPull, void, CConnectionSender*>(this, &CStreamClientPull::ConnectionAccepted);
    this->delegateConnectionReceiverOpen = new Delegate2<CStreamClientPull, void, CConnection*, CConnection::EOpenResult>(this, &CStreamClientPull::ConnectionReceiverOpen);
    this->delegateConnectionReceiverClose = new Delegate2<CStreamClientPull, void, CConnection*, CConnection::ECloseResult>(this, &CStreamClientPull::ConnectionReceiverClose);
    this->delegateConnectionSenderOpen = new Delegate2<CStreamClientPull, void, CConnection*, CConnection::EOpenResult>(this, &CStreamClientPull::ConnectionSenderOpen);
    this->delegateConnectionSenderClose = new Delegate2<CStreamClientPull, void, CConnection*, CConnection::ECloseResult>(this, &CStreamClientPull::ConnectionSenderClose);

    // Decoder
    this->decoderMcast = new CStreamDecoderFrame(this->delegateRecvFrameMcast);

    // Connections
    this->connectionLayer = new CConnectionLayer(
        this->sim,
        this->info->PortConnection(),
        this->info->ClientConnectionsMax(),
        delegateSend,
        this->delegateConnectionRecv,
        this->delegateConnectionAccept,
        this->delegateConnectionAccepted,
        this->info->ConnectionSegmentSize());

    // The deadline delta : the approximate buffering time
    this->scheduleDeadlineDelta = this->info->StreamFrameInterval() * this->info->StreamSegmentSize() * this->info->StreamBufferUcastSizeBuffering();
}

CStreamClientPull::~CStreamClientPull()
{
    // Delete delegates
    delete this->delegateRecvFrameMcast;
    
    delete this->delegateConnectionRecv;
    delete this->delegateConnectionAccept;
    delete this->delegateConnectionAccepted;

    delete this->delegateConnectionReceiverOpen;
    delete this->delegateConnectionReceiverClose;
    delete this->delegateConnectionSenderOpen;
    delete this->delegateConnectionSenderClose;

    // Delete coders
    delete this->decoderMcast;

    // Delete the buffers
    delete this->bufferMcast;
    delete this->bufferUcast;

    // Delete the timers
    delete this->timerPlayMcast;
    delete this->timerPlayUcast;
    delete this->timerSchedule;
    delete this->timerRegister;

    // Delete connections
    delete this->connectionLayer;

    // Delete senders and receivers
    while(!this->senders.empty()) delete this->senders.begin()->second;
    while(!this->sendersConnecting.empty()) delete this->sendersConnecting.begin()->second;
    while(!this->receivers.empty()) delete this->receivers.begin()->second;
}

void CStreamClientPull::Start(CChannel* channel)
{
    assert(channel);

    // Check the state
    assert(STOPPED == this->stateClient);

    // Check the timers are stopped
    assert(!this->timerPlayMcast->IsSet());
    assert(!this->timerPlayUcast->IsSet());
    assert(!this->timerSchedule->IsSet());
    assert(!this->timerRegister->IsSet());

    // Set the channel
    this->channel = channel;

    // Log
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u START %3u (%u)", this->sim->Time(), this->address.Address(), this->channel->Id(), this->channel->Type());
#endif

    // Reset statistics
    this->statRecvFrames = 0;
    this->statDiscardedFrames = 0;
    this->statPlayFrames = 0;
    this->statSuccessFrames[0] = 0;
    this->statSuccessFrames[1] = 0;
    this->statSuccessFrames[2] = 0;
    this->statFailFrames[0] = 0;
    this->statFailFrames[1] = 0;
    this->statFailFrames[2] = 0;

    this->statTimeClientStart = this->sim->Time();
    this->statTimeRecvStart = -1;
    this->statTimePlayStart = -1;
    this->statTimeWait = 0;

    this->statPlayFirstFrame = 0;
    this->statPlayLastFrame = 0;

    this->statSyncDelaySum = 0;
    this->statSyncDelayCount = 0;

    // Check the channel type
    switch(this->channel->Type())
    {
    case CChannel::CHANNEL_MULTICAST: this->StartMcast(); break;
    case CChannel::CHANNEL_UNICAST: this->StartUcast(); break;
    }
}

void CStreamClientPull::StartMcast()
{
    // Start session with phase 1
    this->processFrame = &CStreamClientPull::ProcessFrameMcastFirst;

    // Reset the decoder
    this->decoderMcast->Reset(this->channel->Id());

    // Join the multicast group
    (*this->delegateIgmpJoin)(this->channel->Address(), this->entry);

    // Change the state
    this->stateClient = MCAST_STREAM_FIRST;
}

void CStreamClientPull::StartUcast()
{
    // Check the registration state (client must not be registered)
    assert_client(NOT_REGISTERED == this->stateRegistration);
    assert_client(SCHEDULE_IDLE == this->stateSchedule);

    // Check the neighbor list is empty
    assert_client(this->neighbors.empty());

    // Check the connecting senders list is empty
    assert_client(this->sendersConnecting.empty());

    // Check the senders list is empty
    assert_client(this->senders.empty());

    // Check the receivers list is empty
    assert_client(this->receivers.empty());

    // Check the requests
    assert_client(this->requestBitmaps.empty());
    assert_client(this->requestSegments.empty());

    // Change the client state
    this->stateClient = UCAST_STREAM_FIRST;

    // Set processing function
    this->processFrame = &CStreamClientPull::ProcessFrameUcastFirst;

    // Reset the buffer to the initial segment
    __uint32 initialSegment = this->info->StreamSource(this->channel->Id())->Synchronize(this->sim->Time()) / this->info->StreamSegmentSize();
    this->bufferUcast->Reset(this->channel->Id(), initialSegment);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u RESET BUFFER to ch=%u seg=%u", this->sim->Time(), this->address.Address(), this->channel->Id(), initialSegment);
#endif

    // Start the schedule
    this->TimerSchedule(NULL);
}

void CStreamClientPull::Stop()
{
    // Check the state
    assert_client(STOPPED != this->stateClient);

    // Log
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u STOP %3u", this->sim->Time(), this->address.Address(), this->channel->Id());
#endif

    // Stop client
    switch(this->channel->Type())
    {
    case CChannel::CHANNEL_MULTICAST: this->StopMcast(); break;
    case CChannel::CHANNEL_UNICAST: this->StopUcast(); break;
    }

    // If there are active timer, cancel
    if(this->timerPlayMcast->IsSet()) this->timerPlayMcast->Cancel();
    if(this->timerPlayUcast->IsSet()) this->timerPlayUcast->Cancel();
    if(this->timerSchedule->IsSet()) this->timerSchedule->Cancel();
    if(this->timerRegister->IsSet()) this->timerRegister->Cancel();

    // Finalize statistics
    this->statTimeFinish = this->sim->Time();

    // Set channel to null
    this->channel = NULL;

    // Change the state
    this->stateClient = STOPPED;
}

void CStreamClientPull::StopMcast()
{
    // Leave the multicast group
    (*this->delegateIgmpLeave)(this->channel->Address());

    // If the session is waiting, add the waiting time
    if(MCAST_STREAM_WAIT == this->stateClient) this->statTimeWait += (this->sim->Time() - this->sessionTimeLastStop);
}

void CStreamClientPull::StopUcast()
{
    // Close all receivers
    while(!this->receivers.empty())
    {
        // Get the first receiver
        CStreamPullReceiver::TReceiverList::iterator receiver = this->receivers.begin();

        // Close the connection
        receiver->second->Connection()->Close();

        // Delete the receiver (this will remove the connection tag and the receiver from the list)
        if(this->receivers.begin() == receiver) delete this->receivers.begin()->second;
    }

    // Close all senders
    while(!this->senders.empty())
    {
        // Get the first sender
        CStreamPullSender::TSenderList::iterator sender = this->senders.begin();

        // Close the connection
        sender->second->Connection()->Close();

        // Delete the sender (this will remove the connection tag and the sender from the list)
        if(this->senders.begin() == sender) delete sender->second;
    }
    
    // Clear connecting senders
    while(!this->sendersConnecting.empty())
    {
        // Get the first sender
        CStreamPullSender::TSenderList::iterator sender = this->sendersConnecting.begin();

        // Close the connection
        sender->second->Connection()->Close();

        // Delete the connecting sender (this will remove the connection tag and the sender from the list)
        if(this->sendersConnecting.begin() == sender) delete sender->second;
    }

    // Clear neighbors
    this->neighbors.clear();

    // Clear requests
    this->requestBitmaps.clear();
    this->requestSegments.clear();

    // Deregister
    if(REGISTERED == this->stateRegistration) this->UcastDeregister();

    // Change the state
    this->stateSchedule = SCHEDULE_IDLE;
}

void CStreamClientPull::Recv(CAddress srcAddress, CAddress dstAddress, __uint16 srcPort, __uint16 dstPort, CPacket* packet)
{
    // Process received packets
    if(NULL == packet) return;
    if((dstAddress != this->address) && (!dstAddress.IsMulticast())) return;

    // Process packet depending on UDP destination
    if(this->info->PortStream() == dstPort) this->RecvStream(srcAddress, type_cast<CPacketStream*>(packet));
    if(this->info->PortControl() == dstPort) this->RecvMessage(srcAddress, type_cast<CStreamMessage*>(packet));
    if(this->info->PortConnection() == dstPort) this->RecvConnection(srcAddress, srcPort, dstPort, type_cast<CPacketConnection*>(packet));
}

void CStreamClientPull::RecvStream(CAddress src, CPacketStream* packet)
{
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        CConsole::SetColor(CConsole::DARK_GRAY);
        printf("\n\tT = %7.3lf  RECV STREAM  %u --- %u : ", this->sim->Time(), packet->Stream(), src.Address());
        CConsole::SetColor(CConsole::LIGHT_GRAY);
    }
#endif

    // Send received stream packets to the decoder
    CStreamDecoder::EResult result = this->decoderMcast->Decode(src, packet);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        CConsole::SetColor(CConsole::LIGHT_RED);
        switch(result)
        {
        case CStreamDecoder::FAIL_PACKET_TYPE: printf("FAIL PACKET TYPE"); break;
        case CStreamDecoder::FAIL_STREAM: printf("FAIL STREAM"); break;
        }
        CConsole::SetColor(CConsole::LIGHT_GRAY);
    }
#endif
}

void CStreamClientPull::RecvMessage(CAddress src, CStreamMessage* message)
{
    // Process messages

    switch(message->MessageType())
    {
    case CStreamMessage::STREAM_MESSAGE_BOOT_PULL_RESPONSE: this->RecvMessageBootResponse(src, type_cast<CStreamMessageBootPullResponse*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PULL_BITMAP_REQUEST: this->RecvMessageBitmapRequest(src, type_cast<CStreamMessagePullBitmapRequest*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PULL_BITMAP_RESPONSE: this->RecvMessageBitmapResponse(src, type_cast<CStreamMessagePullBitmapResponse*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PULL_SEGMENT_REQUEST: this->RecvMessageSegmentRequest(src, type_cast<CStreamMessagePullSegmentRequest*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PULL_SEGMENT_RESPONSE: this->RecvMessageSegmentResponse(src, type_cast<CStreamMessagePullSegmentResponse*>(message)); break;
    default: assert_client(0); /* do nothing */
    }
}

void CStreamClientPull::RecvConnection(CAddress srcAddress, __uint16 srcPort, __uint16 dstPort, CPacketConnection* packet)
{
    // Pass the packet to the connection layer
    this->connectionLayer->Recv(srcAddress, srcPort, dstPort, packet);
}

void CStreamClientPull::RecvFrameMcast(CAddress src, CStreamFrame frame)
{
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        CConsole::SetColor(CConsole::LIGHT_CYAN);
        printf("\n\tT = %7.3lf  RECV FRAME (M)  %u", this->sim->Time(), frame.Index());
        CConsole::SetColor(CConsole::LIGHT_GRAY);
    }
#endif

    // Check the state
    if(this->stateClient < MCAST_STREAM_FIRST)
    {
        this->statDiscardedFrames = 0;
        return;
    }

    // Check the decoder passed a correct frame
    if(frame.Stream() != this->channel->Id())
    {
        this->statDiscardedFrames = 0;
        return;
    }

    // Process frame
    (this->*this->processFrame)(frame);

    // Statistics
    this->statRecvFrames++;
}

void CStreamClientPull::ProcessFrameMcastFirst(CStreamFrame frame)
{
    // PLAYBACK Phase 1 : Process first frame to initialize playback buffer

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\t\t[MCAST FIRST]");
#endif

    // Check the state
    assert_client(MCAST_STREAM_FIRST == this->stateClient);

    // Reset the buffer
    this->bufferMcast->Reset(frame.Index());
    
    // Add the frame to the buffer
    this->bufferMcast->Add(frame);

    // Statistics
    this->statTimeRecvStart = this->sim->Time();

    // Switch to process buffering frames (phase 2)
    this->processFrame = &CStreamClientPull::ProcessFrameMcastBuffering;

    // Change the state
    this->stateClient = MCAST_STREAM_BUFFERING;
}

void CStreamClientPull::ProcessFrameMcastBuffering(CStreamFrame frame)
{
    // PLAYBACK Phase 2 : Buffering phase (playback is not started and frames are saved in the buffer)
    
    // Buffering phase finishes when the index of the received frame exceeds the the first frame from the
    // buffer, plus the buffering size

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\t\t[MCAST BUFFERING]");
#endif

    // Check the state
    assert_client(MCAST_STREAM_BUFFERING == this->stateClient);
    
    // Add the frame to the buffer
    this->bufferMcast->Add(frame);
    
    // Check if there are enough frames to start playback (the distance between the first frame in the buffer and the received frame)
    if(frame.Index() >= this->bufferMcast->CurrentIndex() + this->info->StreamBufferMcastSizeBuffering())
    {
        // Find the first independent frame
        for(__uint32 count = 0; count < this->bufferMcast->Count(); count++)
        {
            if(this->bufferMcast->IsCurrentIndependent())
            {
                // First frame found : start playback

                // Start playback : save session and stat information
                this->sessionFrameLastStart = this->bufferMcast->CurrentIndex();
                this->sessionTimeLastStart = this->sim->Time();

                this->statPlayFirstFrame = this->bufferMcast->CurrentIndex();
                this->statTimePlayStart = this->sim->Time();

                // Switch to process playback frames (phase 3)
                this->processFrame = &CStreamClientPull::ProcessFrameMcastPlay;

                // Change the state
                this->stateClient = MCAST_STREAM_PLAY;

                // Call the play timer
                this->TimerPlayMcast(NULL);

                break;
            }
            else this->bufferMcast->Shift();
        }
    }
}

void CStreamClientPull::ProcessFrameMcastPlay(CStreamFrame frame)
{
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\t\t[MCAST PLAY]");
#endif

    // PLAYBACK Phase 3 : Play phase
    assert_client(MCAST_STREAM_PLAY == this->stateClient);

    // Add the frame to the buffer
    this->bufferMcast->Add(frame);
}

void CStreamClientPull::ProcessFrameMcastWait(CStreamFrame frame)
{
    // PLAYBACK Wait : Buffering phase after a buffer underrun (playback is not started and frames are saved in the buffer)
    
    // Buffering phase finishes when the index of the received frame exceeds the the first frame from the
    // buffer, plus the buffering size

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\t\t[MCAST WAIT]");
#endif

    // Check the state
    assert_client(MCAST_STREAM_WAIT == this->stateClient);
    
    // Add the frame to the buffer
    this->bufferMcast->Add(frame);
    
    // Check if there are enough frames to start playback (the distance between the first frame in the buffer and the received frame)
    if(frame.Index() >= this->bufferMcast->CurrentIndex() + this->info->StreamBufferMcastSizeBuffering())
    {
        // Find the first frame
        for(__uint32 count = 0; count < this->bufferMcast->Count(); count++)
        {
            if(this->bufferMcast->HasCurrentIndex())
            {
                // Check the last stop
                assert_client(this->sim->Time() >= this->sessionTimeLastStop);

                // Save statistics
                this->statTimeWait += this->sim->Time() - this->sessionTimeLastStop;

                // Restart playback
                this->sessionFrameLastStart = this->bufferMcast->CurrentIndex();
                this->sessionTimeLastStart = this->sim->Time();

                // Switch to process playback frames (phase 3)
                this->processFrame = &CStreamClientPull::ProcessFrameMcastPlay;

                // Change the state
                this->stateClient = MCAST_STREAM_PLAY;

                // Call the play timer
                this->TimerPlayMcast(NULL);

                break;
            }
            else
            {
                // Increment the number of played frames
                this->statPlayFrames++;

                // Playback sync statistics
                this->statSyncDelaySum += this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferMcast->CurrentIndex());
                this->statSyncDelayCount++;

                // Mark the frame as failed
                assert_client(this->info->StreamSource(this->channel->Id())->FrameType(this->bufferMcast->CurrentIndex()) < 3);
                this->statFailFrames[this->info->StreamSource(this->channel->Id())->FrameType(this->bufferMcast->CurrentIndex())]++;
                
                // Shift the buffer
                this->bufferMcast->Shift();
            }
        }
    }
}

void CStreamClientPull::ProcessFrameUcastFirst(CStreamFrame frame)
{
    // Process first frame

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tADD %u (%u/%u) [UCAST FIRST]", frame.Index(), frame.SegmentIndex(), frame.Segment());
#endif

    // Check the client state
    assert_client(UCAST_STREAM_FIRST == this->stateClient);
    
    // Add the frame to the buffer
    this->bufferUcast->Add(frame);

    // Statistics
    this->statTimeRecvStart = this->sim->Time();

    // Switch to process buffering frames (phase 2)
    this->processFrame = &CStreamClientPull::ProcessFrameUcastBuffering;

    // Change the client state
    this->stateClient = UCAST_STREAM_BUFFERING;
}

void CStreamClientPull::ProcessFrameUcastBuffering(CStreamFrame frame)
{
    // PLAYBACK Phase 2 : Buffering phase (playback is not started and frames are saved in the buffer)
    
    // Buffering phase finishes when the index of the received frame exceeds the the first frame from the
    // buffer, plus the buffering size

    // Check the state
    assert_client(UCAST_STREAM_BUFFERING == this->stateClient);

    // Add the frame to the buffer
    this->bufferUcast->Add(frame);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tADD %u (%u/%u) [UCAST BUFFERING]\t(segment_index=%u  frame_index=%u buffering_size=%u)", frame.Index(), frame.SegmentIndex(), frame.Segment(), this->bufferUcast->CurrentIndexSegment(), this->bufferUcast->CurrentIndexFrame(), this->info->StreamBufferUcastSizeBuffering());
#endif

    // Check if the buffering is complete
    bool bufferingComplete = true;
    for(__uint32 index = 0; (index < this->info->StreamBufferUcastSizeBuffering()) && bufferingComplete; index++)
    {
        bufferingComplete = bufferingComplete && this->bufferUcast->IsSegmentComplete(this->bufferUcast->CurrentIndexSegment() + index);
    }

    if(bufferingComplete)
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_GREEN);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("YES");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

        // Buffering is complete : start playback

        this->sessionFrameLastStart = this->bufferUcast->CurrentIndexFrame();
        this->sessionTimeLastStart = this->sim->Time();

        this->statPlayFirstFrame = this->bufferUcast->CurrentIndexFrame();
        this->statTimePlayStart = this->sim->Time();

        // Switch to process playback frames (phase 3)
        this->processFrame = &CStreamClientPull::ProcessFrameUcastPlay;

        // Change the state
        this->stateClient = UCAST_STREAM_PLAY;

        // Call the play timer
        this->TimerPlayUcast(NULL);

        // Set the registration timer
        if(this->timerRegister->IsSet()) this->timerRegister->Cancel();
        this->timerRegister->SetAfter(this->info->BootRegisterDelay());
    }
    else
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_RED);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("NO");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
    }
}

void CStreamClientPull::ProcessFrameUcastPlay(CStreamFrame frame)
{
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        printf("\t\t[UCAST PLAY]");
        if(this->bufferUcast->IsSegmentComplete(this->bufferUcast->CurrentIndexSegment()))
        {
            CConsole::SetColor(CConsole::LIGHT_GREEN);
            printf("YES");
            CConsole::SetColor(CConsole::LIGHT_GRAY);
        }
        else
        {
            CConsole::SetColor(CConsole::LIGHT_RED);
            printf("NO");
            CConsole::SetColor(CConsole::LIGHT_GRAY);
        }
    }
#endif

    // PLAYBACK Phase 3 : Play phase
    assert_client(UCAST_STREAM_PLAY == this->stateClient);

    // Add the frame to the buffer and if the frame is new send it to the receivers
    this->bufferUcast->Add(frame);
}

void CStreamClientPull::TimerPlayMcast(CTimerInfo* info)
{
    assert_client(MCAST_STREAM_PLAY == this->stateClient);

    // Check if there is a buffer underrun
    if(this->bufferMcast->NumPlay() == 0)
    {
        // Change the state
        this->stateClient = MCAST_STREAM_WAIT;

        // Pause playback
        this->sessionTimeLastStop = this->sim->Time();

        // Switch to process wait frames (phase 3)
        this->processFrame = &CStreamClientPull::ProcessFrameMcastWait;

        return;
    }

    // Play the first frame from the buffer
    this->statPlayFrames++;

    // Playback sync statistics
    this->statSyncDelaySum += this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferMcast->CurrentIndex());
    this->statSyncDelayCount++;

    if(this->bufferMcast->IsCurrentDecodable())
    {
        this->statPlayLastFrame = this->bufferMcast->CurrentIndex();
        assert_client(this->bufferMcast->CurrentFrame()->Type() < 3);
        this->statSuccessFrames[this->bufferMcast->CurrentFrame()->Type()]++;
    }
    else
    {
        assert_client(this->info->StreamSource(this->channel->Id())->FrameType(this->bufferMcast->CurrentIndex()) < 3);
        this->statFailFrames[this->info->StreamSource(this->channel->Id())->FrameType(this->bufferMcast->CurrentIndex())]++;
    }

    // Shift the buffer
    this->bufferMcast->Shift();

    // Schedule next play event
    this->timerPlayMcast->SetAt(this->sessionTimeLastStart + ((__time)(this->bufferMcast->CurrentIndex() - this->sessionFrameLastStart)) / (__time)this->channel->Fps());
}

void CStreamClientPull::TimerPlayUcast(CTimerInfo* info)
{
    assert_client(UCAST_STREAM_PLAY == this->stateClient);

    // Play the first frame from the buffer
    this->statPlayFrames++;

    // Playback sync statistics
    this->statSyncDelaySum += this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferUcast->CurrentIndexFrame());
    this->statSyncDelayCount++;

    if(this->bufferUcast->IsCurrentFrameDecodable())
    {
        assert_client(this->bufferUcast->HasCurrentFrame());

        assert_client(this->bufferUcast->CurrentIndexFrame() == this->bufferUcast->CurrentFrame().Index());
        assert_client(this->bufferUcast->CurrentFrame().Type() < 3);

        this->statPlayLastFrame = this->bufferUcast->CurrentIndexFrame();
        this->statSuccessFrames[this->bufferUcast->CurrentFrame().Type()]++;

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        {
            CConsole::SetColor(CConsole::LIGHT_GREEN);
            printf("\n\t\tT = %7.3lf  PLAY OKAY  %6u\t%c", this->sim->Time(), this->bufferUcast->CurrentIndexFrame(), stringFrameType[this->bufferUcast->CurrentFrame().Type()]);
            CConsole::SetColor(CConsole::LIGHT_GRAY);
            printf("\t\t(time=%.3lf frame_time=%.3lf sync_delay=%.3lf)", this->sim->Time(), this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferUcast->CurrentIndexFrame()),
                this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferUcast->CurrentIndexFrame()));
        }
#endif
    }
    else
    {
        assert_client(this->info->StreamSource(this->channel->Id())->FrameType(this->bufferUcast->CurrentIndexFrame()) < 3);
        this->statFailFrames[this->info->StreamSource(this->channel->Id())->FrameType(this->bufferUcast->CurrentIndexFrame())]++;

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        {
            CConsole::SetColor(CConsole::LIGHT_RED);
            printf("\n\t\tT = %7.3lf  PLAY FAIL  %6u\t%c : %s", this->sim->Time(), this->bufferUcast->CurrentIndexFrame(), stringFrameType[this->info->StreamSource(this->channel->Id())->FrameType(this->bufferUcast->CurrentIndexFrame())],
                this->bufferUcast->HasCurrentFrame()?"DECODE_FAIL":"MISSING");
            CConsole::SetColor(CConsole::LIGHT_GRAY);
            printf("\t\t(time=%.3lf frame_time=%.3lf sync_delay=%.3lf)", this->sim->Time(), this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferUcast->CurrentIndexFrame()),
                this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->bufferUcast->CurrentIndexFrame()));
        }
#endif
    }

    // Shift the buffer
    this->bufferUcast->ShiftFrame();

    if(REGISTERED == this->stateRegistration)
    {
        // Limit peer utilization : if the uplink utilization exceeds the available minus a guard interval of 20 %
        CLink* link = (*this->delegateLink)();
        
        if((link->MeterUtil(0) > 0.8 * link->Bandwidth(0)) || (this->receivers.size() > 0.8*link->Bandwidth(0) / this->channel->Bw()))
        {
            // If the client is registered, deregister
            if(REGISTERED == this->stateRegistration) this->UcastDeregister();

            // If the registration timer is set, cancel the timer
            if(this->timerRegister->IsSet()) this->timerRegister->Cancel();
        }
    }

    // Schedule next play event
    this->timerPlayUcast->SetAt(this->sessionTimeLastStart + ((__time)(this->bufferUcast->CurrentIndexFrame() - this->sessionFrameLastStart)) / (__time)this->channel->Fps());
}

void CStreamClientPull::TimerRegister(CTimerInfo* info)
{
    this->UcastRegister();
}

void CStreamClientPull::TimerSchedule(CTimerInfo* info)
{
    // Check the client and schedule status
    assert_client(this->stateClient >= UCAST_STREAM_FIRST);
    assert_client(this->stateSchedule == SCHEDULE_IDLE);

    // Calculate the schedule segment
    this->scheduleSegment = this->bufferUcast->CurrentIndexSegment();

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u TIMER_SCHEDULE neighbors %u of %u : ", this->sim->Time(), this->address.Address(), this->neighbors.size(), this->info->NumPartners());
#endif

    // If the number of senders is less than the number of partners
    if(this->neighbors.size() < this->info->NumPartners())
    {
        // Go to bootstrap phase

        // Change the schedule state
        this->stateSchedule = SCHEDULE_BOOTSTRAP_REQUEST;

        this->ScheduleBootstrapRequest();
    }
    else
    {
        // Go to bitmap phase

        // Change the schedule state
        this->stateSchedule = SCHEDULE_BITMAP_REQUEST;

        this->ScheduleBitmapRequest();
    }

    // Limit peer utilization : if the uplink utilization exceeds the available minus a guard interval of 20 %
    // If playback has started
    if(UCAST_STREAM_PLAY == this->stateClient)
    {
        // If not registered and registration timer is not set
        if((NOT_REGISTERED == this->stateRegistration) && (!this->timerRegister->IsSet()))
        {       
            CLink* link = (*this->delegateLink)();
            
            if((link->MeterUtil(0) < 0.7 * link->Bandwidth(0)) && (this->receivers.size() < 0.7*link->Bandwidth(0)/this->channel->Bw()))
            {
                // Register
                this->UcastRegister();
            }
        }
    }

    // Reset the schedule timer after the schedule interval
    this->timerSchedule->SetAfter(this->info->ScheduleInterval());
}

void CStreamClientPull::ScheduleBootstrapRequest()
{
    // Check the state
    assert_client(SCHEDULE_BOOTSTRAP_REQUEST == this->stateSchedule);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u BOOTSTRAP_REQUEST channel=%u", this->sim->Time(), this->address.Address(), this->channel->Id());
#endif

    // Send the bootstrap request
    CStreamMessageBootPullRequest* message = new CStreamMessageBootPullRequest(
        this->channel->Id(),
        this->info->NumPartners());

    this->SendMessage(this->channel->Address(), message);

    // Change the state
    this->stateSchedule = SCHEDULE_BOOTSTRAP_RESPONSE;
}

void CStreamClientPull::ScheduleConnectRequest()
{
    // Check the state
    assert_client(SCHEDULE_CONNECT_REQUEST == this->stateSchedule);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u CONNECT_REQUEST to : ", this->sim->Time(), this->address.Address());
#endif

    // Check the connecting senders list is empty
    assert_client(this->sendersConnecting.empty());

    // Assign a number of neighbors as new senders until the number of senders reaches the number of partners or the neighbors set is exhausted
    while((this->senders.size() + this->sendersConnecting.size() < this->info->NumPartners()) && (this->neighbors.size() > 0))
    {
        // Select a random neighbor
        __uint32 rnd = CRand::Generate(this->neighbors.size()-1);       
        TNeighborList::iterator iter = this->neighbors.begin();

        for(__uint32 idx = 0; iter != this->neighbors.end(); iter++, idx++)
            if(idx == rnd)
                break;
        assert_client(iter != this->neighbors.end());

        // Select the new sender
        CAddress address = *iter;

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("%u ", address.Address());
#endif

        // Remove the neighbor from the neighbor list
        this->neighbors.erase(iter);

        // Create a new connection for the new sender
        CConnectionReceiver* receiver;
        CConnectionLayer::EResult result = this->connectionLayer->Create(address, this->info->PortConnection(), &receiver);
        assert_client(CConnectionLayer::SUCCESS == result);

        // Create the sender, add it to the connecting senders list and as tag to the connection
        CStreamPullSender* sender = new CStreamPullSender(address, receiver, &this->sendersConnecting);

        // Add event handler for the connection
        (*receiver->EventOpen()) += this->delegateConnectionReceiverOpen;
        (*receiver->EventClose()) += this->delegateConnectionReceiverClose;

        // Open the connection
        receiver->Open();
    }

    // If the number of connecting sender is non-zero
    if(this->sendersConnecting.size() > 0)
    {
        // Change state to connect response
        this->stateSchedule = SCHEDULE_CONNECT_RESPONSE;
    }
    else
    {
        // Change state to bitmap request
        this->stateSchedule = SCHEDULE_BITMAP_REQUEST;

        // Go to bitmap request
        this->ScheduleBitmapRequest();
    }
}

void CStreamClientPull::ScheduleBitmapRequest()
{
    // Check the state
    assert_client(SCHEDULE_BITMAP_REQUEST == this->stateSchedule);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u BITMAP_REQUEST : [%u -- %u]", this->sim->Time(), this->address.Address(),
        this->scheduleSegment, this->scheduleSegment + this->info->ScheduleSize() - 1);
#endif

    // Check the request list is empty
    assert_client(this->requestBitmaps.empty());

    // Request bitmaps from current senders
    for(CStreamPullSender::TSenderList::iterator iter = this->senders.begin(); iter != this->senders.end(); iter++)
    {
        // Add the sender to the request list
        this->requestBitmaps.insert(pair<CAddress, CStreamPullSender*>(iter->first, iter->second));

        // Send a bitmap request message for the current schedule window
        CStreamMessagePullBitmapRequest* message = new CStreamMessagePullBitmapRequest(
            this->channel->Id(),
            CStreamMessagePull::CONNECTION_YES,
            iter->second->Connection()->Id(),       // Set local connection ID
            iter->second->Connection()->IdEntry(),  // Set local connection ID entry
            this->scheduleSegment,
            this->info->ScheduleSize());

        this->SendMessage(iter->first, message);

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("\n\t\t\tto : %u [%u -- %u]", ((CAddress)iter->first).Address(), this->scheduleSegment, this->scheduleSegment + this->info->ScheduleSize() - 1);
#endif
    }

    // If the number of bitmap request is non-zero
    if(!this->requestBitmaps.empty())
    {
        // Change the state and wait for bitmap responses
        this->stateSchedule = SCHEDULE_BITMAP_RESPONSE;
    }
    else
    {
        // Cancel the schedule and go to idle
        this->stateSchedule = SCHEDULE_IDLE;
    }
}

void CStreamClientPull::SchedulePostBitmapResponse()
{
    // Post bitmap response : execute segment scheduling using DoNET / Coolstreaming algorithm
    
    // Because in our scenario the server is provisioned with much more bandiwidth than the client,
    // which - using DoNET - might lead to be selected whenever is advertised -  we modifiy the algorithm
    // such that the server has a lower preference compared to peers

    // Using preference is not limited to the server : peers may also advertise a lower preference in the
    // bitmap response

    // Check the state
    assert_client(SCHEDULE_POST_BITMAP_RESPONSE == this->stateSchedule);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u POST_BITMAP_REQUEST ", this->sim->Time(), this->address.Address());
#endif

    // Calculate the number of copies with high and low preference
    if(this->senders.size() > PULL_MAX_SENDERS) printf("%u %u %u", this->address.Address(), this->channel->Id(), this->senders.size());
    assert_client(this->senders.size() <= PULL_MAX_SENDERS);

    // Reset the schedule number of copies to zero for each segment in the schedule
    for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
    {
        this->scheduleNumCopiesHi[indexSegment] = 0;
        this->scheduleNumCopiesLo[indexSegment] = 0;
    }

    // The segment size (in bits)
    __bits segmentSize = (__bits)(this->info->StreamSegmentSize() * this->channel->Bw() / this->channel->Fps());

    // For every sender
    __uint8 indexSender = 0;
    for(CStreamPullSender::TSenderList::iterator iter = this->senders.begin(); iter != this->senders.end(); iter++, indexSender++)
    {
        // Get the last response
        CStreamPullSender::CBitmapResponse* response = &iter->second->LastBitmapResponse();

        // Check the last bitmap response is for the current schedule
        assert_client(this->scheduleSegment == response->First());
        assert_client(this->info->ScheduleSize() == response->Count());

        // Save the bandwidth from each sender
        this->scheduleBw[indexSender] = response->Bw();

        if(response->Preference() == CStreamMessagePullBitmapResponse::PREF_HIGH)
        {
            // High preference

            // For every segment in the schedule
            for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
            {
                if(!this->bufferUcast->IsSegmentReceiving(indexSegment + this->scheduleSegment))
                {
                    // Calculate the deadline as frame time (can be negative)
                    __time deadline = (indexSegment - this->scheduleSegment) * this->info->StreamSegmentSize() / this->channel->Fps() + this->scheduleDeadlineDelta;

                    if((response->Bitmap() >> indexSegment) & 1)
                    {
                        // If the segment is in the bitmap
                        this->scheduleBitmapHi[indexSender][indexSegment] = 1;
                        // Set the sender for random access
                        this->scheduleSenderHi[indexSender][indexSegment] = iter->second;
                        // Set the deadline
                        this->scheduleDeadlineHi[indexSender][indexSegment] = deadline;
                        // Set the index for incremental access
                        this->scheduleSenderIndexHi[this->scheduleNumCopiesHi[indexSegment]][indexSegment] = indexSender;
                        // Increment the number of copies
                        this->scheduleNumCopiesHi[indexSegment]++;
                    }
                    else
                    {
                        this->scheduleBitmapHi[indexSender][indexSegment] = 0;
                        this->scheduleSenderHi[indexSender][indexSegment] = NULL;
                    }
                }
            }
        }
        else
        {
            // Low preference

            // For every segment in the schedule
            for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
            {
                if(!this->bufferUcast->IsSegmentReceiving(indexSegment + this->scheduleSegment))
                {
                    // Calculate the deadline as frame time (can be negative)
                    __time deadline = (indexSegment - this->scheduleSegment) * this->info->StreamSegmentSize() / this->channel->Fps() + this->scheduleDeadlineDelta;

                    if((response->Bitmap() >> indexSegment) & 1)
                    {
                        // If the segment is in the bitmap
                        this->scheduleBitmapLo[indexSender][indexSegment] = 1;
                        // Set the sender for random access
                        this->scheduleSenderLo[indexSender][indexSegment] = iter->second;
                        // Set the deadline
                        this->scheduleDeadlineLo[indexSender][indexSegment] = deadline;
                        // Set the index for incremental access
                        this->scheduleSenderIndexLo[this->scheduleNumCopiesHi[indexSegment]][indexSegment] = indexSender;
                        // Increment the number of copies
                        this->scheduleNumCopiesLo[indexSegment]++;
                    }
                    else
                    {
                        this->scheduleBitmapLo[indexSender][indexSegment] = 0;
                        this->scheduleSenderLo[indexSender][indexSegment] = NULL;
                    }
                }
            }
        }
    }

    // Schedule segments

    // Start with scheduling the segments with a single sender
    for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
    {
        if(!this->bufferUcast->IsSegmentReceiving(indexSegment + this->scheduleSegment))
        {
            // Use a high preference sender, if available
            if(this->scheduleNumCopiesHi[indexSegment] == 1)
            {
                __uint8 indexSender = this->scheduleSenderIndexHi[0][indexSegment];

                // Select the only sender
                this->scheduleSenders[indexSegment] = this->scheduleSenderHi[indexSender][indexSegment];

                // Decrease the available time for subsequent segments
                for(__uint32 index = indexSegment + 1; index < this->info->ScheduleSize(); index++)
                {
                    if(this->scheduleBitmapHi[indexSender][index])
                    {
                        this->scheduleDeadlineHi[indexSender][index] -= segmentSize / this->scheduleBw[indexSender];
                    }
                }
            }
            // Else, use a low preference sender, if available
            else if(this->scheduleNumCopiesLo[indexSegment] == 1)
            {
                __uint8 indexSender = this->scheduleSenderIndexLo[0][indexSegment];

                // Select the only sender
                this->scheduleSenders[indexSegment] = this->scheduleSenderLo[indexSender][indexSegment];

                // Decrease the available time for subsequent segments
                for(__uint32 index = indexSegment + 1; index < this->info->ScheduleSize(); index++)
                {
                    if(this->scheduleBitmapLo[indexSender][index])
                    {
                        this->scheduleDeadlineLo[indexSender][index] -= segmentSize / this->scheduleBw[indexSender];
                    }
                }
            }
            // Else, no sender available : set the sender to NULL
            else this->scheduleSenders[indexSegment] = NULL;
        }
    }


    for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
    {
        if(!this->bufferUcast->IsSegmentReceiving(indexSegment + this->scheduleSegment))
        {
            // Use a high preference sender, if available
            if(this->scheduleNumCopiesHi[indexSegment] > 1)
            {
                // Choose the sender with the maximum bandwidth that meets the deadline
                __bitrate senderBwMax = -1;
                __uint8 indexSenderMax = 0;
                for(__uint8 index = 0; index < this->scheduleNumCopiesHi[indexSegment]; index++)
                {
                    __uint8 indexSender = this->scheduleSenderIndexHi[index][indexSegment];

                    // If the sender meets the deadline condition
                    if(this->scheduleDeadlineHi[indexSender][indexSegment])
                    {
                        if(this->scheduleBw[indexSender] > senderBwMax)
                        {
                            senderBwMax = this->scheduleBw[indexSender];
                            indexSenderMax = indexSender;
                        }
                    }
                }

                if(senderBwMax > 0)
                {
                    // Select the sender with the maximum bandwidth
                    this->scheduleSenders[indexSegment] = this->scheduleSenderHi[indexSenderMax][indexSegment];

                    // Decrease the available time for subsequent segments
                    for(__uint32 index = indexSegment + 1; index < this->info->ScheduleSize(); index++)
                    {
                        if(this->scheduleBitmapHi[indexSenderMax][index])
                        {
                            this->scheduleDeadlineHi[indexSenderMax][index] -= segmentSize / this->scheduleBw[indexSenderMax];
                        }
                    }
                }
                else
                {
                    // No sender has been found to meet the deadline
                    this->scheduleSenders[indexSegment] = NULL;
                }
            }
                
            // If a high preference sender was not found use a low preference sender, if available
            if(this->scheduleNumCopiesLo[indexSegment] > 1)
            {
                // Choose the sender with the maximum bandwidth that meets the deadline
                __bitrate senderBwMax = -1;
                __uint8 indexSenderMax = 0;
                for(__uint8 index = 0; index < this->scheduleNumCopiesLo[indexSegment]; index++)
                {
                    __uint8 indexSender = this->scheduleSenderIndexLo[index][indexSegment];

                    // If the sender meets the deadline condition
                    if(this->scheduleDeadlineLo[indexSender][indexSegment])
                    {
                        if(this->scheduleBw[indexSender] > senderBwMax)
                        {
                            senderBwMax = this->scheduleBw[indexSender];
                            indexSenderMax = indexSender;
                        }
                    }
                }

                if(senderBwMax > 0)
                {
                    // Select the sender with the maximum bandwidth
                    this->scheduleSenders[indexSegment] = this->scheduleSenderLo[indexSenderMax][indexSegment];

                    // Decrease the available time for subsequent segments
                    for(__uint32 index = indexSegment + 1; index < this->info->ScheduleSize(); index++)
                    {
                        if(this->scheduleBitmapLo[indexSenderMax][index])
                        {
                            this->scheduleDeadlineLo[indexSenderMax][index] -= segmentSize / this->scheduleBw[indexSenderMax];
                        }
                    }
                }
                else
                {
                    // No sender has been found to meet the deadline
                    this->scheduleSenders[indexSegment] = NULL;
                }
            }
        }
    }

    // Scheduling complete

    // Change the state
    this->stateSchedule = SCHEDULE_SEGMENT_REQUEST;

    // Go to segment request
    this->ScheduleSegmentRequest();
}

void CStreamClientPull::ScheduleSegmentRequest()
{
    // Check the state
    assert_client(SCHEDULE_SEGMENT_REQUEST == this->stateSchedule);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        printf("\n\tT = %7.3lf PEER %3u SEGMENT_REQUEST : [%u -- %u] -> [ ", this->sim->Time(), this->address.Address(),
            this->scheduleSegment, this->scheduleSegment + this->info->ScheduleSize() - 1);
        for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
        {
            if(this->bufferUcast->IsSegmentReceiving(this->scheduleSegment + indexSegment))
                CConsole::SetColor(CConsole::LIGHT_RED);
            else
                CConsole::SetColor(CConsole::LIGHT_GREEN);
            printf("%u ", this->scheduleSenders[indexSegment]->Connection()->RemoteAddress().Address());
            CConsole::SetColor(CConsole::LIGHT_GRAY);
        }
        printf("]");
    }
#endif

    // Send requests according to the scheduling algorithm

    // Check the request list is empty
    assert_client(this->requestSegments.empty());

    // For all senders
    __uint8 indexSender = 0;
    for(CStreamPullSender::TSenderList::iterator iter = this->senders.begin(); iter != this->senders.end(); iter++, indexSender++)
    {
        // Generate request bitmap for this sender
        __uint64 bitmap = 0;

        for(__uint8 indexSegment = 0; indexSegment < this->info->ScheduleSize(); indexSegment++)
        {
            if(!this->bufferUcast->IsSegmentReceiving(indexSegment + this->scheduleSegment))
            {
                if(this->scheduleSenders[indexSegment] == iter->second)
                {
                    bitmap |= (((__uint64)1) << indexSegment);
                }
            }
        }

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("\n\t\t\tSender %3u : %016llX", iter->first, bitmap);
#endif

        // If there is at least one segment to request from this sender
        if(bitmap)
        {
            // Send the request
            CStreamMessagePullSegmentRequest* message = new CStreamMessagePullSegmentRequest(
                this->channel->Id(),
                iter->second->Connection()->Id(),
                iter->second->Connection()->IdEntry(),
                iter->second->Connection()->RemoteId(),
                iter->second->Connection()->RemoteIdEntry(),
                this->scheduleSegment,
                this->info->ScheduleSize(),
                bitmap);

            this->SendMessage(iter->first, message);

            // Add the sender to the request list
            this->requestSegments.insert(pair<CAddress, CStreamPullSender*>(iter->first, iter->second));
        }
    }

    // If the number of requests is non-zero
    if(!this->requestSegments.empty())
    {
        // Change the state and wait for responses
        this->stateSchedule = SCHEDULE_SEGMENT_RESPONSE;
    }
    else
    {
        // Otherwise cancel the schedule and go to idle
        this->stateSchedule = SCHEDULE_IDLE;
    }
}

void CStreamClientPull::RecvMessageBootResponse(CAddress src, CStreamMessageBootPullResponse* message)
{
    // Received bootstrap response

#ifdef LOG
    CConsole::SetColor(CConsole::LIGHT_CYAN);
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u BOOTSTRAP_RESPONSE %u neighbors : ", this->sim->Time(), this->address.Address(), src.Address(),
        message->Count());
    CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

    // If a delayed message, ignore
    if(message->Stream() != this->channel->Id()) return;

    // Check the schedule state
    assert_client(SCHEDULE_BOOTSTRAP_RESPONSE == this->stateSchedule);

    // Assert

    // Save the number of neighbors in the neighbor list
    for(__uint32 index = 0; index < message->Count(); index++)
    {
        // Check the neighbor is not already a sender
        if(this->senders.find(message->Host(index)) == this->senders.end())
        {
            this->neighbors.insert(message->Host(index));
#ifdef LOG
            if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
                printf("%u ", message->Host(index));
#endif
        }
    }

    // If the number of sender is less than the number of partners and the number of neighbors is greater than zero
    if((this->senders.size() < this->info->NumPartners()) && (this->neighbors.size() > 0))
    {
        // Change the state to connect request
        this->stateSchedule = SCHEDULE_CONNECT_REQUEST;

        // Go to connect request
        this->ScheduleConnectRequest();
    }
    else
    {
        // Change the state to bitmap request
        this->stateSchedule = SCHEDULE_BITMAP_REQUEST;

        // Go to bitmap request
        this->ScheduleBitmapRequest();
    }
}

void CStreamClientPull::RecvMessageBitmapRequest(CAddress src, CStreamMessagePullBitmapRequest* message)
{
    // Received bitmap request

    // Verify the requested stream
    if(message->Stream() == this->channel->Id())
    {
        // Generate the bitmap
        __uint64 bitmap = 0LL;

        // For all requested segments
        for(__uint8 index = 0; index < message->Count(); index++)
        {
            // If the requested segments exists in the buffer, set its bit to 1
            if(this->bufferUcast->IsSegmentComplete(message->First() + index))
            {
                bitmap |= ((__uint64)1) << index;
            }
        }

        // Send the reply
        CStreamMessagePullBitmapResponse* reply = new CStreamMessagePullBitmapResponse(
            message->Stream(),
            message->First(),
            message->Count(),
            bitmap,
            (*this->delegateLink)()->Bandwidth(0),
            (*this->delegateLink)()->MeterUtil(0),
            message->Connection(),
            message->ConnectionId(),
            message->ConnectionIdEntry(),
            CStreamMessagePullBitmapResponse::SUCCESS,
            CStreamMessagePullBitmapResponse::PREF_HIGH
            );

        this->SendMessage(src, reply);
    }
    else
    {
        // Send a fail reply with bitmap set to zero
        CStreamMessagePullBitmapResponse* reply = new CStreamMessagePullBitmapResponse(
            message->Stream(),
            message->First(),
            message->Count(),
            0LL,
            0,  // Set bandwidth to zero
            0,  // Set bandwidth utilization to zero
            message->Connection(),
            message->ConnectionId(),
            message->ConnectionIdEntry(),
            CStreamMessagePullBitmapResponse::FAIL,
            CStreamMessagePullBitmapResponse::PREF_HIGH
            );

        this->SendMessage(src, reply);
    }
}

void CStreamClientPull::RecvMessageBitmapResponse(CAddress src, CStreamMessagePullBitmapResponse* message)
{
    // Received bitmap response

    // If a delayed message, ignore
    if(message->Stream() != this->channel->Id()) return;

#ifdef LOG
    CConsole::SetColor(CConsole::LIGHT_CYAN);
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u BITMAP_RESPONSE from %u [%u -- %u]=%016llX  bw=%.0lf bw_util=%.0lf pref=%u : ", this->sim->Time(), this->address.Address(), src.Address(),
        message->First(), message->First() + message->Count() - 1, message->Bitmap(), message->Bw(), message->BwUtil(), message->Preference());
    CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

    // Check the schedule state
    assert_client(SCHEDULE_BITMAP_RESPONSE == this->stateSchedule);

    // Check the source is in the request list
    TRequestList::iterator iter = this->requestBitmaps.find(src);
    assert_client(iter != this->requestBitmaps.end());

    // Get the sender
    CStreamPullSender* sender = iter->second;
    
    // Remove the entry from the request list
    this->requestBitmaps.erase(iter);

    // Check the message against the sender
    assert_client(message->Connection() == CStreamMessagePull::CONNECTION_YES);
    assert_client(message->ConnectionId() == sender->Connection()->Id());
    assert_client(message->ConnectionIdEntry() == sender->Connection()->IdEntry());

    // Save the response information in the sender record
    sender->SetBitmapResponse(CStreamPullSender::CBitmapResponse(
        message->First(),
        message->Count(),
        message->Bitmap(),
        message->Bw(),
        message->Preference()));

    // If the response has been successful


    if(message->Response() != CStreamMessagePullBitmapResponse::SUCCESS)
    {
        CConnectionReceiver* connection = sender->Connection();

        // Delete the sender
        delete sender;

        // Close the connection
        connection->Close();


#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_RED);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("FAIL");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
    }
    else
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_GREEN);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("SUCCESS");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
    }

    // If bitmap request list is empty (all responses have been received)
    if(this->requestBitmaps.empty())
    {
        // Change the state to post bitmap response
        this->stateSchedule = SCHEDULE_POST_BITMAP_RESPONSE;

        // Go to post bitmap response
        this->SchedulePostBitmapResponse();
    }
}

void CStreamClientPull::RecvMessageSegmentRequest(CAddress src, CStreamMessagePullSegmentRequest* message)
{
    // Receiver segment request

    // Get the local connection
    CConnectionSender* sender = type_cast<CConnectionSender*>(this->connectionLayer->Get(message->DstId(), message->DstIdEntry()));
    assert_client(sender);
    assert_client(sender->State() >= CConnection::OPENED);

    __uint64 success = 0;
    __uint64 fail = 0xFFFFFFFFFFFFFFFFLL;
    __uint8 count = 0;

    // If the connection is in OPEN state
    if(sender->State() == CConnection::OPENED)
    {
        success = 0;
        fail = 0;

        // Get the receiver
        assert_client(sender->Tag());
        CStreamPullReceiver* receiver = type_cast<CStreamPullReceiver*>(sender->Tag());
        
        // Verify the receiver
        assert_client(receiver);
        assert_client(receiver->Connection() == sender);

        // Send the requested segments through the receiver
        for(__uint32 sIndex = 0; sIndex < message->Count(); sIndex++)
        {
            // If the segment is requested
            if((message->Bitmap() >> sIndex) & 1)
            {
                __uint32 segmentIndex = message->First() + sIndex;
                // If the segment exists in the buffer and is complete
                if(this->bufferUcast->IsSegmentComplete(segmentIndex))
                {
                    // Get the segment from the buffer
                    CStreamSegment* segment = this->bufferUcast->GetSegment(segmentIndex);

                    __uint32 frameIndexStart = segmentIndex * this->info->StreamSegmentSize();

                    // Send all frames from that segment
                    for(__uint32 fIndex = 0; fIndex < this->info->StreamSegmentSize(); fIndex++)
                    {
                        assert_client((*segment)[fIndex].Valid());
                        receiver->Send((*segment)[fIndex]);
                    }

                    // Mark the segment as success
                    success |= ((__uint64)1) << sIndex;
                }
                else
                {
                    // Mark the segment as fail
                    fail |= ((__uint64)1) << sIndex;
                }
            }
        }

        // Set the count to the requested count
        count = message->Count();
    }

    // Reply to the request
    CStreamMessagePullSegmentResponse* reply = new CStreamMessagePullSegmentResponse(
        message->Stream(),
        message->DstId(),
        message->DstIdEntry(),
        message->SrcId(),
        message->SrcIdEntry(),
        message->First(),
        message->Count(),
        0LL,
        0xFFFFFFFFFFFFFFFFLL
        );

    this->SendMessage(src, reply);
}

void CStreamClientPull::RecvMessageSegmentResponse(CAddress src, CStreamMessagePullSegmentResponse* message)
{
    // Received segment response

    // If a delayed message, ignore
    if(message->Stream() != this->channel->Id()) return;

    // Check the schedule state
    assert_client(SCHEDULE_SEGMENT_RESPONSE == this->stateSchedule);

#ifdef LOG
    CConsole::SetColor(CConsole::LIGHT_CYAN);
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u SEGMENT_RESPONSE : [%u -- %u] : ", this->sim->Time(), this->address.Address(),
        message->First(), message->First() + message->Count() - 1);
    CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

    if(src == this->channel->Address()) return;

    // Check the source is in the request list
    TRequestList::iterator iter = this->requestSegments.find(src);
    assert_client(iter != this->requestSegments.end());

    // Get the sender
    CStreamPullSender* sender = iter->second;
    
    // Remove the entry from the request list
    this->requestSegments.erase(iter);

    // Check the message against the sender
    assert_client(message->DstId() == sender->Connection()->Id());
    assert_client(message->DstIdEntry() == sender->Connection()->IdEntry());

    // Mark in the buffer the segments that have been marked as successful
    for(__uint32 index = 0; index < message->Count(); index++)
    {
        __uint32 segmentIndex = message->First() + index;

        this->bufferUcast->SegmentReceiving(segmentIndex, true);

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        {
            if((message->Success() >> index) & 1) CConsole::SetColor(CConsole::LIGHT_GREEN);
            else if((message->Fail() >> index) & 1) CConsole::SetColor(CConsole::LIGHT_RED);
            else CConsole::SetColor(CConsole::DARK_GRAY);
            printf("%u ", segmentIndex);
            CConsole::SetColor(CConsole::LIGHT_GRAY);
        }
#endif
    }

    // If the requests list is empty (all responses have been received)
    if(this->requestSegments.empty())
    {
        // Change the state to IDLE
        this->stateSchedule = SCHEDULE_IDLE;
    }
}

void CStreamClientPull::ConnectionRecv(CConnectionReceiver* receiver, CPacket* packet)
{
    assert_client(packet->Type() == PACKET_TYPE_STREAM);

    CPacketStreamFrame* packetFrame = type_cast<CPacketStreamFrame*>(packet);
    
    // Ignore frames not for this channel
    if(packetFrame->Stream() != this->channel->Id()) return;

    // Call the frame processing function
    (this->*this->processFrame)(packetFrame->Frame());

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        CConsole::SetColor(CConsole::DARK_GRAY);
        printf("\n\t\t\tT = %7.3lf PEER %3u RECV %u bits  ::  ", this->sim->Time(), this->address.Address(), packet->Size());
        CConsole::SetColor(CConsole::DARK_GREEN);
        printf("%u [%c]", type_cast<CPacketStreamFrame*>(packet)->Frame().Index(),
            stringFrameType[this->info->StreamSource(this->channel->Id())->FrameType(type_cast<CPacketStreamFrame*>(packet)->Frame().Index())]);
        CConsole::SetColor(CConsole::LIGHT_GRAY);
    }
#endif
}

bool CStreamClientPull::ConnectionAccept(CAddress src, CPacket* packet)
{
    CLink* link = (*this->delegateLink)();

    // Only accept connections
    return true;
//  return ((link->MeterUtil(0) < 0.7 * link->Bandwidth(0)) && (this->receivers.size() < 0.7*link->Bandwidth(0)/this->channel->Bw()));
}

void CStreamClientPull::ConnectionAccepted(CConnectionSender* sender)
{
    // Sender connection accepted : hook connection events
    (*sender->EventOpen()) += this->delegateConnectionSenderOpen;
    (*sender->EventClose()) += this->delegateConnectionSenderClose;
}

void CStreamClientPull::ConnectionReceiverOpen(CConnection* receiver, CConnection::EOpenResult result)
{
    // Opening receiver connection

    // Check the state
    assert_client(SCHEDULE_CONNECT_RESPONSE == this->stateSchedule);

#ifdef LOG
    CConsole::SetColor(CConsole::LIGHT_YELLOW);
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u RECIEVER_OPEN : ", this->sim->Time(), this->address.Address());
    CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

    // If connection is successfull
    if(CConnection::OPEN_SUCCESS == result)
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_GREEN);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("SUCCESS");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

        // Get the sender record
        assert_client(receiver->Tag());
        CStreamPullSender* sender = type_cast<CStreamPullSender*>(receiver->Tag());

        // Check the receiver type is connecting and is in the connecting senders list
        assert_client(sender->List() == &this->sendersConnecting);
        assert_client(sender->Type() ==  CStreamPullSender::CONNECTING_SENDER);
        assert_client(this->sendersConnecting.find(receiver->RemoteAddress()) != this->sendersConnecting.end());

        // Change the sender list from connecting senders to senders
        sender->List(&this->senders);

        // Change the sender type
        sender->Type(CStreamPullSender::SENDER);
    }
    else
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_RED);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("FAIL (%u)", result);
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

        // Connection closed : if the connection has a sender tag, delete it (this will remove the tag and the sender from the connection list)
        if(receiver->Tag()) delete receiver->Tag();
    }

    // If the connecting senders list is empty
    if(this->sendersConnecting.empty())
    {
        // Connection establishment for all senders has finished

        // Change state to bitmap request
        this->stateSchedule = SCHEDULE_BITMAP_REQUEST;

        // Go to bitmap request
        this->ScheduleBitmapRequest();
    }
}

void CStreamClientPull::ConnectionReceiverClose(CConnection* receiver, CConnection::ECloseResult result)
{
    // Closing receiver connection

#ifdef LOG
    CConsole::SetColor(CConsole::LIGHT_YELLOW);
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u RECIEVER_CLOSE ", this->sim->Time(), this->address.Address());
    CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

    // Existing connection closed
    assert_client(receiver->Type() == CConnection::REQUESTER);

    // If closing is complete and the tag is set
    if(CConnection::CLOSE_COMPLETE == result)
    {
        // If the receiver has a sender tag, delete it (this will remove the tag an the sender from its associated list)
        if(receiver->Tag()) delete receiver->Tag();
    }
    // Else, do nothing
}

void CStreamClientPull::ConnectionSenderOpen(CConnection* sender, CConnection::EOpenResult result)
{
    // New connection opened

    // Check this is a sender connection
    assert_client(sender->Type() == CConnection::RESPONDER);

    // If the connection has been opened successfully
    if((CConnection::OPEN_SUCCESS == result) && (sender->RemoteAddress() != this->channel->Address()))
    {
        // Create a new receiver
        CStreamPullReceiver* receiver = new CStreamPullReceiver(
            this->sim,
            type_cast<CConnectionSender*>(sender),
            this->info->StreamFrameInterval(),
            sender->Id(),
            &this->receivers
            );
    }
    // Else, do nothing
}

void CStreamClientPull::ConnectionSenderClose(CConnection* sender, CConnection::ECloseResult result)
{
    // Existing connection closed
    assert_client(sender->Type() == CConnection::RESPONDER);

    // If closing is complete
    if(CConnection::CLOSE_COMPLETE == result)
    {
        // If the connection has a tag, delete the tag (this will remove the receiver tag from the connection and its associated list)
        if(sender->Tag()) delete sender->Tag();
    }
    // Else, do nothing
}

void CStreamClientPull::SendMessage(CAddress dst, CStreamMessage* message)
{
    (*this->delegateSend)(this->info->PortControl(), this->info->PortControl(), dst, 128, message);
}

void CStreamClientPull::UcastRegister()
{
    // Register for the current channel

    // Check the client state
    assert_client(this->stateClient >= UCAST_STREAM_FIRST);

    // Check the registration state
    assert_client(this->stateRegistration == NOT_REGISTERED);

    // Send a bootstrap registration message to the server
    CStreamMessageBootPullRegister* message = new CStreamMessageBootPullRegister(
        this->channel->Id(),
        this->address
        );

    this->SendMessage(this->channel->Address(), message);

    // Change the registration state
    this->stateRegistration = REGISTERED;
}

void CStreamClientPull::UcastDeregister()
{
    // Deregister for the current channel

    // Check the client state
    assert_client(this->stateClient >= UCAST_STREAM_FIRST);

    // Check the registration state
    assert_client(this->stateRegistration == REGISTERED);

    // Send a bootstrap registration message to the server
    CStreamMessageBootPullDeregister* message = new CStreamMessageBootPullDeregister(
        this->channel->Id(),
        this->address
        );

    this->SendMessage(this->channel->Address(), message);

    // Change the registration state
    this->stateRegistration = NOT_REGISTERED;
}


void CStreamClientPull::Finalize()
{
}