StreamClientPush.cpp

#include "Headers.h"
#include "StreamClientPush.h"
#include "Rand.h"
#include "Console.h"
#include "Debug.h"

CStreamClientPush::CStreamClientPush(
    CSimHandler*                                    sim,
    CInfoPush*                                      info,
    CAddress                                        address,
    IDelegate2<void, CAddress, CPacketStream*>*     delegateSendStream,
    IDelegate2<void, CAddress, CStreamMessage*>*    delegateSendMessage,
    IDelegate2<void, CAddress, __uint32>*           delegateIgmpJoin,
    IDelegate1<void, CAddress>*                     delegateIgmpLeave,
    IDelegate0<CLink*>*                             delegateLink,
    __uint32                                        entry,
    __bitrate                                       bw
    ) : CStreamClient(sim)
{
    assert(info);

    assert(delegateSendStream);
    assert(delegateSendMessage);
    assert(delegateIgmpJoin);
    assert(delegateIgmpLeave);
    assert(delegateLink);

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

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

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

    // Create the buffer
    this->buffer = new CStreamBuffer(this->info->StreamBufferSize(), this->info->StreamBufferSizeHistory());

    // Unicast session
    this->bw = bw;
    this->bwUsed = 0;

    // Multicast session
    this->entry = entry;

    // Create the timer
    this->timerPlayMcast = new CTimer<CStreamClientPush>(
        this->sim,
        this,
        &CStreamClientPush::TimerPlayMcast
        );
    this->timerPlayUcast = new CTimer<CStreamClientPush>(
        this->sim,
        this,
        &CStreamClientPush::TimerPlayUcast
        );
    this->timerBoot = new CTimer<CStreamClientPush>(
        this->sim,
        this,
        &CStreamClientPush::TimerBoot
        );
    this->timerRegister = new CTimer<CStreamClientPush>(
        this->sim,
        this,
        &CStreamClientPush::TimerRegister
        );
    this->timerBuffer = new CTimer<CStreamClientPush>(
        this->sim,
        this,
        &CStreamClientPush::TimerBuffer
        );

    // Set the interface
    this->entry = entry;

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

    // Delegates
    this->delegateRecvFrameMcast = new Delegate2<CStreamClientPush, void, CAddress, CStreamFrame>(this, &CStreamClientPush::RecvFrameMcast);
    this->delegateRecvFrameUcast = new Delegate2<CStreamClientPush, void, CAddress, CStreamFrame>(this, &CStreamClientPush::RecvFrameUcast);

    // Encoder
    this->encoder = new CStreamEncoderFrame(this->delegateSendStream);

    // Decoders
    this->decoder = NULL;
    this->decoderMcast = new CStreamDecoderFrame(this->delegateRecvFrameMcast);
    this->decoderUcast = new CStreamDecoderFrame(this->delegateRecvFrameUcast);
}

CStreamClientPush::~CStreamClientPush()
{
    // Delete delegates
    delete this->delegateRecvFrameMcast;
    delete this->delegateRecvFrameUcast;

    // Delete coders
    delete this->encoder;
    delete this->decoderMcast;
    delete this->decoderUcast;

    // Delete the buffer
    delete this->buffer;

    // Delete the timers
    delete this->timerPlayMcast;
    delete this->timerPlayUcast;
    delete this->timerBoot;
    delete this->timerRegister;
    delete this->timerBuffer;
}

