StreamServerPull.cpp

#include "Headers.h"
#include "StreamServerPull.h"
#include "Shuffle.h"

CStreamServerPull::CStreamServerPull(
    CSimHandler*                                                        sim,
    CInfoPull*                                                          info,
    IDelegate5<void, __uint16, __uint16, CAddress, __byte, CPacket*>*   delegateSend,
    IDelegate0<CLink*>*                                                 delegateLink
    ) : CStreamServer(sim)
{
    // Set the info
    this->info = info;

    assert(delegateSend);
    assert(delegateLink);

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

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

    // Create the bootstrap
    this->bootstraps = new CStreamBootPull[this->info->NumChannelsUnicast()];

    // Connection layer
    this->connectionLayer = new CConnectionLayer(
        this->sim,
        this->info->PortConnection(),
        this->info->ServerConnectionsMax(),
        this->delegateSend,
        this->delegateConnectionRecv,
        this->delegateConnectionAccept,
        this->delegateConnectionAccepted,
        this->info->ConnectionSegmentSize());
}

CStreamServerPull::~CStreamServerPull()
{
    // Bootstrap
    delete[] this->bootstraps;

    // Connections
    delete this->connectionLayer;

    // Delete delegates
    delete this->delegateConnectionRecv;
    delete this->delegateConnectionAccept;
    delete this->delegateConnectionAccepted;

    delete this->delegateConnectionSenderOpen;
    delete this->delegateConnectionSenderClose;

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

void CStreamServerPull::Start()
{
    // Do nothing : all requests are asynchronous
}

void CStreamServerPull::Stop()
{
    // Do nothing : all requests are asynchronous
}

void CStreamServerPull::Recv(CAddress srcAddress, __uint16 srcPort, __uint16 dstPort, CPacket* packet)
{
    if(this->info->PortControl() == dstPort) this->RecvMessage(srcAddress, type_cast<CStreamMessage*>(packet));
    if(this->info->PortConnection() == dstPort) this->RecvConnection(srcAddress, srcPort, dstPort, type_cast<CPacketConnection*>(packet));
}

void CStreamServerPull::RecvMessage(CAddress src, CStreamMessage* message)
{   
    // Check the message
    assert(message);

    // Process the message depending on the message type
    switch(message->MessageType())
    {
    case CStreamMessage::STREAM_MESSAGE_BOOT_PULL_REQUEST: this->RecvMessageQuery(src, type_cast<CStreamMessageBootPullRequest*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PULL_REGISTER: this->RecvMessageRegister(src, type_cast<CStreamMessageBootPullRegister*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_BOOT_PULL_DEREGISTER: this->RecvMessageDeregister(src, type_cast<CStreamMessageBootPullDeregister*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PULL_BITMAP_REQUEST: this->RecvMessageBitmapRequest(src, type_cast<CStreamMessagePullBitmapRequest*>(message)); break;
    case CStreamMessage::STREAM_MESSAGE_PULL_SEGMENT_REQUEST: this->RecvMessageSegmentRequest(src, type_cast<CStreamMessagePullSegmentRequest*>(message)); break;
    default: assert(0);
    }
}

void CStreamServerPull::RecvConnection(CAddress srcAddress, __uint16 srcPort, __uint16 dstPort, CPacketConnection* packet)
{
    this->connectionLayer->Recv(srcAddress, srcPort, dstPort, packet);
}

void CStreamServerPull::RecvMessageQuery(CAddress src, CStreamMessageBootPullRequest* message)
{
    // Process query
    
    // Get the index of the unicast stream
    __uint32 stream = this->info->IndexGlobalToUnicast(message->Stream());

    // Reply with a response
    CStreamMessageBootPullResponse* reply;

    // Retrieve a random set of neighbors
        // Get all registered hosts
    CStreamBootPull::List* hosts = this->bootstraps[stream].Fetch();

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

    __uint32 count = 0;

    // Add the server
//  chosts[count++] = this->info->Channel(message->Stream())->Address();

    for(CStreamBootPull::List::iterator iter = hosts->begin(); iter != hosts->end(); iter++)
    {
        if((*iter) != src) chosts[count++] = *iter;
    }

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

    // If the number of found hosts is less than the number of neighbors, also add the server
    if(0 == count)
    {
        chosts[count++] = this->info->Channel(message->Stream())->Address();
    }

    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 CStreamMessageBootPullResponse(message->Stream(), count);

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

        CShuffle shuffle(count);

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

    delete[] chosts;

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

void CStreamServerPull::RecvMessageRegister(CAddress src, CStreamMessageBootPullRegister* message)
{
    // Register a host

    // Get the index of the unicast stream
    __uint32 stream = this->info->IndexGlobalToUnicast(message->Stream());

    this->bootstraps[stream].Register(message->Address());
}

void CStreamServerPull::RecvMessageDeregister(CAddress src, CStreamMessageBootPullDeregister* message)
{
    // Deregister a host

    // Get the index of the unicast stream
    __uint32 stream = this->info->IndexGlobalToUnicast(message->Stream());

    this->bootstraps[stream].Deregister(message->Address());
}

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

    // Generate the bitmap : the server always returns complete bitmap with LOW preference and channel bitrate bandwidth
    __uint64 bitmap = 0xFFFFFFFFFFFFFFFFLL;

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

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

    // The server always return a successfull reply and sends all requested segments through the receiver
    
    // Get the local connection
    CConnectionSender* sender = type_cast<CConnectionSender*>(this->connectionLayer->Get(message->DstId(), message->DstIdEntry()));
    assert(sender);
    assert(sender->State() >= CConnection::OPENED);

    __uint64 success = 0;
    __uint64 fail = 0xFFFFFFFFFFFFFFFFLL;

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

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

        // Get the source
        CStreamSource* source = this->info->StreamSource(message->Stream());

        // Send the requested segments through the receiver
        for(__uint32 sIndex = 0; sIndex < message->Count(); sIndex++)
        {
            // If the segment is requested
            if((message->Bitmap() >> sIndex) & 1)
            {
                __uint32 segmentIndex = message->First() + sIndex;
                __uint32 frameIndexStart = segmentIndex * this->info->StreamSegmentSize();

                // Send all frames from that segment
                for(__uint32 fIndex = 0; fIndex < this->info->StreamSegmentSize(); fIndex++)
                {
                    receiver->Send(source->Fetch(frameIndexStart + fIndex));
                }
            }
        }

        // Set the sucess to all one, fail to zero
        success = 0xFFFFFFFFFFFFFFFFLL;
        fail = 0LL;
    }

    // Reply to the request
    CStreamMessagePullSegmentResponse* reply = new CStreamMessagePullSegmentResponse(
        message->Stream(),
        message->DstId(),
        message->DstIdEntry(),
        message->SrcId(),
        message->SrcIdEntry(),
        message->First(),
        message->Count(),
        success,
        fail
        );
    this->SendMessage(src, reply);
}

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

void CStreamServerPull::ConnectionRecv(CConnectionReceiver* receiver, CPacket* packet)
{
    // Do nothing
}

bool CStreamServerPull::ConnectionAccept(CAddress src, CPacket* packet)
{
    // Accept all connections
    return true;
}

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

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

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

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

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

    // If closing is complete
    if(CConnection::CLOSE_COMPLETE == result)
    {
        // If the connection has a tag, delete the tag
        if(sender->Tag()) delete sender->Tag();
    }
    // Else, do nothing
}

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