StreamClientPushSelect.cpp

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

CStreamClientPushSelect::CLayer::CLayer(
    __uint32                layer,
    CSimHandler*            sim,
    CStreamClientPushSelect*    client,
    CLayersInfo*            info,
    void                    (CStreamClientPushSelect::*timerBootHandler)(CTimerInfo*),
    void                    (CStreamClientPushSelect::*timerRegisterHandler)(CTimerInfo*),
    void                    (CStreamClientPushSelect::*timerBufferHandler)(CTimerInfo*)
    )
{
    // Layer
    this->layer = layer;

    // Layer state
    this->stateLayer = LAYER_IDLE;
    this->stateRegister = NOT_REGISTERED;

    // Layers info
    this->info = info;

    // Process function
    this->process = NULL;

    // Timers
    this->timerRegister = new CTimer<CStreamClientPushSelect>(sim, client, timerRegisterHandler);
    this->timerBuffer = new CTimer<CStreamClientPushSelect>(sim, client, timerBufferHandler);
}

CStreamClientPushSelect::CLayer::~CLayer()
{
    // Delete timer
    delete this->timerRegister;
    delete this->timerBuffer;
}

CStreamClientPushSelect::CLayersInfo::CLayersInfo()
{
    this->numBuffering = 0;
    this->numReady = 0;
    this->numRegistered = 0;
}

void CStreamClientPushSelect::CLayer::Start(CChannel* channel)
{
    assert(LAYER_IDLE == this->stateLayer);
    assert(NOT_REGISTERED == this->stateRegister);

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

    // Set layer state
    this->stateLayer = LAYER_BUFFERING;

    // Check the timers
    assert(!this->timerBuffer->IsSet());
    assert(!this->timerRegister->IsSet());

    // Update info
    this->info->IdleToBuffering();

    // Reset streaming info
    this->neighbors.clear();
    this->receivers.clear();
    this->sender = channel->Address();
    this->process = NULL;
}

void CStreamClientPushSelect::CLayer::Stop()
{
    assert(LAYER_IDLE != this->stateLayer);

    // Set the layer state
    this->stateLayer = LAYER_IDLE;

    // Cancel timers
    if(this->timerRegister->IsSet()) this->timerRegister->Cancel();
    if(this->timerBuffer->IsSet()) this->timerBuffer->Cancel();
}

void CStreamClientPushSelect::CLayer::Buffering()
{
    assert(LAYER_READY == this->stateLayer);

    this->stateLayer = LAYER_BUFFERING;

    // Change the info
    this->info->ReadyToBuffering();
}

void CStreamClientPushSelect::CLayer::Ready()
{
    assert(LAYER_BUFFERING == this->stateLayer);

    this->stateLayer = LAYER_READY;

    // Change the info
    this->info->BufferingToReady();
}

CStreamClientPushSelect::CStreamClientPushSelect(
    CSimHandler*                                    sim,
    CInfoPushSelect*                                    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;

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

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

    // Create the buffers
    this->buffer = new CStreamBufferMulti(
        this->info->NumLayers(),
        this->info->StreamBufferSize(),
        this->info->StreamBufferSizeHistory(),
        this->info->MpegGopSize()
        );

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

    // Multicast session
    this->entry = entry;

    // Create the timers
    this->timerPlay = new CTimer<CStreamClientPushSelect>(
        this->sim,
        this,
        &CStreamClientPushSelect::TimerPlay
        );

    this->timerBoot = new CTimer<CStreamClientPushSelect>(
        this->sim,
        this,
        &CStreamClientPushSelect::TimerBoot
        );

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

    // Delegates
    this->delegateRecvFrame = new Delegate2<CStreamClientPushSelect, void, CAddress, CStreamFrame>(this, &CStreamClientPushSelect::RecvFrame);

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

    // Decoder
    this->decoder = NULL;
    this->decoder = new CStreamDecoderFrame(this->delegateRecvFrame);

    // Layers
    this->layers = new CLayer*[this->info->NumLayers()];

    for(__uint32 index = 0; index < this->info->NumLayers(); index++)
    {
        this->layers[index] = new CLayer(
            index,
            this->sim,
            this,
            &this->layersInfo,
            &CStreamClientPushSelect::TimerBoot,
            &CStreamClientPushSelect::TimerRegister,
            &CStreamClientPushSelect::TimerBuffer
            );
    }
}

