StreamServerPushSelect.cpp

#include "Headers.h"
#include "StreamServerPushSelect.h"

CStreamServerPushSelect::CStreamServerPushSelect(
    CSimHandler*                                    sim,
    CInfoPushSelect*                                    info,
    CStreamSource*                                  source,
    IDelegate2<void, CAddress, CPacketStream*>*     delegateSendStream,
    IDelegate2<void, CAddress, CStreamMessage*>*    delegateSendMessage
    ) : CStreamServer(sim)
{
    // Set source
    this->source = source;

    // Set info
    this->info = info;

    this->layers = this->info->NumLayers();
    this->gopLength = this->info->MpegGopSize();

    // 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<CStreamServerPushSelect>(
        sim,
        this,
        &CStreamServerPushSelect::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 list of receivers
    this->receivers = new set<CAddress>[this->layers];

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

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

void CStreamServerPushSelect::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 CStreamServerPushSelect::Stop()
{
    // If the timer is set, cancel
    if(this->timer->IsSet()) this->timer->Cancel();
}

void CStreamServerPushSelect::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 CStreamServerPushSelect::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<CStreamMessageBootPushSelectRequest*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_MULTI_REGISTER: this->RecvMessageRegister(src, type_cast<CStreamMessageBootPushSelectRegister*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PUSH_MULTI_DEREGISTER: this->RecvMessageDeregister(src, type_cast<CStreamMessageBootPushSelectDeregister*>(message)); break;
    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;
    }
}

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

void CStreamServerPushSelect::RecvMessageQueryStream(CAddress src, CStreamMessageBootPushSelectRequest* message)
{
    // Process query

    // Get the peer index of the child
    __uint32 child = this->info->HostAddressToIndex(src);

    // Determine the parent for each layer
    CPeer** peers = this->info->ModelSelect()->Select(this->sim->Time(), this->source->Channel()->Id(), child);

    // Set the list of hosts for each layer
    CAddress hosts[MAX_LAYERS];

    for(__uint32 layer = 0; layer < this->layers; layer++)
    {
        hosts[layer] = peers[layer] ? peers[layer]->Address() : this->source->Channel()->Address();
    }

    // Set the peer information
    this->info->ModelSelect()->

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

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

void CStreamServerPushSelect::RecvMessageQueryLayer(CAddress src, CStreamMessageBootPushSelectRequest* message)
{
    // Process query

    // Get the peer index of the child
    __uint32 child = this->info->HostAddressToIndex(src);

    // Determine the parent for each layer
    CPeer* peer = this->info->ModelSelect()->Select(this->sim->Time(), this->source->Channel()->Id(), message->Layer(), child);

    // Reply with a response
    CStreamMessageBootPushSelectResponse* reply = new CStreamMessageBootPushSelectResponse(
        this->source->Channel()->Id(),
        message->Layer(),
        peer ? peer->Address() : this->source->Channel()->Address()
        );

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

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

    // Add the host to 

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

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

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

void CStreamServerPushSelect::RecvMessageJoin(CAddress src, CStreamMessagePushSelectJoin* message)
{
    // Process join

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

void CStreamServerPushSelect::RecvMessageLeave(CAddress src, CStreamMessagePushSelectLeave* message)
{
    // Process leave

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

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