MediaConnection.cpp

Go to the documentation of this file.

/***************************** License **********************************

 Copyright (C) 2008 by Communicative Machines
 http://www.cmlabs.com   All rights reserved.

 This library is free software; you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public
 License as published by the Free Software Foundation; either
 version 2.1 of the License, or (at your option) any later version.

 This library is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 Lesser General Public License for more details.

 You should have received a copy of the GNU Lesser General Public
 License along with this library; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

************************************************************************/
#include "MediaConnection.h"



namespace cmlabs {

PreFetchRequest::PreFetchRequest(const JTime& t, const JString& command, int ms) {
        time = t;
        cmd = command;
        timeout=ms;
}

PreFetchRequest::PreFetchRequest(const JString& id, const JString& command, int ms) {
        this->id = id;
        cmd = command;
        timeout=ms;
}

Object* PreFetchRequest::clone() const {
        PreFetchRequest* req = new PreFetchRequest(time, cmd, timeout);
        req->requestTime = requestTime;
        return req;
}

bool PreFetchRequest::equals(const Object* obj) const {
        if (Object::equals(obj))
                return true;
        if ((obj == NULL) || (!isSameClass(obj)))
                return 0;
//      if (!obj->getClass().equals(this->getClass()))
//              return false;
        PreFetchRequest* req = (PreFetchRequest*) obj;
        return ( (req->time.equals(&time)) && (req->cmd.equals(cmd)) );
}

int PreFetchRequest::getTimeout() {
        JTime now;
        int ms = timeout - (now-requestTime);
        if (ms < 0)
                ms = 50;
        return ms;
}














MediaConnection::MediaConnection(const JString& name, const TCPLocation& serverlocation, long hardBufferMaxSize, long softBufferMaxSize) : JThread() {
        eventQueue = NULL;
        continuousCon = NULL;
        serverCon = NULL;
        myName = name;
        allowMessageDrop = false;
//      clientProfile = new ConnectionProfile(myName);
//      serverProfile = NULL;
        shouldContinue = true;
        server = serverlocation;
//      network = new Network(this);
//      network->name = "MediaConnection_Net";

        mediaServer = NULL;
        if (isInSameExecutable(server)) {
                mediaServer = (MediaServer*) server.pCommunicator;
        }
        if (mediaServer == NULL) {
                mediaStream = new MediaStream(server.name, softBufferMaxSize, hardBufferMaxSize);
        }
        else {
                mediaStream = NULL;
        }
        comRecorder = NULL;
}

MediaConnection::~MediaConnection() {

        comRecorder = NULL;

        if (isInContinuousReceive())
                stopContinuousBackgroundReceive();

        unsubscribeAllChannels();

        if (continuousCon != NULL) {
                continuousCon->reset();
                delete(continuousCon);
                continuousCon = NULL;
        }

        if ((shouldContinue) && (mediaStream != NULL)) {
                shouldContinue = false;
                wait(100);
        }
        
        delete(mediaStream);
        delete(serverCon);

        delete(eventQueue);
        eventQueue = NULL;

//      delete(clientProfile);
//      if (serverProfile != NULL)
//              delete(serverProfile);
//      delete(network);
}

bool MediaConnection::reinit(const JString& name, const TCPLocation& serverlocation, long hardBufferMaxSize, long softBufferMaxSize) {

        comRecorder = NULL;

        // Delete everything
        if (isInContinuousReceive())
                stopContinuousBackgroundReceive();

        unsubscribeAllChannels();

        if (continuousCon != NULL) {
                continuousCon->reset();
                delete(continuousCon);
                continuousCon = NULL;
        }

        if ((shouldContinue) && (mediaStream != NULL)) {
                shouldContinue = false;
                wait(100);
        }
        
        delete(mediaStream);
        delete(serverCon);
        delete(eventQueue);

        // Start again
        eventQueue = NULL;
        continuousCon = NULL;
        serverCon = NULL;
        myName = name;
        shouldContinue = true;
        server = serverlocation;

        mediaServer = NULL;
        if (isInSameExecutable(server)) {
                mediaServer = (MediaServer*) server.pCommunicator;
        }
        if (mediaServer == NULL) {
                mediaStream = new MediaStream(server.name, softBufferMaxSize, hardBufferMaxSize);
        }
        else {
                mediaStream = NULL;
        }

        return init();
}

bool MediaConnection::reinitWithTraining(const JString& name, const TCPLocation& serverlocation, long hardBufferMaxSize, long softBufferMaxSize) {

        comRecorder = NULL;

        // Delete everything
        if (isInContinuousReceive())
                stopContinuousBackgroundReceive();

        unsubscribeAllChannels();

        if (continuousCon != NULL) {
                continuousCon->reset();
                delete(continuousCon);
                continuousCon = NULL;
        }

        if ((shouldContinue) && (mediaStream != NULL)) {
                shouldContinue = false;
                wait(100);
        }
        
        delete(mediaStream);
        delete(serverCon);
        delete(eventQueue);

        // Start again
        eventQueue = NULL;
        continuousCon = NULL;
        serverCon = NULL;
        myName = name;
        shouldContinue = true;
        server = serverlocation;

        mediaServer = NULL;
        if (isInSameExecutable(server)) {
                mediaServer = (MediaServer*) server.pCommunicator;
        }
        if (mediaServer == NULL) {
                mediaStream = new MediaStream(server.name, softBufferMaxSize, hardBufferMaxSize);
        }
        else {
                mediaStream = NULL;
        }

        return initWithTraining();
}

bool MediaConnection::init() {

        if (pingServer() < 0) {
        //      printf("Couldn't ping server...\n"); fflush(stdout);
                return false;
        }

        if (isInSameExecutable(server)) {
//              clientProfile->isDirectlyConnected = true;
                return true;
        }

        if (serverCon == NULL) {
//              printf("No ServerCon..."); fflush(stdout);
                return false;
        }

        return true;
//      return serverCon->trainNetworkProfile();
}

bool MediaConnection::initWithTraining() {

        if (pingServer() < 0) {
        //      printf("Couldn't ping server..."); fflush(stdout);
                return false;
        }

        if (isInSameExecutable(server)) {
//              clientProfile->isDirectlyConnected = true;
                return true;
        }

        if (serverCon == NULL) {
        //      printf("No ServerCon..."); fflush(stdout);
                return false;
        }

//      return true;
        return serverCon->trainNetworkProfile();
}

bool MediaConnection::isConnected() {
        if (isInSameExecutable(server))
                return true;
        if (serverCon == NULL)
                return false;
        return serverCon->isConnected();
}


void MediaConnection::run() {

//      JString id;
        PreFetchRequest* req;
        JTime t;
        // JTime now;
        DataSample* sample;
        JString str;
        int timeout;

        int waittime = 50;

//      printf("MediaConnection starting...\n");

        while ((shouldContinue) && (mediaStream != NULL)) {
                
                // now = JTime();

        /*      if (isInContinuousReceive()) {
                        sample = receiveContinuousSample(50);
                        if (sample != NULL) {
                                mediaStream->addSample(sample);
                                // printf("[%s] Client added network sample at %s - count: %d...\n", (char*) now.printTimeMS(), (char*) sample->getTimestamp().printTime(), mediaStream->getCount());
                        }
                        waittime = 50;
                }
                else
                        waittime = 50;*/

//              if ((id = requests.waitForNewEntryID(waittime)).length() > 0) {

                        if ( (req = (PreFetchRequest*) requests.waitForNewEntry(waittime)) != NULL ) {
                                
                                if (req->cmd.length() == 0) {
                                        sample = this->getDataSample(req->time);
                                        if (sample != NULL) {
                                                mediaStream->addSample(sample);
                                                successRequests.add(req);
                                        }
                                        else
                                                failedRequests.add(req);
                                }
                                else if (req->cmd.equalsIgnoreCase("FirstAfter")) {
                                        timeout = req->getTimeout();
                                        str = JString::format("WAIT_FOR_FIRST_SAMPLE %d", timeout);
                                        sample = askRemoteServerForData(str, req->time.clone(), timeout);
                                        if (sample != NULL) {
                                                mediaStream->addSample(sample);
                                                successRequests.add(req);
                                        }
                                        else
                                                failedRequests.add(req);
                                }
                                else if (req->cmd.equalsIgnoreCase("FirstAfterID")) {
                                        timeout = req->getTimeout();
                                        str = JString::format("WAIT_FOR_FIRST_AFTERID %d", timeout);
                                        sample = askRemoteServerForData(str, req->id.clone(), timeout);
                                        if (sample != NULL) {
                                                mediaStream->addSample(sample);
                                                successRequests.add(req);
                                        }
                                        else
                                                failedRequests.add(req);
                                }
                                else if (req->cmd.equalsIgnoreCase("LastAfter")) {
                                        timeout = req->getTimeout();
                                        str = JString::format("WAIT_FOR_LAST_SAMPLE %d", timeout);
                                        sample = askRemoteServerForData(str, req->time.clone(), timeout);
                                        if (sample != NULL) {
                                                mediaStream->addSample(sample);
                                                successRequests.add(req);
                                        }
                                        else
                                                failedRequests.add(req);
                                }
                        /*      else if (req->cmd.equalsIgnoreCase("StartContinuousReceive")) {
                                        sample = askRemoteServerForData("START_CONTINUOUS_RECEIVE", req->time.clone());
                                        if (isInContinuousReceive())
                                                successRequests.add(req);
                                        else
                                                failedRequests.add(req);
                                }
                                else if (req->cmd.equalsIgnoreCase("StopContinuousReceive")) {
                                        if (continuousCon != NULL) {
                                                continuousCon->reset();
                                                continuousCon = NULL;
                                        }
                                        successRequests.add(req);
                                }*/
                                else
                                        failedRequests.add(req);

                                pendingRequests.remove(pendingRequests.getPos(req));
                                while (successRequests.getCount() > 100)
                                        successRequests.removeFirst();
                                while (failedRequests.getCount() > 100)
                                        failedRequests.removeFirst();

                                newDataAvailable.post();
                        }
                //}
        }
}






bool MediaConnection::reset() {
        comRecorder = NULL;
        // allowMessageDrop = false;
        this->stopContinuousBackgroundReceive();
        requests.removeAll();
        unsubscribeAllChannels();
        // Ideally should empty the mediaStream, but some of the samples could be in use
        return true;
}


Object* MediaConnection::clone() const {
        return NULL;
}

Message* MediaConnection::netObjectReceive(Message *inMsg, NetworkConnection *con) {

        bool noreply = (inMsg->noreply.length() > 0);

        Message* outMsg = NULL;
        if (!shouldContinue) {
                if (!noreply)
                        outMsg = new Message(myName, inMsg->getFrom(), "RECEIVE_ACCEPT");
                delete(inMsg);
                return outMsg;
        }

        DataSample* sample;
        if ( (inMsg->getType().equalsIgnoreCase("BinaryOnly")) &&
             ( (sample = (DataSample*) inMsg->takeObject()) != NULL ) ) {
                if (sample->params.get("SubscriptionEventMatch").length() > 0)
                        addSubscriptionEvent(sample);
                else
                        mediaStream->addSample(sample);
        }
        else {
                // printf("#### Got unexpected mediasample from server! ####\n");
        }

        if (!noreply)
                outMsg = new Message(myName, inMsg->getFrom(), "RECEIVE_ACCEPT");
        delete(inMsg);
        return outMsg;
}

int MediaConnection::getSubscriptionEventQueueSize() {
        if (eventQueue == NULL)
                return -1;
        return eventQueue->getCount();
}

bool MediaConnection::addSubscriptionEvent(DataSample* sample) {
        if (eventQueue == NULL)
                eventQueue = new ObjectQueue();
        // printf("[%s] Receiving channel sample\n", (char*) sample->timestamp.printTimeMS()); fflush(stdout);
        eventQueue->add(sample);
        if (eventQueue->getCount() > 50) {
                Object* obj = eventQueue->retrieveEntry(0);
                delete(obj);
        }
        return true;
}

DataSample* MediaConnection::waitForSubscriptionEvent(int timeout) {
        if (eventQueue == NULL)
                eventQueue = new ObjectQueue();
        DataSample* sample = (DataSample*) eventQueue->waitForNewEntry(timeout);        
        if ((sample != NULL) && (mediaStream != NULL))
                mediaStream->addSample(sample);
        return sample;
}

JString MediaConnection::getName() {
        return myName;
}

bool MediaConnection::allowMessageDropping(bool allow) {
        allowMessageDrop = allow;
        if (serverCon != NULL)
                serverCon->allowMessageDropping(allow);
        return true;
}

bool MediaConnection::handleMultipleWriters(bool allow) {
        if (mediaServer != NULL) {
                return mediaServer->handleMultipleWriters(allow);
        }
        else {
                if (allow)
                        return (askRemoteServerForType("HANDLE_MULTIPLE_WRITERS").equalsIgnoreCase("HANDLE_MULTIPLE_WRITERS_SUCCESS"));
                else
                        return (askRemoteServerForType("HANDLE_SINGLE_WRITER").equalsIgnoreCase("HANDLE_SINGLE_WRITER_SUCCESS"));
        }
}


JString MediaConnection::getServerName() {
        if (mediaServer != NULL) {
                return mediaServer->getName();
        }
        else {
                return askRemoteServerForString("GET_NAME");
        }
}

InfoItem* MediaConnection::getInfo() {
        if (mediaServer != NULL) {
                return mediaServer->getInfo();
        }
        else {
                return (InfoItem*) askRemoteServerForObject("GET_INFO");
        }
}

InfoItem* MediaConnection::getInfoAllStats() {
        if (mediaServer != NULL) {
                return mediaServer->getInfoAllStats();
        }
        else {
                return (InfoItem*) askRemoteServerForObject("GET_INFO_ALL_STATS");
        }
}


bool MediaConnection::isInSameExecutable(const TCPLocation& loc) {

        JString addr;
        if (serverCon != NULL)
                addr = serverCon->getLocalIPAddress();
        else {
                JSocket* sock = new JSocket("", 80);
                addr = sock->getLocalIPAddress();
                delete(sock);
        }
        if ((addr.length() > 0) && (loc.addr.length() > 0)) {
                if (!addr.equalsIgnoreCase(loc.addr))
                        return false;
        }

        JString procid = getProcID();
        if (!procid.equalsIgnoreCase(loc.procid))
                return false;

        return true;
}


//////////////////////////////////////////////////////////////////////////
// API for remote information
//////////////////////////////////////////////////////////////////////////

JTime MediaConnection::getOldestTimestamp() {

        JTime t;
        
        if (mediaServer != NULL) {
                return mediaServer->getOldestTimestamp();
        }
        else {
                JString xml = askRemoteServerForString("GET_OLDEST_TIMESTAMP");
                if (xml.looksLikeXML())
                        t = JTime(xml);
                else
                        t.setInvalid();
        }
        return t;
}

JTime MediaConnection::getNewestTimestamp() {

        JTime t;
        
        if (mediaServer != NULL) {
                return mediaServer->getNewestTimestamp();
        }
        else {
                JString xml = askRemoteServerForString("GET_NEWEST_TIMESTAMP");
                if (xml.looksLikeXML())
                        t = JTime(xml);
                else
                        t.setInvalid();
        }
        return t;
}

DataSample* MediaConnection::getOldestDataSample() {

        DataSample* sample = NULL;
        
        if (mediaServer != NULL) {
                return mediaServer->getOldestSample(myName);
        }
        else {
                sample = askRemoteServerForData("GET_OLDEST_DATASAMPLE", mediaStream->getTimestampList());
                if (sample == NULL) {
                        // We may have a local copy, then
                        return mediaStream->getOldestSample();
                }
        }
        return sample;
}

DataSample* MediaConnection::getNewestDataSample() {

        DataSample* sample = NULL;
        
        if (mediaServer != NULL) {
                return mediaServer->getNewestSample(myName);
        }
        else {
                sample = askRemoteServerForData("GET_NEWEST_DATASAMPLE", mediaStream->getTimestampList());
                if (sample == NULL) {
                        // We may have a local copy, then
                        return mediaStream->getNewestSample();
                }
        }
        return sample;
}

long MediaConnection::getCount() {

        if (mediaServer != NULL) {
                return mediaServer->getCount();
        }
        else {
                JString str = askRemoteServerForString("GET_COUNT");
                return str.toLong();
        }
}

long MediaConnection::getSize() {

        if (mediaServer != NULL) {
                return mediaServer->getTotalDataSize();
        }
        else {
                JString str = askRemoteServerForString("GET_SIZE");
                return str.toLong();
        }
}

ObjectCollection* MediaConnection::getDataSampleList() {

        if (mediaServer != NULL) {
                return mediaServer->getDataSampleList();
        }
        else {
                ObjectCollection* col = (ObjectCollection*) askRemoteServerForObject("GET_SAMPLE_LIST");
                return col;
        }
}

//////////////////////////////////////////////////////////////////////////
// API for retrieval
//////////////////////////////////////////////////////////////////////////


DataSample* MediaConnection::getDataSample(const JString& id) {
        if (mediaServer != NULL) {
                return mediaServer->getDataSample(myName, id);
        }
        else {
                return askRemoteServerForData("GET_SAMPLE_BY_ID", id.clone());
        }
}


//! Either from local buffer or from remote
DataSample* MediaConnection::getDataSample(const JTime& time) {
        if (mediaServer != NULL) {
                return mediaServer->getDataSample(myName, time);
        }
        else {
                DataSample* sample = this->mediaStream->getSampleAt(time);
                if (sample != NULL)
                        return sample;
                sample = askRemoteServerForData("GET_SAMPLE_BY_TIME", time.clone());
                if (sample != NULL) {
                        mediaStream->addSample(sample);
                        return sample;
                }
                else
                        return NULL;
        }
}

//! Either from local buffer or from remote
ObjectCollection* MediaConnection::getDataSamples(const JTime& t1, const JTime& t2) {

        if (mediaServer != NULL) {
                ObjectCollection* col = mediaServer->getDataSamples(myName, t1, t2);
                if (col != NULL)
                        col->noDelete();
                return col;
        }
        else {
                ObjectCollection* col = new ObjectCollection();
                col->add(t1.clone());
                col->add(t2.clone());
                col = askRemoteServerForDataCollection("GET_SAMPLES_BY_TIME", col);
                if (col != NULL) {
                        mediaStream->addSamples(col, false);
                        col->noDelete();
                }
                return col;
        }
}


//////////////////////////////////////////////////////////////////////////
// API for remote adding
//////////////////////////////////////////////////////////////////////////

//! Add one sample
bool MediaConnection::addDataSample(DataSample* sample) {

        if ((comRecorder != NULL) && (!comRecorder->isPaused()))
                comRecorder->addNewSample((DataSample*)sample->clone());

        if (mediaServer != NULL) {
                return mediaServer->addDataSample(myName, sample);
        }
        else {
                return sendSampleToRemoteServer("ADD_SAMPLE", sample);
        }
}

//! Add many samples
bool MediaConnection::addDataSamples(ObjectCollection* samples) {

        if ((comRecorder != NULL) && (samples != NULL) && (!comRecorder->isPaused())) {
                for (DataSample* sample = (DataSample*) samples->getFirst(); sample != NULL; sample = (DataSample*) samples->getNext()) {
                        comRecorder->addNewSample((DataSample*)sample->clone());
                }
        }

        if (mediaServer != NULL) {
                return mediaServer->addDataSamples(myName, samples);
        }
        else {
                return sendSamplesToRemoteServer("ADD_SAMPLES", samples);
        }
}



//////////////////////////////////////////////////////////////////////////
// API for local buffer information
//////////////////////////////////////////////////////////////////////////

JTime MediaConnection::getOldestBufferTimestamp() {

        JTime t;
        if (mediaServer != NULL) {
                return mediaServer->getOldestTimestamp();
        }
        else if (mediaStream != NULL) {
                return mediaStream->getOldestSampleTime();
        }
        else {
                t.setInvalid();
                return t;
        }
}

JTime MediaConnection::getNewestBufferTimestamp() {

        JTime t;
        if (mediaServer != NULL) {
                return mediaServer->getNewestTimestamp();
        }
        else if (mediaStream != NULL) {
                return mediaStream->getNewestSampleTime();
        }
        else {
                t.setInvalid();
                return t;
        }
}

long MediaConnection::getBufferCount() {

        JTime t;
        if (mediaServer != NULL) {
                return mediaServer->getCount();
        }
        else if (mediaStream != NULL) {
                return mediaStream->getCount();
        }
        else {
                return 0;
        }
}






//////////////////////////////////////////////////////////////////////////
// API for waiting for latest information
//////////////////////////////////////////////////////////////////////////

DataSample* MediaConnection::waitForFirstSampleAfter(const JString& id, long ms) {
        if (mediaServer != NULL) {
                return mediaServer->waitForFirstSampleAfter(myName, id, ms);
        }
        else {
                DataSample* sample = NULL;
                if (isInContinuousReceive()) {
                        return mediaStream->waitForFirstSampleAfter(id, ms);
                }
                else {
                        PreFetchRequest* req = new PreFetchRequest(id, "FirstAfterID", ms);
                        if (this->isRequestPending(req)) {
                                sample = waitForRequestToComplete(req, ms);
                        }
                        else if (this->didRequestSucceed(req)) {
                                sample = mediaStream->waitForFirstSampleAfter(id, ms);
                        }
                        else {
                                JString str = JString::format("WAIT_FOR_FIRST_AFTERID %d", ms);
                                sample = askRemoteServerForData(str, id.clone(), ms);
                        }
                        delete(req);
                        if (sample != NULL)
                                mediaStream->addSample(sample);
                        return sample;
                }
        }
        return NULL;
}

DataSample* MediaConnection::waitForLastSampleAfter(const JString& id, long ms) {
        if (mediaServer != NULL) {
                return mediaServer->waitForLastSampleAfter(myName, id, ms);
        }
        else {
                DataSample* sample = NULL;
                if (isInContinuousReceive()) {
                        return mediaStream->waitForLastSampleAfter(id, ms);
                }
                else {
                        PreFetchRequest* req = new PreFetchRequest(id, "LastAfterID", ms);
                        if (this->isRequestPending(req)) {
                                sample = waitForRequestToComplete(req, ms);
                        }
                        else if (this->didRequestSucceed(req)) {
                                // sample = mediaStream->waitForFirstSampleAfter(time, ms);
                                sample = mediaStream->waitForLastSampleAfter(id, ms);
                        }
                        else {
                                JString str = JString::format("WAIT_FOR_LAST_AFTERID %d", ms);
                                sample = askRemoteServerForData(str, id.clone(), ms);
                        }
                        delete(req);
                        if (sample != NULL)
                                mediaStream->addSample(sample);
                        return sample;
                }
        }
        return NULL;
}


DataSample* MediaConnection::waitForFirstSampleAfter(const JTime& time, long ms) {

        if (mediaServer != NULL) {
                return mediaServer->waitForFirstSampleAfter(myName, time, ms);
        }
        else {
                DataSample* sample = NULL;
                if (isInContinuousReceive()) {
                        return mediaStream->waitForFirstSampleAfter(time, ms);
                }
                else {
                        PreFetchRequest* req = new PreFetchRequest(time, "FirstAfter", ms);
                        if (this->isRequestPending(req)) {
                                sample = waitForRequestToComplete(req, ms);
                        }
                        else if (this->didRequestSucceed(req)) {
                                sample = mediaStream->waitForFirstSampleAfter(time, ms);
                        }
                        else {
                                JString str = JString::format("WAIT_FOR_FIRST_SAMPLE %d", ms);
                                sample = askRemoteServerForData(str, time.clone(), ms);
                        }
                        delete(req);
                        if (sample != NULL)
                                mediaStream->addSample(sample);
                        return sample;
                }
        }
        return NULL;
}

DataSample* MediaConnection::waitForLastSampleAfter(const JTime& time, long ms) {

        if (mediaServer != NULL) {
                return mediaServer->waitForLastSampleAfter(myName, time, ms);
        }
        else {
                DataSample* sample = NULL;
                if (isInContinuousReceive()) {
                        return mediaStream->waitForLastSampleAfter(time, ms);
                }
                else {
                        PreFetchRequest* req = new PreFetchRequest(time, "LastAfter", ms);
                        if (this->isRequestPending(req)) {
                                sample = waitForRequestToComplete(req, ms);
                        }
                        else if (this->didRequestSucceed(req)) {
                                // sample = mediaStream->waitForFirstSampleAfter(time, ms);
                                sample = mediaStream->waitForLastSampleAfter(time, ms);
                        }
                        else {
                                JString str = JString::format("WAIT_FOR_LAST_SAMPLE %d", ms);
                                sample = askRemoteServerForData(str, time.clone(), ms);
                        }
                        delete(req);
                        if (sample != NULL)
                                mediaStream->addSample(sample);
                        return sample;
                }
        }
        return NULL;
}

DataSample* MediaConnection::waitForDataSample(const JTime& time, long ms) {

        if (mediaServer != NULL) {
                return mediaServer->waitForSampleAt(myName, time, ms);
        }
        // Has this been requested by a prefetch?
        else {
                DataSample* sample = NULL;
                if (isInContinuousReceive()) {
                        if ( (sample = mediaStream->getSampleAt(time)) != NULL)
                                return sample;
                        JTime start = JTime();
                        long msgone;
                        while ( (msgone = start.getAge()) < ms) {
                                newDataAvailable.wait(ms - msgone);
                                if ( (sample = mediaStream->getSampleAt(time)) != NULL)
                                        return sample;
                        }
                        return mediaStream->getSampleAt(time);
                }
                else {
                        PreFetchRequest* req = new PreFetchRequest(time);
                        if (this->isRequestPending(req)) {
                                sample = waitForRequestToComplete(req, ms);
                        }
                        else if (this->didRequestSucceed(req)) {
                                sample = mediaStream->getSampleAt(time);
                        }
                        else {
                                if (prefetchDataSample(time))
                                        sample = waitForDataSample(time, ms);
                        }
                        delete(req);
                        if (sample != NULL)
                                mediaStream->addSample(sample);
                        return sample;
                }
        }
        return NULL;
}

DataSample* MediaConnection::waitForRequestToComplete(PreFetchRequest* req, long ms) {

        DataSample* sample = NULL;

        if (this->isRequestPending(req)) {
                JTime start;
                while (start.getAge() < ms) {
                        newDataAvailable.wait(ms);
                        if (didRequestSucceed(req))
                                break;
                        // now = JTime();
                }
        }
        
        if (this->didRequestSucceed(req)) {
                sample = mediaStream->getSampleAt(req->time);
        }
        else if (this->didRequestFail(req)) {
        }
        
        return sample;
}



//////////////////////////////////////////////////////////////////////////
// API for prefetching
//////////////////////////////////////////////////////////////////////////


bool MediaConnection::prefetchLastDataSampleAter(const JTime& t) {

        if (mediaServer != NULL) {
                return true;
        }
        else {
                PreFetchRequest* req = new PreFetchRequest(t, "LastAfter");
                if (isRequestPending(req)) {
                        delete(req);
                        return true;
                }
                requests.add(req->clone());
                pendingRequests.add(req);
                return true;
        }
}

bool MediaConnection::prefetchFirstDataSampleAter(const JTime& t) {

        if (mediaServer != NULL) {
                return true;
        }
        else {
                PreFetchRequest* req = new PreFetchRequest(t, "FirstAfter");
                if (isRequestPending(req)) {
                        delete(req);
                        return true;
                }
                requests.add(req->clone());
                pendingRequests.add(req);
                return true;
        }
}





//! Start a retrieve to local buffer
bool MediaConnection::prefetchDataSamples(const JTime& t1, const JTime& t2) {

        if (mediaServer != NULL) {
                return true;
        }
        else {
                ObjectCollection* samples = this->getDataSampleList();
                DataSample* sample;
                for (int n=0; n<samples->getCount(); n++) {
                        if ( (sample = (DataSample*) samples->get(n)) != NULL ) {
                                prefetchDataSample(sample->getTimestamp());
                        }
                }
                delete(samples);
                return true;
        }
}

//! Start a retrieve to local buffer
bool MediaConnection::prefetchDataSample(const JTime& t) {

        if (mediaServer != NULL) {
                return true;
        }
        else {
                PreFetchRequest* req = new PreFetchRequest(t);
                if (isRequestPending(req)) {
                        delete(req);
                        return true;
                }
                requests.add(req->clone());
                pendingRequests.add(req);
                return true;
        }
}




//! Ask the request queue

bool MediaConnection::hasRequestBeenRequested(PreFetchRequest* req) {
        if (pendingRequests.contains(req))
                return true;
        else if (successRequests.contains(req))
                return true;
        else if (failedRequests.contains(req))
                return true;
        else
                return false;
}

bool MediaConnection::isRequestPending(PreFetchRequest* req) {
        return (pendingRequests.contains(req));
}

bool MediaConnection::hasRequestBeenExecuted(PreFetchRequest* req) {
        if (successRequests.contains(req))
                return true;
        else if (failedRequests.contains(req))
                return true;
        else
                return false;
}

bool MediaConnection::didRequestSucceed(PreFetchRequest* req) {
        return (successRequests.contains(req));
}

bool MediaConnection::didRequestFail(PreFetchRequest* req) {
        return (successRequests.contains(req));
}











bool MediaConnection::startContinuousBackgroundReceive() {

        if (mediaServer != NULL)
                return true;

        if (isInContinuousReceive())
                return true;

        if (continuousCon != NULL) {
                continuousCon->reset();
                delete(continuousCon);
        }

        continuousCon = new NetworkConnection(server, new NetMessageProtocol(), this);
        if (!continuousCon->initializeAsReceiver(this)) {
                delete(continuousCon);
                continuousCon = NULL;
                return false;
        }
        if (!continuousCon->isConnected()) {
                continuousCon->reset();
                delete(continuousCon);
                continuousCon = NULL;
                return false;
        }
        return true;

/*      JTime t;
        PreFetchRequest* req = new PreFetchRequest(t, "StartContinuousReceive");
        if (isRequestPending(req)) {
                delete(req);
                return true;
        }
        requests.add(req->clone());
        pendingRequests.add(req);
        return true;*/
}

bool MediaConnection::stopContinuousBackgroundReceive() {

        if (mediaServer != NULL)
                return true;

        if (!isInContinuousReceive())
                return true;

        continuousCon->reset();
        delete(continuousCon);
        continuousCon = NULL;
        return true;

/*      JTime t;
        PreFetchRequest* req = new PreFetchRequest(t, "StopContinuousReceive");
        if (isRequestPending(req)) {
                delete(req);
                return true;
        }
        requests.add(req->clone());
        pendingRequests.add(req);
        return true;*/
}







//////////////////////////////////////////////////////////////////////////
// Private API for prefetching
//////////////////////////////////////////////////////////////////////////







ObjectCollection* MediaConnection::askRemoteServerForDataCollection(const JString& question, Object* obj, int timeout) {
        Message* msg = new Message(myName, server.name, question, obj);
        Message* rmsg = askRemoteServer(msg, timeout);

        if (rmsg == NULL)
                return NULL;

        ObjectCollection* answer = (ObjectCollection*) rmsg->takeObject();

        if (answer != NULL) {
//              try {
                        if ( (answer->getCount() < 0) || (answer->getCount() > 1000000) ) {
                                delete(answer);
                                answer = NULL;
                        }
//              }
//              catch(...) {
//                      delete(answer);
//                      answer = NULL;
//              }
        }

        delete(rmsg);
        return answer;
}

DataSample* MediaConnection::askRemoteServerForData(const JString& question, Object* obj, int timeout) {
        Message* msg = new Message(myName, server.name, question, obj);
        Message* rmsg = askRemoteServer(msg, timeout);

        if (rmsg == NULL)
                return NULL;

        DataSample* answer = (DataSample*) rmsg->takeObject();

        if (answer != NULL) {
//              try {
//                      if (!answer->getClass().equalsIgnoreCase("DataSample")) {
//                              delete(answer);
//                              answer = NULL;
//                      }
//              }
//              catch(...) {
//                      delete(answer);
//                      answer = NULL;
//              }
        }

        delete(rmsg);
        return answer;
}

bool MediaConnection::sendSampleToRemoteServer(const JString& question, DataSample* sample) {
        Message* msg = new Message(myName, server.name, question, sample);
        msg->noreply = "Yes";
        Message* rmsg = askRemoteServer(msg);

        if (rmsg == NULL)
                return false;

        JString type = rmsg->getType();
        delete(rmsg);
        return((type.equalsIgnoreCase("DATA_ACCEPTED")) || (type.equalsIgnoreCase("RECEIVE_ACCEPT")));
}

bool MediaConnection::sendSamplesToRemoteServer(const JString& question, ObjectCollection* samples) {
        Message* msg = new Message(myName, server.name, question, samples);
        msg->noreply = "Yes";
        Message* rmsg = askRemoteServer(msg);

        if (rmsg == NULL)
                return false;

        JString type = rmsg->getType();
        delete(rmsg);
        return((type.equalsIgnoreCase("DATA_ACCEPTED")) || (type.equalsIgnoreCase("RECEIVE_ACCEPT")));
}




Object* MediaConnection::askRemoteServerForObject(const JString& question, int timeout) {
        Message* msg = new Message(myName, server.name, question);
        Message* rmsg = askRemoteServer(msg, timeout);

        if (rmsg == NULL)
                return NULL;

        Object* answer = rmsg->takeObject();
        
        delete(rmsg);
        return answer;
}

JString MediaConnection::askRemoteServerForString(const JString& question, int timeout) {

        Message* msg = new Message(myName, server.name, question);
        Message* rmsg = askRemoteServer(msg, timeout);

        if (rmsg == NULL)
                return "";

        JString answer = rmsg->getContent();
        
        delete(rmsg);
        return answer;
}

JString MediaConnection::askRemoteServerForType(const JString& question, int timeout) {

        Message* msg = new Message(myName, server.name, question);
        Message* rmsg = askRemoteServer(msg, timeout);

        if (rmsg == NULL)
                return "";

        JString answer = rmsg->type;
        
        delete(rmsg);
        return answer;
}

Message* MediaConnection::askRemoteServer(Message* msg, int timeout) {

/*      bool isContinuousStartRequest = false;
        if (msg->getType().equalsIgnoreCase("START_CONTINUOUS_RECEIVE"))
                isContinuousStartRequest = true;*/

        if (msg == NULL) return NULL;

        if (timeout < 0)
                timeout = MSGTIMEOUT;
        else
                timeout += MSGTIMEOUT;

//      Message* backupMsg = (Message*) msg->clone();
        Message* rmsg = NULL;
//      NetworkConnection* con = NULL;

        if (connectionAccess.EnterMutex(timeout)) {
                if (serverCon == NULL) {
                        serverCon = new NetworkConnection(server, new NetMessageProtocol(), NULL);
                        serverCon->allowMessageDropping(allowMessageDrop);
                }
                if (!serverCon->isConnected()) {
                        serverCon->reset();
                        serverCon->restart(server, new NetMessageProtocol(), NULL);
                        serverCon->allowMessageDropping(allowMessageDrop);
                }

                if (msg->noreply.length() > 0) {
                //      rmsg = serverCon->sendReceiveObject(msg, timeout);
                        if (!serverCon->sendObjectWhenReady(msg))
                                delete(msg);
                        rmsg = new Message("", "", "RECEIVE_ACCEPT");
                }
                else {
                        rmsg = serverCon->sendReceiveObject(msg, timeout);
                        if (rmsg == NULL) {
                                // printf("No return message on sendReceiveObject...\n"); fflush(stdout);
                                delete(msg);
                        }
                        else
                                delete(msg);
                }

                connectionAccess.LeaveMutex();
        }
        else {
                // Maybe we should open a second connection along side with the first...
//              printf("Mutex Timeout...\n"); fflush(stdout);
        }

//      delete(backupMsg);

/*      if ( (rmsg != NULL) && (isContinuousStartRequest) && (rmsg->getType().equalsIgnoreCase("STARTED_CONTINUOUS_RECEIVE")) ) {
                continuousCon = serverCon;
        }*/

        return rmsg;
}


ConnectionProfile MediaConnection::getServerConnectionProfile() {

        if (mediaServer != NULL) {
                return mediaServer->getConnectionProfile(myName);
        }
        // Has this been requested by a prefetch?
        else {
                ConnectionProfile* prof = (ConnectionProfile*) askRemoteServerForObject("GET_CONNECTION_PROFILE", MSGTIMEOUT);
                ConnectionProfile profile;
                if (prof == NULL)
                        return profile;
                profile = *prof;
                delete(prof);
                return profile;
        }
}


ConnectionProfile MediaConnection::getConnectionProfile() {
        ConnectionProfile profile;      
        if (isInSameExecutable(server)) {
                profile.isDirectlyConnected = true;
        }
        else if (serverCon != NULL) {
                return serverCon->localProfile;
        }
        return profile;
}


long MediaConnection::pingServer() {

//      if (clientProfile->isDirectlyConnected)
//              return 0;

        if (isInSameExecutable(server))
                return 0;

        Message* msg = new Message(myName, server.name, "PING");
        JTime start;
        Message* rmsg = askRemoteServer(msg);
        if (rmsg == NULL) {
        //      printf("No return message...\n"); fflush(stdout);
                return -1;
        }
        JTime end;
        long res = -1;
        if (rmsg->getType().equalsIgnoreCase("PING_SUCCESS"))
                res = end.microDifference(start);
//      else
//              printf("Got Wrong Reply: '%s'...\n", (char*) rmsg->type); fflush(stdout);
        delete(rmsg);
        return res;
}

/*
long MediaConnection::uploadServerTest(long size) {

        if (clientProfile->isDirectlyConnected)
                return 0;

        DataSample* sample = new DataSample();
        char* dat = new char[size];
        sample->giveData(dat, size);
        Message* msg = new Message(myName, server.name, "UPSTREAM_TEST", sample);

        JTime start;
        Message* rmsg = askRemoteServer(msg);
        if (rmsg == NULL)
                return -1;
        JTime end;
        long res = -1;
        if (rmsg->getType().equalsIgnoreCase("UPSTREAM_TEST_SUCCESS"))
                res = end.microDifference(start);
        delete(rmsg);
        return res;
}

long MediaConnection::downloadServerTest(long size) {

        if (clientProfile->isDirectlyConnected)
                return 0;

        Message* msg = new Message(myName, server.name, "DOWNSTREAM_TEST", JString(size), "");

        JTime start;
        Message* rmsg = askRemoteServer(msg);
        if (rmsg == NULL)
                return -1;
        JTime end;
        long res = -1;
        if (rmsg->getType().equalsIgnoreCase("DOWNSTREAM_TEST_SUCCESS")) {
                DataSample* sample = (DataSample*) rmsg->getObject();
                if (sample != NULL)
                        if (sample->getDataSize() == size)
                                res = end.microDifference(start);
        }
        delete(rmsg);
        return res;
}
*/

bool MediaConnection::isInContinuousReceive() {
        if (mediaServer != NULL)
                return true;
        return ((continuousCon != NULL) && (continuousCon->isConnected()));
}

/*
DataSample* MediaConnection::receiveContinuousSample(int timeout) {

        if (continuousCon == NULL) {
                wait(5);
                return NULL;
        }

        Message* msg = continuousCon->receiveObject(timeout);
        if (msg == NULL)
                return NULL;
        DataSample* sample = (DataSample*) msg->takeObject();
        delete(msg);

        return sample;
}
*/
/*
bool MediaConnection::benchmark() {

        if (isInSameExecutable(server))
                return true;

        double micro = pingServer();
        if (micro < 0)
                return false;

        int n;
        micro = 0;
        for (n=0; n<10; n++) {
                micro += pingServer();
        }
        double avgPing = ((double)micro) / 10.0;

        int multi = 0;
        micro = -1;
        long datasize = 512;
        while (micro < 100000) {
                if (micro <= 0)
                        multi = 2;
                else {
                        multi = (int)(100000.0/micro);
                        if (multi < 2)
                                multi = 2;
                        else if (multi > 64)
                                multi = 64;
                }
                // printf("Multi: %d DataSize: %d -> %d\n", multi, datasize, datasize*multi);
                datasize = datasize * multi;

                micro = uploadServerTest(datasize);
                if (micro < 0)
                        return false;
                micro -= avgPing;
        }

        if (datasize < 10)
                datasize = 10;
        else if (datasize > 1024*1024*1024)
                datasize = 1024*1024*1024;

        double res;
        micro = 0;
        for (n=0; n<5; n++) {
                res = uploadServerTest(datasize);
                if (res < 0)
                        return false;
                micro += (res - avgPing);
        }
        double avgUpload = ((double)micro) / n;

        micro = 0;
        for (n=0; n<5; n++) {
                res = downloadServerTest(datasize);
                if (res < 0)
                        return false;
                micro += (res - avgPing);
        }
        double avgDownload = ((double)micro) / n;

        clientProfile->setSpeed( (long)(datasize*(1000000.0/avgUpload)) , (long)(datasize*(1000000.0/avgDownload)) );

        Message* msg = new Message(myName, server.name, "SET_CONNECTION_PROFILE", clientProfile->clone());
        Message* rmsg = askRemoteServer(msg);
        delete(rmsg);

        if (DEBUGLEVEL(STATUS))
                printf("MediaConnection %s benchmark [%s]: %s\n", (char*) myName, (char*) JString::bytifySize(datasize), (char*) clientProfile->getSpeedString()); fflush(stdout);

        return true;
}
*/











bool MediaConnection::createChannel(const JString& name, const JString& fieldname) {
        if (mediaServer != NULL) {
                return mediaServer->createChannel(name, fieldname);
        }
        else {
                Message* msg = new Message(myName, server.name, "CREATE_CHANNEL",
                        JString::format("%s::%s", (char*) name, (char*) fieldname), "");
                Message* rmsg = askRemoteServer(msg);
                if ((rmsg == NULL) || (!rmsg->type.equalsIgnoreCase("CREATE_CHANNEL_SUCCESS"))) {
                        delete(rmsg);
                        return false;
                }
                delete(rmsg);
                return true;
        }
}

bool MediaConnection::destroyChannel(const JString& name) {
        if (mediaServer != NULL) {
                return mediaServer->destroyChannel(name);
        }
        else {
                Message* msg = new Message(myName, server.name, "DESTROY_CHANNEL", name, "");
                Message* rmsg = askRemoteServer(msg);
                if ((rmsg == NULL) || (!rmsg->type.equalsIgnoreCase("CREATE_CHANNEL_SUCCESS"))) {
                        delete(rmsg);
                        return false;
                }
                delete(rmsg);
                return true;
        }
}

ObjectCollection* MediaConnection::searchChannel(const JString& name, double val1, double val2) {
        if (mediaServer != NULL) {
                ObjectCollection* col = mediaServer->searchChannel(name, val1, val2);
                if (col != NULL) col->noDelete();
                return col;
        }
        else {
                ObjectCollection* col = new ObjectCollection();
                col->add(new JString(name));
                col->add(new JString(val1));
                col->add(new JString(val2));
                return askRemoteServerForDataCollection("GET_SAMPLES_BY_CHANNEL", col);
        }
}

bool MediaConnection::subscribeChannel(const JString& channel, double val1, double val2) {
        JString subscription = JString::format("%s::%f::%f", (char*) channel, val1, val2);
        if (mediaServer != NULL) {
                subscriptions.add(subscription);
                if (!mediaServer->directSubscribers.contains(myName))
                        mediaServer->directSubscribers.put(myName, this);
                return mediaServer->subscribeChannel(myName, channel, val1, val2);
        }
        else {
                Message* msg = new Message(myName, server.name, "SUBSCRIBE_CHANNEL", subscription, "");
                Message* rmsg = askRemoteServer(msg);
                if ((rmsg == NULL) || (!rmsg->type.equalsIgnoreCase("SUBSCRIBE_CHANNEL_SUCCESS"))) {
                        delete(rmsg);
                        return false;
                }
                delete(rmsg);
                subscriptions.add(subscription);
                if (eventQueue == NULL)
                        eventQueue = new ObjectQueue();
                return startContinuousBackgroundReceive();
        }
}

bool MediaConnection::unsubscribeChannel(const JString& channel, double val1, double val2) {
        JString subscription = JString::format("%s::%f::%f", (char*) channel, val1, val2);
        if (mediaServer != NULL) {
                subscriptions.remove(subscription);
                if (subscriptions.getCount() == 0)
                        mediaServer->directSubscribers.remove(myName);
                return mediaServer->unsubscribeChannel(myName, channel, val1, val2);
        }
        else {
                Message* msg = new Message(myName, server.name, "UNSUBSCRIBE_CHANNEL", subscription, "");
                Message* rmsg = askRemoteServer(msg);
                if ((rmsg == NULL) || (!rmsg->type.equalsIgnoreCase("UNSUBSCRIBE_CHANNEL_SUCCESS"))) {
                        delete(rmsg);
                        return false;
                }
                delete(rmsg);
                subscriptions.remove(subscription);
                if (subscriptions.getCount() == 0)
                        stopContinuousBackgroundReceive();
                if (eventQueue != NULL)
                        eventQueue->removeAll();
                return true;
        }
}

bool MediaConnection::unsubscribeAllChannels() {
        if (mediaServer != NULL) {
                subscriptions.removeAll();
                mediaServer->directSubscribers.remove(myName);
                return mediaServer->unsubscribeAllChannels(myName);
        }
        else {
                Message* msg = new Message(myName, server.name, "UNSUBSCRIBE_ALL_CHANNELS");
                Message* rmsg = askRemoteServer(msg);
                if ((rmsg == NULL) || (!rmsg->type.equalsIgnoreCase("UNSUBSCRIBE_ALL_CHANNELS_SUCCESS"))) {
                        delete(rmsg);
                        return false;
                }
                delete(rmsg);
                subscriptions.removeAll();
                stopContinuousBackgroundReceive();
                if (eventQueue != NULL)
                        eventQueue->removeAll();
                return true;
        }
}




















} // namespace cmlabs

---