CStreamClientPushSelect::~CStreamClientPushSelect()
{
    // Delete delegates
    delete this->delegateRecvFrame;

    // Delete coders
    delete this->encoder;
    delete this->decoder;

    // Delete the buffers
    delete this->buffer;

    // Delete the timers
    delete this->timerPlay;
    delete this->timerBoot;

    // Delete the layers
    for(__uint32 index = 0; index < this->info->NumLayers(); index++)
        delete this->layers[index];
    delete[] this->layers;
}

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

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

    // Check the timers are stopped
    assert(!this->timerPlay->IsSet());
    assert(!this->timerBoot->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 registration state (client must not be registered)
    assert(0 == this->layersInfo.NumRegistered());

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

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

    // Start layers
    for(__uint32 index = 0; index < this->info->NumLayers(); index++)
        this->layers[index]->Start(this->channel);

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

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

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

    // Close all receivers
    this->StreamClose();

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

    // Deregister and stop all layers
    for(__uint32 index = 0; index < this->info->NumLayers(); index++)
    {
        // If layer is registered, deregister
        if(REGISTERED == this->layers[index]->StateRegister()) this->StreamDeregister(index);

        this->layers[index]->Stop();
    }

    // Reset layers info
    this->layersInfo.Reset();

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

    // If there are active timer, cancel
    if(this->timerPlay->IsSet()) this->timerPlay->Cancel();
    if(this->timerBoot->IsSet()) this->timerBoot->Cancel();

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

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

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


void CStreamClientPushSelect::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 CStreamClientPushSelect::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->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 CStreamClientPushSelect::RecvMessage(CAddress src, CStreamMessage* message)
{
    // Process messages
    switch(message->MessageType())
    {
    case CStreamMessage::STREAM_MESSAGE_PUSH_MULTI_JOIN: this->RecvMessageJoin(src, type_cast<CStreamMessagePushSelectJoin*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PUSH_MULTI_LEAVE: this->RecvMessageLeave(src, type_cast<CStreamMessagePushSelectLeave*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PUSH_MULTI_CLOSE: this->RecvMessageClose(src, type_cast<CStreamMessagePushSelectClose*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_MULTI_RESPONSE: this->RecvMessageBootResponse(src, type_cast<CStreamMessageBootPushSelectResponse*>(message)); break;
    default: assert(0); /* do nothing */
    }
}

void CStreamClientPushSelect::RecvFrame(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 (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;
    }

    // Calculate the layer of the current frame
    __uint32 layer = this->buffer->IndexToLayer(frame.Index());
    assert(layer < this->info->NumLayers());

    // Process frame by client (client processing is done first)
    assert(this->processFrame);
    (this->*this->processFrame)(frame);

    // Process frame by layer (layer processing is done last)
    assert(this->layers[layer]->ProcessFrame());
    (this->*this->layers[layer]->ProcessFrame())(frame, layer);

    // Statistics
    this->statRecvFrames++;
}

void CStreamClientPushSelect::ProcessFrameFirst(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());

    // Calculate the layer
    __uint32 layer = this->buffer->IndexToLayer(frame.Index());
    
    // Add the frame to the buffer and if the frame is new send it to the receivers
    if(this->buffer->Add(layer, frame)) this->SendStream(frame, layer);

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

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

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

void CStreamClientPushSelect::ProcessFrameBuffering(CStreamFrame frame)
{
    // PLAYBACK Phase 2 : Buffering phase (playback is not started and frames are saved in the buffer)
    
    // Buffering phase finishes when the minimum number of layers is in the ready state

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\t\t[UCAST BUFFERING]\t(frame_index=%u  layers_ready=%u  layers_min=%u)", frame.Index(), this->layersInfo.NumReady(), this->info->NumLayersMin());
#endif

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

    // If the minimum number of layers is in the READY state
    if(this->layersInfo.NumReady() >= this->info->NumLayersMin())
    {
        // Find the first independent frame
        for(__uint32 count = 0; count < this->buffer->Size(); count++)
        {
            // If the current frame is independent
            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 = &CStreamClientPushSelect::ProcessFramePlay;

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

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

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

void CStreamClientPushSelect::ProcessFramePlay(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);

    // Do nothing
}

void CStreamClientPushSelect::ProcessFrameWait(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 there is a minimum number of layers in the READY state

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

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

    // Calculate layer
    __uint32 layer = this->buffer->IndexToLayer(frame.Index());

    // Check if the minimum number of layers is in the READY state
    if(this->layersInfo.NumReady() >= this->info->NumLayersMin())
    {
        // Find the first frame
        for(__uint32 count = 0; count < this->buffer->Count(layer); 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 = &CStreamClientPushSelect::ProcessFramePlay;

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

                // Call the play timer
                this->TimerPlay(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())]++;

#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 CStreamClientPushSelect::ProcessFrameReset(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());

    // Switch all layers to the buffering state
    for(__uint32 index = 0; index < this->info->NumLayers(); index++)
    {
        // If the layer is not in buffering state, switch to buffering state
        if(this->layers[index]->StateLayer() == LAYER_READY) this->layers[index]->Buffering();
        this->layers[index]->Process(&CStreamClientPushSelect::ProcessFrameLayerBuffering);

        // If the layer is registered, deregister
        if(this->layers[index]->StateRegister() == REGISTERED) this->layers[index]->Deregister();

        // If there is a registration timer set for this layer, cancel
        if(this->layers[index]->TimerRegister()->IsSet()) this->layers[index]->TimerRegister()->Cancel();

        // If there is a buffering timer set for this layer, cancel
        if(this->layers[index]->TimerBuffer()->IsSet()) this->layers[index]->TimerBuffer()->Cancel();
    }

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

void CStreamClientPushSelect::ProcessFrameLayerBuffering(CStreamFrame frame, __uint32 layer)
{
    // PROCESS LAYER FRAME Phase 1 : 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(LAYER_BUFFERING == this->layers[layer]->StateLayer());
    
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\t\t[UCAST LAYER BUFFERING]\t(frame_index=%u  layer=%u  frame_layer_index=%u)", frame.Index(), layer, this->buffer->IndexToLayerIndex(frame.Index()));
#endif

    // Add the frame to the buffer
    if(this->buffer->Add(layer, frame))
    {
        // If the frame was added to the buffer

#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_GREEN);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("\t\tBUFFER HIT");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

        // Check if there are enough frames to start playback (the distance between the first frame in the buffer and the received frame)
        if(this->buffer->IndexToLayerIndex(frame.Index()) >= this->buffer->LayerIndex(layer) + this->info->StreamBufferSizeBuffering() - 1)
        {
#ifdef LOG
            CConsole::SetColor(CConsole::LIGHT_GREEN);
            if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
                printf("\tREADY : YES (frame_layer_index=%u buffer_layer_index=%u buffering=%u)",
                this->buffer->IndexToLayerIndex(frame.Index()), this->buffer->LayerIndex(layer), this->info->StreamBufferSizeBuffering()
                );
            CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

            // Change the layer into the ready state
            this->layers[layer]->Ready();

            // Switch to process buffering frames for this layer (phase 3)
            this->layers[layer]->Process(&CStreamClientPushSelect::ProcessFrameLayerReady);

            // If the buffering timer is active, cancel the timer
            if(this->layers[layer]->TimerBuffer()->IsSet()) this->layers[layer]->TimerBuffer()->Cancel();

            //  Set the registration timer for this layer
            if(this->layers[layer]->TimerRegister()->IsSet()) this->layers[layer]->TimerRegister()->Cancel();
            
            assert(this->layers[layer]->StateRegister() == NOT_REGISTERED);
            this->layers[layer]->TimerRegister()->SetAfter(this->info->BootRegisterDelay(), this->layers[layer]);
        }
        else
        {
#ifdef LOG
            CConsole::SetColor(CConsole::LIGHT_RED);
            if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
                printf("\tREADY : NO (frame_layer_index=%u buffer_layer_index=%u buffering=%u)",
                this->buffer->IndexToLayerIndex(frame.Index()), this->buffer->LayerIndex(layer), this->info->StreamBufferSizeBuffering()
                );
            CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
        }

        this->SendStream(frame, layer);
    }
    else
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_RED);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("\t\tBUFFER MISS");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
    }
}

void CStreamClientPushSelect::ProcessFrameLayerReady(CStreamFrame frame, __uint32 layer)
{
    // PLAYBACK Phase 2 : Play phase

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\t\t[UCAST LAYER READY]\t(frame_index=%u  layer=%u  frame_layer_index=%u)", frame.Index(), layer, this->buffer->IndexToLayerIndex(frame.Index()));
#endif

    // Add frame to the buffer
    if(this->buffer->Add(layer, frame))
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_GREEN);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("\t\tBUFFER HIT");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

        this->SendStream(frame, layer);
    }
    else
    {
#ifdef LOG
        CConsole::SetColor(CConsole::LIGHT_RED);
        if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
            printf("\t\tBUFFER MISS");
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
    }
}

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

#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u BOOTSTRAP REQUEST %3u (STREAM)", 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)
    CStreamMessageBootPushSelectRequest* message = new CStreamMessageBootPushSelectRequest(
        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 CStreamClientPushSelect::StreamJoin()
{
    // Streaming STEP 2a : Joining (start streaming from the senders)

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

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

    // Send a join request to the senders for all layers
    for(__uint32 layer = 0; layer < this->info->NumLayers(); layer++)
    {
        this->StreamJoin(layer);

        // Get ready to receive the stream : start streaming session with phase 1
        this->layers[layer]->Process(&CStreamClientPushSelect::ProcessFrameLayerBuffering);

        // Check the layer state is buffering
        assert(this->layers[layer]->StateLayer() == LAYER_BUFFERING);
    }
    
    // Set the processing function to first frame
    this->processFrame = &CStreamClientPushSelect::ProcessFrameFirst;

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

void CStreamClientPushSelect::StreamJoin(__uint32 layer)
{
#ifdef LOG
    if((LOG_ADDRESS == this->address.Address()) && (LOG_CHANNEL == this->channel->Id()))
        printf("\n\tT = %7.3lf PEER %3u JOIN %3u/%u -> %u", this->sim->Time(), this->address.Address(), this->channel->Id(), layer, this->layers[layer]->Sender().Address());
#endif

    CStreamMessagePushSelectJoin* message = new CStreamMessagePushSelectJoin(this->channel->Id(), layer);

    (*this->delegateSendMessage)(this->layers[layer]->Sender(), message);
}

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

    // Send a leave to the sender for each layer
    for(__uint32 layer = 0; layer < this->info->NumLayers(); layer++)
    {
        this->StreamLeave(layer);
    }
}

void CStreamClientPushSelect::StreamLeave(__uint32 layer)
{
    // Streaming buffer underrun for this layer : Leave
    CStreamMessagePushSelectLeave* message = new CStreamMessagePushSelectLeave(this->channel->Id(), layer);

    (*this->delegateSendMessage)(this->layers[layer]->Sender(), message);
}

void CStreamClientPushSelect::StreamRegister(__uint32 layer)
{
    // 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 layer state (layer must be in ready state in order to register)
    if(this->layers[layer]->StateLayer() != LAYER_READY) return;

    // Check the registration state
    if(this->layers[layer]->StateRegister() != NOT_REGISTERED) return;

    CStreamMessageBootPushSelectRegister* message = new CStreamMessageBootPushSelectRegister(
        this->channel->Id(),
        layer,
        this->address,
        this->layers[layer]->Sender());

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

    // Change the registration state
    this->layers[layer]->Register();
}

void CStreamClientPushSelect::StreamDeregister(__uint32 layer)
{
    // Check the resgistration state
    assert(this->layers[layer]->StateRegister() == REGISTERED);

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

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

    // Change the registration state
    this->layers[layer]->Deregister();
}

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

    // Send the close to all layers / all receivers
    for(__uint32 layer = 0; layer < this->info->NumLayers(); layer++)
    {
        for(set<CAddress>::iterator iter = this->layers[layer]->Receivers()->begin(); iter != this->layers[layer]->Receivers()->end(); iter++)
        {
            CStreamMessagePushSelectClose* message = new CStreamMessagePushSelectClose(this->channel->Id(), layer);

            (*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->layers[layer]->Receivers()->clear();
    }

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

void CStreamClientPushSelect::SendStream(CStreamFrame frame, __uint32 layer)
{
    // 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\tT = %7.3lf  SEND FRAME (U)  %u : [ ", this->sim->Time(), frame.Index());
    }
#endif

    for(set<CAddress>::iterator iter = this->layers[layer]->Receivers()->begin(); iter != this->layers[layer]->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 CStreamClientPushSelect::RecvMessageJoin(CAddress src, CStreamMessagePushSelectJoin* 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->layers[message->Layer()]->StateRegister()) { 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
    __bitrate bwLayer = this->channel->Bw() / this->info->NumLayers();
    if(this->bw - this->bwUsed < this->info->StreamBwMargin() * bwLayer)
    {

#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(message->Layer());
        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->layers[message->Layer()]->Receivers()->insert(src);

    // Estimate the increase in bandwidth usage
    this->bwUsed += bwLayer;
}

void CStreamClientPushSelect::RecvMessageLeave(CAddress src, CStreamMessagePushSelectLeave* 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->layers[message->Layer()]->Receivers()->find(src) != this->layers[message->Layer()]->Receivers()->end())
    {
        // Remove the host from the receivers list
        this->layers[message->Layer()]->Receivers()->erase(src);

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

void CStreamClientPushSelect::RecvMessageClose(CAddress src, CStreamMessagePushSelectClose* 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->layers[message->Layer()]->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->layers[message->Layer()]->Neighbors()->size() > 0)
    {
        // Select a new random neighbor as a receiver for this layer
        __uint32 index = CRand::Generate(this->layers[message->Layer()]->Neighbors()->size());
        assert(index < this->layers[message->Layer()]->Neighbors()->size());

        __uint32 idx = 0;
        for(set<CAddress>::iterator iter = this->layers[message->Layer()]->Neighbors()->begin(); iter != this->layers[message->Layer()]->Neighbors()->end(); iter++, idx++)
        {
            if(idx == index)
            {
                this->layers[message->Layer()]->Sender(*iter);                  // Set the sender
                break;
            }
        }

        // Remove the sender from the neighbors list
        this->layers[message->Layer()]->Neighbors()->erase(this->layers[message->Layer()]->Sender());

        // If the number of remaining neighbors is less than the threshold, refresh the list of neighbors
        if(this->layers[message->Layer()]->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 (%u neighbors remaining / %u threshold)", this->sim->Time(), this->address.Address(), this->channel->Id(), message->Layer(), this->layers[message->Layer()]->Neighbors()->size(), this->info->BootRefreshThreshold());
#endif
            // Send a boot push query message to the bootstrap server (the channel address)
            CStreamMessageBootPushSelectRequest* request = new CStreamMessageBootPushSelectRequest(
                this->channel->Id(),
                message->Layer(),
                this->info->BootQueryMax()
                );

            (*this->delegateSendMessage)(this->channel->Address(), request);
        }
    }
    // Else, use the server
    else
    {
        this->layers[message->Layer()]->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 -> %u", this->sim->Time(), this->address.Address(), this->channel->Id(), message->Layer(), this->layers[message->Layer()]->Sender().Address());
#endif

    // Send a JOIN to the new sender
    CStreamMessagePushSelectJoin* reply = new CStreamMessagePushSelectJoin(this->channel->Id(), message->Layer());

    (*this->delegateSendMessage)(this->layers[message->Layer()]->Sender(), reply);
}

void CStreamClientPushSelect::RecvMessageBootResponse(CAddress src, CStreamMessageBootPushSelectResponse* 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 response is for a global bootstrap request
    if(message->ResponseType() == CStreamMessageBootPushSelectResponse::STREAM)
    {
        // If the boot timer is set, cancel (layer bootstrap requests do not have boot timers)
        if(this->timerBoot->IsSet()) this->timerBoot->Cancel();
    }

    // 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

        // Check the response is for the stream
        assert(message->ResponseType() == CStreamMessageBootPushSelectResponse::STREAM);

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

        // Process the response for each layer
        for(__uint32 layer = 0; layer < this->info->NumLayers(); layer++)
        {
            if(message->Count(layer) > 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(layer));
                assert(sender < message->Count(layer));

                // Save the neighbors
                for(__uint32 host = 0; host < sender; host++)
                    this->layers[layer]->Neighbors()->insert(message->Host(layer, host));
                for(__uint32 host = sender+1; host < message->Count(layer); host++)
                    this->layers[layer]->Neighbors()->insert(message->Host(layer, host));

                // Save the sender
                this->layers[layer]->Sender(message->Host(layer, 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

        // Check the response is for one layer only
        assert(message->ResponseType() == CStreamMessageBootPushSelectResponse::LAYER);

        // Process the response
            // Clear the neighbors list
        this->layers[message->Layer()]->Neighbors()->clear();

            // Save the neighbors
        for(__uint32 host = 0; host < message->Count(0); host++)
            this->layers[message->Layer()]->Neighbors()->insert(message->Host(0, host));
    }
}

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

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

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

    // Calculate the layer of the current frame
    __uint32 layer = this->buffer->IndexToLayer(this->buffer->CurrentIndex());

    // If the current layer is READY, check if there is a buffer underrun
    if((this->layers[layer]->StateLayer() == LAYER_READY) && (this->buffer->NumPlay(layer) == 0))
    {
        // Change the layer state to buffering
        assert(this->layers[layer]->StateLayer() == LAYER_READY);
        this->layers[layer]->Buffering();

        // Switch to process buffering frames (phase 2)
        this->layers[layer]->Process(&CStreamClientPushSelect::ProcessFrameLayerBuffering);

        // If the host is registered for this layer, deregister
        if(REGISTERED == this->layers[layer]->StateRegister()) this->StreamDeregister(layer);

        // Set a buffer underrun timer for the layer; if the buffer underrun is not fixed until the timer expires, select a new sender for this layer
        assert(!this->layers[layer]->TimerBuffer()->IsSet());
        this->layers[layer]->TimerBuffer()->SetAfter(this->info->StreamBufferUnderrunTimeout(), this->layers[layer]);
    }

    // Check if the number of layers in the READY state is lower than the minimum
    if(this->layersInfo.NumReady() < this->info->NumLayersMin())
    {
        // Change the state
        this->stateClient = UCAST_STREAM_WAIT;

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

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

#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);
        }
#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);
        }
#endif
    }

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

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

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

    // Bootstrap timer : resend the boostrap query for all layers
    CStreamMessageBootPushSelectRequest* message = new CStreamMessageBootPushSelectRequest(
        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 CStreamClientPushSelect::TimerRegister(CTimerInfo* info)
{
    // Get the layer from the timer info
    assert(info);
    CLayer* layer = type_cast<CLayer*>(info);

    // Register
    this->StreamRegister(layer->Layer());
}

void CStreamClientPushSelect::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 or play state
    if((UCAST_STREAM_WAIT != this->stateClient) && (UCAST_STREAM_PLAY != this->stateClient)) return;

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

    // Get the layer
    CLayer* layer = type_cast<CLayer*>(info);

    // Check the layer is in buffering state
    assert(LAYER_BUFFERING == layer->StateLayer());

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

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

        __uint32 idx = 0;
        for(set<CAddress>::iterator iter = layer->Neighbors()->begin(); iter != layer->Neighbors()->end(); iter++, idx++)
        {
            if(idx == index)
            {
                layer->Sender(*iter);
                break;
            }
        }

        // Remove the sender from the neighbors list
        layer->Neighbors()->erase(layer->Sender());

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

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

    // Send a JOIN to the new sender
    this->StreamJoin(layer->Layer());

    // If the client is in the wait state and the number of ready layers is less than the minimum number,
    // 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)
    if((UCAST_STREAM_WAIT == this->stateClient) && (this->layersInfo.NumReady() < this->info->NumLayersMin()))
    {
        // Change the client state
        this->stateClient = UCAST_STREAM_RESET;

        // Switch to reset
        this->processFrame = &CStreamClientPushSelect::ProcessFrameReset;
    }
}

void CStreamClientPushSelect::Finalize()
{
}