void CStreamClientPush::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->timerBoot->IsSet());
    assert(!this->timerRegister->IsSet());
    assert(!this->timerBuffer->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 CStreamClientPush::StartMcast()
{
    // Start session with phase 1
    this->processFrame = &CStreamClientPush::ProcessFrameMcastFirst;

    // Set decoder
    this->decoder = this->decoderMcast;

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

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

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

void CStreamClientPush::StartUcast()
{
    // Check the registration state (client must not be registered)
    assert(NOT_REGISTERED == this->stateRegistration);

    // Set decoder
    this->decoder = this->decoderUcast;

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

    // Set the session variables
    this->neighbors.clear();                // Clear the neighbors
    this->sender = channel->Address();      // Set sender to the server

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

    // Go to streaming STEP 1
    this->StreamBootstrap();
}

void CStreamClientPush::Stop()
{
    // Check the state
    assert(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->timerBoot->IsSet()) this->timerBoot->Cancel();
    if(this->timerRegister->IsSet()) this->timerRegister->Cancel();
    if(this->timerBuffer->IsSet()) this->timerBuffer->Cancel();

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

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

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

void CStreamClientPush::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 CStreamClientPush::StopUcast()
{
    // Close all receivers
    this->StreamClose();

    // Leave the stream
    this->StreamLeave();

    // If registered, deregister from bootstrap server
    if(REGISTERED == this->stateRegistration) this->StreamDeregister();

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

void CStreamClientPush::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));
}

void CStreamClientPush::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\t\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->decoder->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 CStreamClientPush::RecvMessage(CAddress src, CStreamMessage* message)
{
    // Process messages
    switch(message->MessageType())
    {
    case CStreamMessage::STREAM_MESSAGE_PUSH_JOIN: this->RecvMessageJoin(src, type_cast<CStreamMessagePushJoin*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PUSH_LEAVE: this->RecvMessageLeave(src, type_cast<CStreamMessagePushLeave*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PUSH_CLOSE: this->RecvMessageClose(src, type_cast<CStreamMessagePushClose*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_RESPONSE: this->RecvMessageBootResponse(src, type_cast<CStreamMessageBootPushResponse*>(message)); break;
    default: assert(0); /* do nothing */
    }
}

void CStreamClientPush::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\t\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 CStreamClientPush::RecvFrameUcast(CAddress src, CStreamFrame frame)
{
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        CConsole::SetColor(CConsole::LIGHT_CYAN);
        printf("\n\t\tT = %7.3lf  RECV FRAME (U)  %u", this->sim->Time(), frame.Index());
        CConsole::SetColor(CConsole::LIGHT_GRAY);
    }
#endif

    // Check the client state (if not a client state that allows receiving frames, ignore)
    if(this->stateClient < UCAST_STREAM_FIRST)
    {
        this->statDiscardedFrames++;
        return;
    }

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

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

    // Statistics
    this->statRecvFrames++;
}

void CStreamClientPush::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(MCAST_STREAM_FIRST == this->stateClient);

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

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

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

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

void CStreamClientPush::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(MCAST_STREAM_BUFFERING == this->stateClient);
    
    // Add the frame to the buffer
    this->buffer->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->buffer->CurrentIndex() + this->info->StreamBufferSizeBuffering())
    {
        // Find the first independent frame
        for(__uint32 count = 0; count < this->buffer->Count(); count++)
        {
            if(this->buffer->IsCurrentIndependent())
            {
                // First frame found : start playback

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

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

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

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

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

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

void CStreamClientPush::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(MCAST_STREAM_PLAY == this->stateClient);

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

void CStreamClientPush::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(MCAST_STREAM_WAIT == this->stateClient);
    
    // Add the frame to the buffer
    this->buffer->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->buffer->CurrentIndex() + this->info->StreamBufferSizeBuffering())
    {
        // Find the first frame
        for(__uint32 count = 0; count < this->buffer->Count(); count++)
        {
            if(this->buffer->HasCurrentIndex())
            {
                // Check the last stop
                assert(this->sim->Time() >= this->sessionTimeLastStop);

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

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

                // Switch to process playback frames (phase 3)
                this->processFrame = &CStreamClientPush::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->buffer->CurrentIndex());
                this->statSyncDelayCount++;

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

void CStreamClientPush::ProcessFrameUcastFirst(CStreamFrame frame)
{
    // PROCESS 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[UCAST FIRST]");
#endif

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

    // Reset the buffer
    this->buffer->Reset(frame.Index());
    
    // Add the frame to the buffer and if the frame is new send it to the receivers
    if(this->buffer->Add(frame)) this->SendStream(frame);

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

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

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

void CStreamClientPush::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

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\t\t[UCAST BUFFERING]\t(frame_index=%u  current_index=%u  buffering_size=%u  buffer_num=%u)", frame.Index(), this->buffer->CurrentIndex(), this->info->StreamBufferSizeBuffering(), this->buffer->Count());
#endif

    // Check the state
    assert(UCAST_STREAM_BUFFERING == this->stateClient);
    
    // Add the frame to the buffer and if the frame is new send it to the receivers
    if(this->buffer->Add(frame)) this->SendStream(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->buffer->CurrentIndex() + this->info->StreamBufferSizeBuffering())
    {
        // Find the first independent frame
        for(__uint32 count = 0; count < this->buffer->Count(); count++)
        {
            if(this->buffer->IsCurrentIndependent())
            {
                // First frame found : start playback

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

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

                // Switch to process playback frames (phase 3)
                this->processFrame = &CStreamClientPush::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());

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

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

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

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

void CStreamClientPush::ProcessFrameUcastWait(CStreamFrame frame)
{
    // PLAYBACK Wait : Buffering phase after a buffer underrun (playback is not started and frames are saved in the buffer)

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

    // 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(UCAST_STREAM_WAIT == this->stateClient);
    
    // Add the frame to the buffer and if the frame is new send it to the receivers
    if(this->buffer->Add(frame)) this->SendStream(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->buffer->CurrentIndex() + this->info->StreamBufferSizeBuffering())
    {
        // Find the first frame
        for(__uint32 count = 0; count < this->buffer->Count(); count++)
        {
            if(this->buffer->HasCurrentIndex())
            {
                // Check the last stop
                assert(this->sim->Time() >= this->sessionTimeLastStop);

#ifdef LOG
                if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
                {
                    CConsole::SetColor(CConsole::LIGHT_YELLOW);
                    printf("\n\tT = %7.3lf WAIT : %7.3lf (%7.3lf - %7.3lf)", this->sim->Time(), this->sim->Time() - this->sessionTimeLastStop, this->sessionTimeLastStop, this->sim->Time());
                    CConsole::SetColor(CConsole::LIGHT_GRAY);
                }
#endif
                // Save statistics
                this->statTimeWait += this->sim->Time() - this->sessionTimeLastStop;

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

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

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

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

                // Set the registration timer (cancel and reset)
                if(this->timerRegister->IsSet()) this->timerRegister->Cancel();
                this->timerRegister->SetAfter(this->info->BootRegisterDelay());

                // If a buffer underrun timer is set, cancel
                if(this->timerBuffer->IsSet()) this->timerBuffer->Cancel();

                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->buffer->CurrentIndex());
                this->statSyncDelayCount++;

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

#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->buffer->CurrentIndex(), stringFrameType[this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex())],
                        this->buffer->HasCurrentIndex()?"DECODE_FAIL":"MISSING");
                    CConsole::SetColor(CConsole::LIGHT_GRAY);
                }
#endif
                // Shift the buffer
                this->buffer->Shift();
            }
        }
    }
}

void CStreamClientPush::ProcessFrameUcastReset(CStreamFrame frame)
{
    // PLAYBACK Reset : Reset the buffer after a long buffer underrun (playback is not started and frames are saved in the buffer)
    
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\t\t[UCAST RESET]");
#endif

    // After the buffer is reset by the first frame, switch to wait state and wait for the buffer to fill again
    // Check the state
    assert(UCAST_STREAM_RESET == this->stateClient);
    
    // Reset the buffer
    this->buffer->Reset(frame.Index());

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

    // Switch to the wait state
    this->processFrame = &CStreamClientPush::ProcessFrameUcastWait;
    
    // Change the state
    this->stateClient = UCAST_STREAM_WAIT;
}

void CStreamClientPush::StreamBootstrap()
{
    // Streaming STEP 1 : Bootstraping (connect to bootstrap server and retrieve list of neighbors)

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

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

    // Send a boot push query message to the bootstrap server (the channel address)
    CStreamMessageBootPushRequest* message = new CStreamMessageBootPushRequest(
        this->channel->Id(),
        this->info->BootQueryMax()
        );

    // Send the message
    (*this->delegateSendMessage)(this->channel->Address(), message);

    // Activate the timer to resend the query after a timeout period
    this->timerBoot->SetAfter(this->info->BootQueryTimeout());

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

void CStreamClientPush::StreamJoin()
{
    // Streaming STEP 2a : Joining (start streaming from the senders)

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

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

    if(NULL == this->channel) return;

    // Send a join request to the sender
    CStreamMessagePushJoin* message = new CStreamMessagePushJoin(this->channel->Id());

    (*this->delegateSendMessage)(this->sender, message);

    // Get ready to receive the stream : start streaming session with phase 1
    this->processFrame = &CStreamClientPush::ProcessFrameUcastFirst;

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

void CStreamClientPush::StreamLeave()
{
    // Streaming STEP 3a : Leave (inform the senders)

    // Send a leave to the sender
    CStreamMessagePushLeave* message = new CStreamMessagePushLeave(this->channel->Id());

    (*this->delegateSendMessage)(this->sender, message);
}

void CStreamClientPush::StreamRegister()
{
    // Streaming STEP 2b : Register (register to bootstrap server)

    // Check the client state (client must start playing in order to register)
    assert(this->stateClient >= UCAST_STREAM_FIRST);

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

    CStreamMessageBootPushRegister* message = new CStreamMessageBootPushRegister(this->channel->Id(), this->address, this->sender);

    (*this->delegateSendMessage)(this->channel->Address(), message);

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

void CStreamClientPush::StreamDeregister()
{
    // Check the resgistration state
    assert(REGISTERED == this->stateRegistration);

    // Streaming STEP 3b : Deregister (deregister from bootstrap server)
    CStreamMessageBootPushDeregister* message = new CStreamMessageBootPushDeregister(this->channel->Id(), this->address);

    (*this->delegateSendMessage)(this->channel->Address(), message);

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

void CStreamClientPush::StreamClose()
{
    // Streaming STEP ASYNC : Close (sent to receivers upon leaving or when receiving a request that cannot be served)

    for(set<CAddress>::iterator iter = this->receivers.begin(); iter != this->receivers.end(); iter++)
    {
        CStreamMessagePushClose* message = new CStreamMessagePushClose(this->channel->Id());

        (*this->delegateSendMessage)(*iter, message);

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

    // Clear the receivers list
    this->receivers.clear();

    // Set the used bandwidth to zero
    this->bwUsed = 0;
}

void CStreamClientPush::SendStream(CStreamFrame frame)
{
    // Check the client state
    assert(this->stateClient >= UCAST_STREAM_FIRST);

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        CConsole::SetColor(CConsole::DARK_GRAY);
        printf("\n\t\tT = %7.3lf  SEND FRAME (U)  %u : [ ", this->sim->Time(), frame.Index());
    }
#endif

    for(set<CAddress>::iterator iter = this->receivers.begin(); iter != this->receivers.end(); iter++)
    {
        // Send frame to the encoder
        this->encoder->Encode(*iter, frame);

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

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

void CStreamClientPush::RecvMessageJoin(CAddress src, CStreamMessagePushJoin* message)
{
    // Process received JOIN

    // Check whether the client can serve the channel
    if(NULL == this->channel) { this->RejectJoin(src, message); return; }
    if(this->channel->Id() != message->Stream()) { this->RejectJoin(src, message); return; }
    
    // Check whether the client is registered
    if(REGISTERED != this->stateRegistration) { this->RejectJoin(src, message); return; }

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

    // Check whether the peer would have bandwidth to serve the channel considering the current usage
    if(this->bw - this->bwUsed < this->info->StreamBwMargin() * this->channel->Bw())
    {

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

        // Reject the request
        this->RejectJoin(src, message);
        // Deregister
        this->StreamDeregister();
        return;
    }

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

    // Acceptance conditions met : add the host to the receivers list
    this->receivers.insert(src);

    // Estimate the increase in bandwidth usage
    this->bwUsed += this->channel->Bw();
}

void CStreamClientPush::RecvMessageLeave(CAddress src, CStreamMessagePushLeave* message)
{
    // Process received LEAVE

    // Check whether the peer is serving the channel (may be a delayed message)
    if(NULL == this->channel) return;
    if(this->channel->Id() != message->Stream()) return;

    // Check if the host is in the receivers list
    if(this->receivers.find(src) != this->receivers.end())
    {
        // Remove the host from the receivers list
        this->receivers.erase(src);

        // Estimate the decrease in bandwidth usage
        this->bwUsed -= this->channel->Bw();
    }
}

void CStreamClientPush::RecvMessageClose(CAddress src, CStreamMessagePushClose* message)
{
    // Process received CLOSE

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

    // Check the close is for current channel
    if(NULL == this->channel) return;
    if(this->channel->Id() != message->Stream()) return;

    // Check the close has been sent by the current sender (otherwise, ignore)
    if(src != this->sender) return;

    // Check the state of the client (if the close message is delayed)
    if(this->stateClient < UCAST_STREAM_FIRST) return;

    // If more neighbors are available
    if(this->neighbors.size() > 0)
    {
        // Select a new random neighbor as a receiver for this layer
        __uint32 index = CRand::Generate(this->neighbors.size());
        assert(index < this->neighbors.size());

        __uint32 idx = 0;
        for(set<CAddress>::iterator iter = this->neighbors.begin(); iter != this->neighbors.end(); iter++, idx++)
        {
            if(idx == index)
            {
                this->sender = *iter;                   // Set the sender
                break;
            }
        }

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

        // If the number of remaining neighbors is less than the threshold, refresh the list of neighbors
        if(this->neighbors.size() <= this->info->BootRefreshThreshold())
        {
#ifdef LOG
            if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
                printf("\n\tT = %7.3lf PEER %3u BOOTSTRAP RE-REQUEST %3u (%u neighbors remaining / %u threshold)", this->sim->Time(), this->address.Address(), this->channel->Id(), this->neighbors.size(), this->info->BootRefreshThreshold());
#endif
            // Send a boot push query message to the bootstrap server (the channel address)
            CStreamMessageBootPushRequest* request = new CStreamMessageBootPushRequest(
                this->channel->Id(),
                this->info->BootQueryMax()
                );

            (*this->delegateSendMessage)(this->channel->Address(), request);
        }
    }
    // Else, use the server
    else
    {
        this->sender = this->channel->Address();
    }

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

    // Send a JOIN to the new sender
    CStreamMessagePushJoin* reply = new CStreamMessagePushJoin(this->channel->Id());

    (*this->delegateSendMessage)(this->sender, reply);
}

void CStreamClientPush::RecvMessageBootResponse(CAddress src, CStreamMessageBootPushResponse* message)
{
    // Process received BOOT RESPONSE

    // If no current channel, do nothing (delayed response)
    if(NULL == this->channel) return;

    // If the message is not for this channel, do nothing (delayed response)
    if(message->Stream() != this->channel->Id()) return;

    // If the boot timer is active, cancel
    if(this->timerBoot->IsSet()) this->timerBoot->Cancel();

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
    {
        printf("\n\tT = %7.3lf PEER %3u BOOTSTRAP RESPONSE %3u : [%u neighbors] : ", this->sim->Time(), this->address.Address(), this->channel->Id(), message->Count());
        for(__uint32 host = 0; host < message->Count(); host++) printf("%u ", message->Host(host).Address());
    }
#endif

    // Process the response, depending on the client state
    if(UCAST_BOOTSTRAP_RESPONSE == this->stateClient)
    {
        // CASE 1 : Bootstrap request sent at the beginning of the session; this populates the neighbor
        // list and changes the state of the client


        // Process the response
        if(message->Count() > 0)
        {
            // If there are neighbors available (otherwise, the server is used by default)

            // Choose a random neighbor to the the sender for this layer
            __uint32 sender = CRand::Generate(message->Count());
            assert(sender < message->Count());

            // Save the neighbors
            for(__uint32 host = 0; host < sender; host++)
                this->neighbors.insert(message->Host(host));
            for(__uint32 host = sender+1; host < message->Count(); host++)
                this->neighbors.insert(message->Host(host));

            // Save the sender
            this->sender = message->Host(sender);
        }

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

        // Go to streaming STEP 2a
        this->StreamJoin();
    }
    else
    {
        // CASE 2 : Refresh bootstrap request sent during a session; this refreshes the neighbor
        // list but does not change the state of the client or the sender

        // Process the response
        // Clear the neighbors list
        this->neighbors.clear();

        // Save the neighbors
        for(__uint32 host = 0; host < message->Count(); host++)
            this->neighbors.insert(message->Host(host));
    }
}

void CStreamClientPush::RejectJoin(CAddress src, CStreamMessagePushJoin* message)
{
    // Reject JOIN request : send a CLOSE message to the sender
    CStreamMessagePushClose* reply = new CStreamMessagePushClose(message->Stream());

    (*this->delegateSendMessage)(src, reply);
}

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

    // Check if there is a buffer underrun
    if(this->buffer->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 = &CStreamClientPush::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->buffer->CurrentIndex());
    this->statSyncDelayCount++;

    if(this->buffer->IsCurrentDecodable())
    {
        this->statPlayLastFrame = this->buffer->CurrentIndex();
        assert(this->buffer->CurrentFrame()->Type() < 3);
        this->statSuccessFrames[this->buffer->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->buffer->CurrentIndex(), stringFrameType[this->buffer->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->buffer->CurrentIndex()),
                this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->buffer->CurrentIndex()));
        }
#endif
    }
    else
    {
        assert(this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex()) < 3);
        this->statFailFrames[this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex())]++;

#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->buffer->CurrentIndex(), stringFrameType[this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex())],
                this->buffer->HasCurrentIndex()?"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->buffer->CurrentIndex()),
                this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->buffer->CurrentIndex()));
        }
#endif
    }

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

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

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

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

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

        // If the host is registered, deregister
        if(REGISTERED == this->stateRegistration) this->StreamDeregister();

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

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

        // Set a buffer underrun timer: if the buffer underrun is not fixed until the timer expires, select a new sender
        if(this->timerBuffer->IsSet()) this->timerBuffer->Cancel();

        // Set the info
        this->timerBufferChannel = this->channel->Id();
        this->timerBuffer->SetAfter(this->info->StreamBufferUnderrunTimeout());

#ifdef LOG
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        {
            CConsole::SetColor(CConsole::LIGHT_YELLOW);
            printf("\n\t\tT = %7.3lf  PLAY WAIT", this->sim->Time());
            CConsole::SetColor(CConsole::LIGHT_GRAY);
        }
#endif
        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->buffer->CurrentIndex());
    this->statSyncDelayCount++;

    if(this->buffer->IsCurrentDecodable())
    {
        assert(this->buffer->HasCurrentIndex());
        assert(this->buffer->CurrentIndex() == this->buffer->CurrentFrame()->Index());
        assert(this->buffer->CurrentFrame()->Type() < 3);

        this->statPlayLastFrame = this->buffer->CurrentIndex();
        this->statSuccessFrames[this->buffer->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->buffer->CurrentIndex(), stringFrameType[this->buffer->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->buffer->CurrentIndex()),
                this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->buffer->CurrentIndex()));
        }
#endif
    }
    else
    {
        assert(this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex()) < 3);
        this->statFailFrames[this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex())]++;

#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->buffer->CurrentIndex(), stringFrameType[this->info->StreamSource(this->channel->Id())->FrameType(this->buffer->CurrentIndex())],
                this->buffer->HasCurrentIndex()?"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->buffer->CurrentIndex()),
                this->sim->Time() - this->info->StreamSource(this->channel->Id())->FrameTime(this->buffer->CurrentIndex()));
        }
