StreamServerPushUcastMulti.cpp

#include "Headers.h"
#include "StreamServerPushUcastMulti.h"

CStreamServerPushUcastMulti::CStreamServerPushUcastMulti(
    CSimHandler*                                    sim,
    CStreamSource*                                  source,
    IDelegate2<void, CAddress, CPacketStream*>*     delegateSendStream,
    IDelegate2<void, CAddress, CStreamMessage*>*    delegateSendMessage,
    __uint32                                        layers,
    __byte                                          gopLength
    ) : CStreamServer(sim)
{
    // Set source
    this->source = source;

    // Set lower layer delegates
    this->delegateSendStream = delegateSendStream;
    this->delegateSendMessage = delegateSendMessage;

    // Create the server timer and hook it to the timer method
    this->timer = new CTimer<CStreamServerPushUcastMulti>(
        sim,
        this,
        &CStreamServerPushUcastMulti::Timer);

    // Set the schedule rate (the channel fps)
    this->scheduleRate = this->source->Channel()->Fps();

    // Set the number of layers
    this->layers = layers;

    // Set the GOP length
    this->gopLength = gopLength;

    // Create the bootstrap
    this->bootstrap = new CStreamBootPushMulti(this->layers, this->source->Channel()->Address());

    // Create the list of receivers
    this->receivers = new set<CAddress>[this->layers];

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

CStreamServerPushUcastMulti::~CStreamServerPushUcastMulti()
{
    // Timer
    delete this->timer;
    // Bootstrap
    delete this->bootstrap;
    // Encoder
    delete this->encoder;
    // Receivers
    delete[] this->receivers;
}

void CStreamServerPushUcastMulti::Start()
{
    // Check the server timer is stopped
    assert(!this->timer->IsSet());

    // Synchronize with the stream source
    this->source->Synchronize(this->sim->Time(), this->scheduleTime, this->scheduleFrame);

    // Set the delta frame to zero
    this->deltaFrame = 0;

    // If the schedule time is the current time, call the timer function
    if(this->scheduleTime == this->sim->Time())
        this->Timer(NULL);
    else    // Else, set the timer at the scheduled time
        this->timer->SetAt(this->scheduleTime);
}

void CStreamServerPushUcastMulti::Stop()
{
    // If the timer is set, cancel
    if(this->timer->IsSet()) this->timer->Cancel();
}

void CStreamServerPushUcastMulti::Timer(CTimerInfo* info)
{
    // Process server timer

    // Fetch the frame from the source
    CStreamFrame frame = this->source->Fetch(this->scheduleFrame + this->deltaFrame);

    // Calculate the current layer
    __uint32 layer = (frame.Index() / this->gopLength) % this->layers;

    // Encode the frame to all destinations (the destination is the receiver unicast address)
    for(set<CAddress>::iterator iter = this->receivers[layer].begin(); iter != this->receivers[layer].end(); iter++)
    {
        this->encoder->Encode(*iter, frame);
    }

    // Increment the delta frame
    this->deltaFrame++;

    // Set a new timer
    this->timer->SetAt(this->scheduleTime + ((__time)this->deltaFrame)/this->scheduleRate);
}

void CStreamServerPushUcastMulti::Recv(CAddress src, CStreamMessage* message)
{   
    // Check the message
    assert(message);
    assert(message->Stream() == this->source->Channel()->Id());

    // Process the message depending on the message type
    switch(message->MessageType())
    {
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_MULTI_REQUEST: this->RecvMessageQuery(src, type_cast<CStreamMessageBootPushMultiRequest*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_MULTI_REGISTER: this->RecvMessageRegister(src, type_cast<CStreamMessageBootPushMultiRegister*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_MULTI_DEREGISTER: this->RecvMessageDeregister(src, type_cast<CStreamMessageBootPushMultiDeregister*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PUSH_MULTI_JOIN: this->RecvMessageJoin(src, type_cast<CStreamMessagePushMultiJoin*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PUSH_MULTI_LEAVE: this->RecvMessageLeave(src, type_cast<CStreamMessagePushMultiLeave*>(message)); break;
    }
}

void CStreamServerPushUcastMulti::RecvMessageQuery(CAddress src, CStreamMessageBootPushMultiRequest* message)
{
    // Verify the type of query
    switch(message->RequestType())
    {
    case CStreamMessageBootPushMultiRequest::STREAM: this->RecvMessageQueryStream(src, message); break; // Request for stream (all layers)
    case CStreamMessageBootPushMultiRequest::LAYER: this->RecvMessageQueryLayer(src, message); break;   // Request for layer
    }
}

void CStreamServerPushUcastMulti::RecvMessageQueryStream(CAddress src, CStreamMessageBootPushMultiRequest* message)
{
    // Process query

    // Reply with a response
    CStreamMessageBootPushMultiResponse* reply = new CStreamMessageBootPushMultiResponse(
        this->source->Channel()->Id(),
        this->layers
        );

    for(__uint32 layer = 0; layer < this->layers; layer++)
    {
        // Retrieve a random set of neighbors for each layer
            // Get all registered hosts
        CStreamBootPushMulti::List* hosts = this->bootstrap->Fetch(layer);

        // Copy all hosts locally to remove the source
        CAddress* chosts = new CAddress[hosts->size()];

        __uint32 count = 0;
        for(CStreamBootPushMulti::List::iterator iter = hosts->begin(); iter != hosts->end(); iter++)
        {
            if(!this->bootstrap->IsDependent(layer, src, iter->first)) chosts[count++] = iter->first;
        }

        assert(count <= hosts->size());

        if(count <= message->Count())
        {
            // If the number of hosts is less than or equal to the requested number, put all in the reply
            reply->Set(layer, count);

            for(__uint32 index = 0; index < count; index++)
                reply->Host(layer, index) = chosts[index];
        }
        else
        {
            // If the number of hosts is greater than the requested number, choose a random subset
            reply->Set(layer, message->Count());

            CShuffle shuffle(count);

            for(__uint32 index = 0; index < message->Count(); index++)
                reply->Host(layer, index) = chosts[shuffle[index]];
        }

        delete[] chosts;
    }

    // Send the message through the delegate
    (*this->delegateSendMessage)(src, reply);
}

void CStreamServerPushUcastMulti::RecvMessageQueryLayer(CAddress src, CStreamMessageBootPushMultiRequest* message)
{
    // Process query
    
    // Reply with a response
    CStreamMessageBootPushMultiResponse* reply;

    // Retrieve a random set of neighbors
        // Get all registered hosts
    CStreamBootPushMulti::List* hosts = this->bootstrap->Fetch(message->Layer());

    // Copy all hosts locally to remove any ineligible hosts (i.e. the source hosts and descendants)
    CAddress* chosts = new CAddress[hosts->size()];

    __uint32 count = 0;
    for(CStreamBootPushMulti::List::iterator iter = hosts->begin(); iter != hosts->end(); iter++)
    {
        if(!this->bootstrap->IsDependent(message->Layer(), src, iter->first)) chosts[count++] = iter->first;
    }

    assert(count <= hosts->size());

    if(count <= message->Count())
    {
        // If the number of hosts is less than or equal to the requested number, put all in the reply
        reply = new CStreamMessageBootPushMultiResponse(this->source->Channel()->Id(), message->Layer(), count);

        for(__uint32 index = 0; index < count; index++)
            reply->Host(0, index) = chosts[index];
    }
    else
    {
        // If the number of hosts is greater than the requested number, choose a random subset
        reply = new CStreamMessageBootPushMultiResponse(this->source->Channel()->Id(), message->Layer(), message->Count());

        CShuffle shuffle(count);

        for(__uint32 index = 0; index < message->Count(); index++)
            reply->Host(0, index) = chosts[shuffle[index]];
    }

    delete[] chosts;

    // Send the message through the delegate
    (*this->delegateSendMessage)(src, reply);
}

void CStreamServerPushUcastMulti::RecvMessageRegister(CAddress src, CStreamMessageBootPushMultiRegister* message)
{
    // Register a host
    assert(message->Stream() == this->source->Channel()->Id());

    this->bootstrap->Register(message->Layer(), message->Address(), message->Sender());
}

void CStreamServerPushUcastMulti::RecvMessageDeregister(CAddress src, CStreamMessageBootPushMultiDeregister* message)
{
    // Deregister a host
    assert(message->Stream() == this->source->Channel()->Id());

    this->bootstrap->Deregister(message->Layer(), src);
}

void CStreamServerPushUcastMulti::RecvMessageJoin(CAddress src, CStreamMessagePushMultiJoin* message)
{
    // Process join

    // Add the source to the list of receivers for the specified layer
    this->receivers[message->Layer()].insert(src);
}

void CStreamServerPushUcastMulti::RecvMessageLeave(CAddress src, CStreamMessagePushMultiLeave* message)
{
    // Process leave

    // Remove the source from the list of receivers for the specified layer
    this->receivers[message->Layer()].erase(src);
}

void CStreamServerPushUcastMulti::Finalize()
{
    // Call base class finalizer
    CStreamServer::Finalize();
}