#endif
    }

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

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

void CStreamClientPush::TimerBoot(CTimerInfo* info)
{
    if(NULL == this->channel) return;

    // Bootstrap timer : resend the boostrap query
    CStreamMessageBootPushRequest* message = new CStreamMessageBootPushRequest(
        this->channel->Id(),
        this->info->BootQueryMax()
        );

    (*this->delegateSendMessage)(this->channel->Address(), message);

    // Activate the timer to resend the query after a timeout period
    this->timerBoot->SetAfter(this->info->BootQueryTimeout());
}

void CStreamClientPush::TimerRegister(CTimerInfo* info)
{
    // Register
    this->StreamRegister();
}

void CStreamClientPush::TimerBuffer(CTimerInfo* info)
{
    // Buffer underrun timer : if a buffer underrun instance is not self-repairing (i.e. caused by delayed packets),
    // select a new source

    // First, check the buffer underrun instance is genuine : the client must be in a wait state
    if(UCAST_STREAM_WAIT != this->stateClient) return;

    // Check the timer is set for the current channel
    if(NULL == this->channel) return;
    assert(timerBufferChannel == this->channel->Id());
    if(this->channel->Id() != this->timerBufferChannel) return;

    // Then, send a leave message to the old sender
    this->StreamLeave();

    // Select a new sender from the neighbors list : if more neighbors are available
    if(this->neighbors.size() > 0)
    {
        // Select a new random neighbor as a receiver for this layer
        __uint32 index = CRand::Generate(this->neighbors.size());
        assert(index < this->neighbors.size());

        __uint32 idx = 0;
        for(set<CAddress>::iterator iter = this->neighbors.begin(); iter != this->neighbors.end(); iter++, idx++)
        {
            if(idx == index)
            {
                this->sender = *iter;
                break;
            }
        }

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

        // If the number of remaining neighbors is less than the threshold, refresh the list of neighbors
        if(this->neighbors.size() <= this->info->BootRefreshThreshold())
        {
            // Send a boot push query message to the bootstrap server (the channel address)
            CStreamMessageBootPushRequest* message = new CStreamMessageBootPushRequest(
                this->channel->Id(),
                this->info->BootQueryMax()
                );

            (*this->delegateSendMessage)(this->channel->Address(), message);
        }
    }
    // Else, use the server
    else this->sender = this->channel->Address();

    // Send a JOIN to the new sender
    CStreamMessagePushJoin* reply = new CStreamMessagePushJoin(this->channel->Id());

    (*this->delegateSendMessage)(this->sender, reply);

    // Switch the client to the reset state (after a long buffer underrun, the client must be resync with the the frames sent by the source)
    this->stateClient = UCAST_STREAM_RESET;
    this->processFrame = &CStreamClientPush::ProcessFrameUcastReset;
}

void CStreamClientPush::Finalize()
{
}