MediaServer.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 "MediaServer.h"
#include "MediaConnection.h"

namespace cmlabs {

MediaServer::MediaServer(const JString& streamname, long hardMaxSize, long softMaxSize, long maxCount, long maxBackupCount) : JThread() {

//      debugLevel = 10;

        directSubscribers.noDelete();
        lastContinuousSend.setInvalid();
        continuousCons.noDelete();
        shouldContinue = true;
        name = streamname;
        if (name.length() == 0)
                name = createUniqueID("stream");

        mediaStream = new MediaStream(name, softMaxSize, hardMaxSize, maxCount, maxBackupCount);

        network = new Network(this);
        network->name = JString::format("MediaServer_%s_Net", (char*) name);
        network->addProtocol(new NetMessageProtocol());
//      network->addProtocol(new NetOriginalMessageProtocol());
        network->addProtocol(new NetHTTPProtocol());
        network->addProtocol(new NetTelnetProtocol(JString("Welcome to MediaServer"), "MediaServer > "));
        network->setPreferredNetProtocol(new NetMessageProtocol());
}

MediaServer::~MediaServer() {
        if (shouldContinue) {
                shouldContinue = false;
                wait(100);
        }

        delete(network);
        delete(mediaStream);
}

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

bool MediaServer::init(int port) {
        if (network->init(port))
                network->start();
        else
                return false;
        return true;
}

JString MediaServer::getName() {
        return name;
}

int MediaServer::getConnectionCount() {
//      return serverNetworkProfiles.getCount() + serverProfiles.getCount();
        return serverProfiles.getCount();
}

ConnectionProfile MediaServer::getConnectionProfile(const JString& client) {
//      ConnectionProfile* profile = (ConnectionProfile*) serverNetworkProfiles.get(client);
        ConnectionProfile* profile = (ConnectionProfile*) serverProfiles.get(client);
//      if (profile == NULL)
//              profile = (ConnectionProfile*) serverProfiles.get(client);
        if (profile == NULL)
                return ConnectionProfile();
        return *profile;
}

bool MediaServer::handleMultipleWriters(bool allow) {
        return mediaStream->handleMultipleWriters(allow);
}

long MediaServer::getTotalDataSize() {
        return mediaStream->getTotalDataSize();
}

long MediaServer::getTotalDataMemUse() {
        return mediaStream->getTotalDataMemUse();
}

int MediaServer::getCount() {
        return mediaStream->getCount();
}

double MediaServer::getTotalDataEverHandled() {
        return mediaStream->getTotalDataEverHandled();
}

double MediaServer::getInputDataRate() {
        return mediaStream->getInputDataRate();
}

double MediaServer::getOutputDataRate() {
        return mediaStream->getOutputDataRate();
}

InfoItem* MediaServer::getInfo() {
        InfoItem* info = mediaStream->getInfo();
        info->setEntry("CurrentConnections", JString(getConnectionCount()));
        return info;
}

InfoItem* MediaServer::getInfoAllStats() {
        ObjectDictionary* profiles = (ObjectDictionary*) serverProfiles.clone();
        InfoItem* info = getInfo();
        info->setObject(profiles);
        return info;
}


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

        //printf("MediaServer get msg: %s\n", (char*)inMsg->printShort());

        Message* outMsg = NULL;
        JString from = inMsg->getFrom();
        JString type = inMsg->getType();
        JString content = inMsg->getContent();

//      long inputBytes = inMsg->getBinarySize(-1);
//      long outputBytes = 0;
        JTime now;
        Collection names;
        bool includeInStats = false;

        if ((type.equals("HTTP")) && (inMsg->getObject() != NULL)) {
                HTMLPage* page = serveHTTPRequest(inMsg->getOrigin(), (HTTPRequest*) inMsg->getObject());
                outMsg = new Message("MediaServer", from, type);
                outMsg->setObject(page);
        }
        else if ((type.equals("HTTPForward")) && (inMsg->getObject() != NULL)) {
                HTMLPage* page = serveHTTPRequest(inMsg->get("httpOrigin"), (HTTPRequest*) inMsg->getObject());
                outMsg = new Message("MediaServer", from, type);
                outMsg->setObject(page);
        }
        else if (type.equals("TelnetInput")) {
                outMsg = serveTelnetInput(from, inMsg->getOrigin(), content);
        }
        else if (type.equalsIgnoreCase("PING")) {
                outMsg = new Message("MediaServer", from, "PING_SUCCESS");
        }
        else {
                JTime t, t2;
                DataSample* sample;
                ObjectCollection* samples;

                outMsg = new Message(name, inMsg->getFrom(), "RECEIVE_ACCEPT");
                if (type.equalsIgnoreCase("GET_COUNT")) {
                        outMsg->setContent(JString(this->getCount()), "");
                }
                else if (type.equalsIgnoreCase("GET_SIZE")) {
                        // printf("Get Size: %d\n", this->getTotalDataSize());
                        outMsg->setContent(JString(this->getTotalDataSize()), "");
                }
                else if (type.equalsIgnoreCase("GET_NEWEST_TIMESTAMP")) {
                        t = this->getNewestTimestamp();
                        if (t.isValid())
                                outMsg->setObject(t.clone());
                }
                else if (type.equalsIgnoreCase("GET_OLDEST_TIMESTAMP")) {
                        t = this->getOldestTimestamp();
                        if (t.isValid())
                                outMsg->setObject(t.clone());
                }
                else if (type.equalsIgnoreCase("GET_NAME")) {
                        outMsg->setContent(this->getName(), "");
                }
                else if (type.equalsIgnoreCase("GET_SAMPLE_LIST")) {
                        includeInStats = true;
                        ObjectCollection* col = this->getDataSampleList();
                        if (col != NULL)
                                outMsg->setObject(col);
                }
                else if (type.equalsIgnoreCase("GET_NEWEST_DATASAMPLE")) {
                        includeInStats = true;
                        sample = this->getNewestSample(from);
                        ObjectCollection* timestamps = (ObjectCollection*)inMsg->getObject();
                        if (sample != NULL) {
                                t = sample->getTimestamp();
                                if ((timestamps != NULL) && (timestamps->contains(&t))) {
                                        // leave the object as NULL = use local copy
                                }
                                else {
                                        outMsg->setObject(sample->clone());
                                }
                        }
                }
                else if (type.equalsIgnoreCase("GET_OLDEST_DATASAMPLE")) {
                        includeInStats = true;
                        sample = this->getOldestSample(from);
                        ObjectCollection* timestamps = (ObjectCollection*)inMsg->getObject();
                        if (sample != NULL) {
                                t = sample->getTimestamp();
                                if ((timestamps != NULL) && (timestamps->contains(&t))) {
                                        // leave the object as NULL = use local copy
                                }
                                else {
                                        outMsg->setObject(sample->clone());
                                }
                        }
                }
                else if (type.equalsIgnoreCase("GET_SAMPLE_BY_TIME")) {
                        includeInStats = true;
                        // printf("MediaServer get sample by time...\n%s\n", (char*)this->mediaStream->print());
                        JTime* tme = (JTime*)inMsg->getObject();
                        if ((tme != NULL) && (tme->isValid())) {
                                sample = this->getDataSample(from, *tme);
                                if (sample != NULL) {
                                //      sample->setInUseBy(createUniqueID("user"), 10000);
                                        outMsg->setObject(sample->clone());
                                        if (DEBUGLEVEL(DEBUG))
                                                printf("MediaServer found sample at time: %s...\n%s\n", (char*) tme->printTimeMS(), (char*)this->mediaStream->print());
                                }
                                else
                                        if (DEBUGLEVEL(DEBUG))
                                                printf("MediaServer didn't find sample at time: %s...\n%s\n", (char*) tme->printTimeMS(), (char*)this->mediaStream->print());
                        }
                        else if (tme != NULL) {
                                if (DEBUGLEVEL(DEBUG))
                                        printf("MediaServer didn't find sample at invalid time: %s...\n%s\n", (char*) tme->printTimeMS(), (char*)this->mediaStream->print());
                        }
                        else {
                                if (DEBUGLEVEL(DEBUG))
                                        printf("MediaServer didn't find sample at invalid time: NULL...\n%s\n", (char*)this->mediaStream->print());
                        }
                }
                else if (type.equalsIgnoreCase("GET_SAMPLE_BY_ID")) {
                        includeInStats = true;
                        JString* ss = (JString*)inMsg->getObject();
                        if ((ss != NULL) && (ss->length() > 0)) {
                                sample = this->getDataSample(from, *ss);
                                if (sample != NULL) {
                                //      sample->setInUseBy(createUniqueID("user"), 10000);
                                        outMsg->setObject(sample->clone());
                                }
                        }
                }
                else if (type.equalsIgnoreCase("GET_SAMPLES_BY_TIME")) {
                        includeInStats = true;
                        ObjectCollection* times = (ObjectCollection*) inMsg->getObject();
                        if ((times != NULL) && (times->getCount() == 2)) {
                                t = *(JTime*) times->get(0);
                                t2 = *(JTime*) times->get(1);
                                samples = this->getDataSamples(from, t, t2);
                                if (samples != NULL) {
                                        samples->noDelete();
                                        ObjectCollection* samplesCopy = (ObjectCollection*) samples->clone();
                                        delete(samples);
                                //      DataSample* sample;
                                /*      for (int n=0; n<samples->getCount(); n++) {
                                                if ( (sample = (DataSample*) samples->get(n)) != NULL) {
                                                        sample->setInUseBy(createUniqueID("user"), 10000);
                                                }
                                        }*/
                                        outMsg->setObject(samplesCopy);
                                }
                        }
                }
                else if (type.equalsIgnoreCase("CREATE_CHANNEL")) {
                        names = inMsg->content.split("::");
                        if (createChannel(names.get(0), names.get(1)))
                                outMsg->setType("CREATE_CHANNEL_SUCCESS");
                        else
                                outMsg->setType("CREATE_CHANNEL_FAILED");
                }
                else if (type.equalsIgnoreCase("DESTROY_CHANNEL")) {
                        if (destroyChannel(inMsg->content))
                                outMsg->setType("DESTROY_CHANNEL_SUCCESS");
                        else
                                outMsg->setType("DESTROY_CHANNEL_FAILED");
                }
                else if (type.equalsIgnoreCase("GET_SAMPLES_BY_CHANNEL")) {
                        includeInStats = true;
                        ObjectCollection* vals = (ObjectCollection*) inMsg->getObject();
                        if ((vals != NULL) && (vals->getCount() == 3)) {
                                JString channel = *(JString*)vals->get(0);
                                double val1 = ((JString*)vals->get(1))->toDouble();
                                double val2 = ((JString*)vals->get(2))->toDouble();
                                samples = this->searchChannel(channel, val1, val2);
                                if (samples != NULL) {
                                        samples->noDelete();
                                        ObjectCollection* samplesCopy = (ObjectCollection*) samples->clone();
                                        delete(samples);
                                //      DataSample* sample;
                                /*      for (int n=0; n<samples->getCount(); n++) {
                                                if ( (sample = (DataSample*) samples->get(n)) != NULL) {
                                                        sample->setInUseBy(createUniqueID("user"), 10000);
                                                }
                                        }*/
                                        outMsg->setObject(samplesCopy);
                                }
                        }
                }
                else if (type.equalsIgnoreCase("SUBSCRIBE_CHANNEL")) {
                        names = inMsg->content.split("::");
                        if (subscribeChannel(from, names.get(0), names.get(1).toDouble(), names.get(2).toDouble()))
                                outMsg->setType("SUBSCRIBE_CHANNEL_SUCCESS");
                        else
                                outMsg->setType("SUBSCRIBE_CHANNEL_FAILED");
                }
                else if (type.equalsIgnoreCase("UNSUBSCRIBE_CHANNEL")) {
                        names = inMsg->content.split("::");
                        if (unsubscribeChannel(from, names.get(0), names.get(1).toDouble(), names.get(2).toDouble()))
                                outMsg->setType("UNSUBSCRIBE_CHANNEL_SUCCESS");
                        else
                                outMsg->setType("UNSUBSCRIBE_CHANNEL_FAILED");
                }
                else if (type.equalsIgnoreCase("UNSUBSCRIBE_ALL_CHANNELS")) {
                        if (unsubscribeAllChannels(from))
                                outMsg->setType("UNSUBSCRIBE_ALL_CHANNELS_SUCCESS");
                        else
                                outMsg->setType("UNSUBSCRIBE_ALL_CHANNELS_FAILED");
                }
                else if (type.equalsIgnoreCase("ADD_SAMPLE")) {
                        includeInStats = true;
                        outMsg->setType("DATA_NOT_ACCEPTED");
                        sample = (DataSample*)inMsg->takeObject();
                        if (sample != NULL) {
                                if (this->addDataSample(from, sample))
                                        outMsg->setType("DATA_ACCEPTED");
                        //      printf("[%s] Server added network sample at %s %d (%s)...\n", (char*) now.printTimeMS(), (char*) sample->getTimestamp().printTime(), sample->timestamp.atomicOffset, (char*) sample->name);
                                fflush(stdout);
                        }
                        if (inMsg->noreply.length() > 0) {
                                delete(outMsg);
                                outMsg = NULL;
                        }
                }
                else if (type.equalsIgnoreCase("ADD_SAMPLES")) {
                        includeInStats = true;
                        outMsg->setType("DATA_NOT_ACCEPTED");
                        samples = (ObjectCollection*)inMsg->takeObject();
                        if (samples != NULL) {
                                if (this->addDataSamples(from, samples))
                                        outMsg->setType("DATA_ACCEPTED");
                                delete(samples);
                        }
                        if (inMsg->noreply.length() > 0) {
                                delete(outMsg);
                                outMsg = NULL;
                        }
                }
                else if (type.startsWithIgnoreCase("WAIT_FOR_FIRST_SAMPLE")) {
                        includeInStats = true;
                        int ms = type.substring(22).toInt();
                        JTime* tme = (JTime*)inMsg->getObject();
                        if ((tme != NULL) && (tme->isValid()) && (ms >= 0)) {
                                sample = this->waitForFirstSampleAfter(from, *tme, ms);
                                if (sample != NULL)
                                        outMsg->setObject(sample->clone());
                                /*else {
                                        JTime now;
                                        addUnitTestLog(JString::format("\nWait %d ms for first failed: %s at %s\n", ms, (char*) tme->print(), (char*) now.print()));
                                        addUnitTestLog(JString::format("Media Content:\n%s\n", (char*) mediaStream->print()));
                                }*/
                        }
                }
                else if (type.startsWithIgnoreCase("WAIT_FOR_FIRST_AFTERID")) {
                        includeInStats = true;
                        int ms = type.substring(23).toInt();
                        JString* ss2 = (JString*)inMsg->getObject();
                        if (ss2 == NULL)
                                ss2 = &inMsg->content;
                        if ((ss2 != NULL) && (ss2->length() > 0) && (ms >= 0)) {
                                sample = this->waitForFirstSampleAfter(from, *ss2, ms);
                                if (sample != NULL) {
                                //      printf(".");
                                        outMsg->setObject(sample->clone());
                                }
                                /*else {
                                        JTime now;
                                        addUnitTestLog(JString::format("\nWait %d ms for first failed: %s at %s\n", ms, (char*) tme->print(), (char*) now.print()));
                                        addUnitTestLog(JString::format("Media Content:\n%s\n", (char*) mediaStream->print()));
                                }*/
                        }
                }
                else if (type.startsWithIgnoreCase("WAIT_FOR_LAST_AFTERID")) {
                        includeInStats = true;
                        now = JTime();
                        int ms = type.substring(22).toInt();
                        JString* ss3 = (JString*)inMsg->getObject();
                        if (ss3 == NULL)
                                ss3 = &inMsg->content;
                        if ((ss3 != NULL) && (ss3->length() > 0) && (ms >= 0)) {
                                sample = this->waitForLastSampleAfter(from, *ss3, ms);
                                if (sample != NULL) {
                                        // printf("Found last sample after %s: (%d) %s\n", (char*) tme->print(), (int)sample, (char*) sample->print());
                                        outMsg->setObject(sample->clone());
                                }
                        }
                }
                else if (type.startsWithIgnoreCase("WAIT_FOR_LAST_SAMPLE")) {
                        includeInStats = true;
                        now = JTime();
                        int ms = type.substring(21).toInt();
                        JTime* tme = (JTime*)inMsg->getObject();
                        if ((tme != NULL) && (tme->isValid()) && (ms >= 0)) {
                                sample = this->waitForLastSampleAfter(from, *tme, ms);
                                if (sample != NULL) {
                                        // printf("Found last sample after %s: (%d) %s\n", (char*) tme->print(), (int)sample, (char*) sample->print());
                                        outMsg->setObject(sample->clone());
                                }
                        }
                }
                else if (type.equalsIgnoreCase("START_CONTINUOUS_RECEIVE")) {
                /*      if (!lastContinuousSend.isValid()) {
                                lastContinuousSend = *(JTime*)inMsg->getObject();
                        }
                        continuousCons.add(con);
                        // Just to be sure to avoid deadlocking...
                        if (!lastContinuousSend.isValid())
                                lastContinuousSend = *(JTime*)inMsg->getObject();
                        outMsg->setType("STARTED_CONTINUOUS_RECEIVE");
                        printf("Starting Continuous Server at %s (%s)...\n", (char*) lastContinuousSend.printTime(), (char*) inMsg->getObject()->print());*/
                        outMsg->setType("CONTINUOUS_RECEIVE_FAILED");
                }
/*              else if (type.equalsIgnoreCase("UPSTREAM_TEST")) {
                        sample = (DataSample*) inMsg->getObject();
                        if (sample == NULL)
                                outMsg->setType("FAILED_NO_DATA");
                        else if (sample->getDataSize() == 0)
                                outMsg->setType("FAILED_WRONG_DATA_SIZE");
                        else
                                outMsg->setType("UPSTREAM_TEST_SUCCESS");
                }
                else if (type.equalsIgnoreCase("DOWNSTREAM_TEST")) {
                        long reqsize = content.toLong();
                        if ((reqsize <= 0) || (reqsize > 1024*1024*10))
                                outMsg->setType("FAILED_WRONG_DATA_SIZE");

                        sample = new DataSample();
                        char* dat = new char[reqsize];
                        sample->giveData(dat, reqsize);
                        outMsg->setObject(sample);
                        outMsg->setType("DOWNSTREAM_TEST_SUCCESS");
                }
                else if (type.equalsIgnoreCase("SET_CONNECTION_PROFILE")) {
                        ConnectionProfile* profile = (ConnectionProfile*) inMsg->takeObject();
                        if (profile != NULL) {
                                clientProfiles.put(from, profile);
                                outMsg->setType("SET_CONNECTION_PROFILE_SUCCESS");
                        }
                        else {
                                outMsg->setType("SET_CONNECTION_PROFILE_FAILED");
                        }
                }*/
                else if (type.equalsIgnoreCase("GET_CONNECTION_PROFILE")) {
                        ConnectionProfile profile = getConnectionProfile(from);
                        outMsg->setObject(profile.clone());
                        outMsg->setType("GET_CONNECTION_PROFILE_SUCCESS");
                }
                else if (type.equalsIgnoreCase("GET_INFO")) {
                        InfoItem* info = getInfo();
                        outMsg->setObject(info);
                        outMsg->setType("GET_INFO_SUCCESS");
                }
                else if (type.equalsIgnoreCase("GET_INFO_ALL_STATS")) {
                        InfoItem* info = getInfoAllStats();
                        outMsg->setObject(info);
                        outMsg->setType("GET_ALL_STATS_SUCCESS");
                }
                else if (type.equalsIgnoreCase("HANDLE_MULTIPLE_WRITERS")) {
                        if (handleMultipleWriters(true)) {
                                outMsg->setType("HANDLE_MULTIPLE_WRITERS_SUCCESS");
                        }
                        else {
                                outMsg->setType("HANDLE_MULTIPLE_WRITERS_FAILED");
                        }
                }
                else if (type.equalsIgnoreCase("HANDLE_SINGLE_WRITER")) {
                        if (handleMultipleWriters(false))
                                outMsg->setType("HANDLE_SINGLE_WRITER_SUCCESS");
                        else
                                outMsg->setType("HANDLE_SINGLE_WRITER_FAILED");
                }
                
        }

//      outputBytes = outMsg->getBinarySize(-1);
//      addNetworkDataStat(from, inputBytes, outputBytes);
        // printf("Server request: '%s' answer: '%s'\n", (char*) type, (char*) outMsg->getType());

        if (includeInStats)
                serverNetworkProfiles.put(from, con->localProfile.clone());
        delete(inMsg);
        return outMsg;
}



bool MediaServer::addDataStat(const JString& name, long upbytes, long downbytes, const JTime& time) {
        ConnectionProfile* profile = (ConnectionProfile*) serverProfiles.get(name);
        if (profile == NULL) {
                profile = new ConnectionProfile(name);
                serverProfiles.put(name, profile);
        }
        return profile->addDataStat(upbytes, downbytes, time);
}
/*
bool MediaServer::addNetworkDataStat(JString name, long upbytes, long downbytes, JTime time) {
        ConnectionProfile* profile = (ConnectionProfile*) serverNetworkProfiles.get(name);
        if (profile == NULL) {
                profile = new ConnectionProfile(name);
                serverNetworkProfiles.put(name, profile);
        }
        return profile->addDataStat(upbytes, downbytes, time);
}
*/
TCPLocation MediaServer::getLocation() {
        TCPLocation loc = TCPLocation(network->getLocalIPAddress(), network->getPort(), name);
        loc.procid = getProcID();
        loc.pCommunicator = (void*) this;
        return loc;
}


void MediaServer::run() {

        int n;
        DataSample* sample = NULL;
        JTime lastAct;
        NetworkConnection* con;
        Message* msg;
        JTime now;
        int count;
        ObjectCollection* subs;
        MediaConnection* mcon;
        JString remotename;

        while (shouldContinue) {
        //      printf("(%d/%d)...\n", network->remoteReceiverCons->getCount(), directSubscribers.getCount()); fflush(stdout);
                if ((network->remoteReceiverCons->getCount() > 0) || (directSubscribers.getCount() > 0)) {
                        if (!lastContinuousSend.isValid())
                                lastContinuousSend = JTime();
                        sample = waitForFirstSampleAfter(lastContinuousSend, 50);
                        if (sample != NULL) {
                                // printf("[%s] New Server Sample (%d/%d)...\n", (char*) sample->timestamp.printTimeMS(), network->remoteReceiverCons->getCount(), directSubscribers.getCount()); fflush(stdout);
                                lastContinuousSend = sample->getTimestamp();
                                count = 0;
                                for (n=0; n<directSubscribers.getCount(); n++) {
                                        remotename = directSubscribers.getKey(n);
                                        subs = (ObjectCollection*) subscriptions.get(remotename);
                                        mcon = (MediaConnection*) directSubscribers.get(n);
                                        if ((subs != NULL) && (mcon != NULL)) {
                                                if (subscriptionMatch(subs, sample)) {
                                                        mcon->addSubscriptionEvent(sample);
                                                        addDataStat(remotename, 0, sample->size);
                                                        count++;
                                                }
                                        }
                                }
                                for (n=0; n<network->remoteReceiverCons->getCount(); n++) {
                                        con = (NetworkConnection*) network->remoteReceiverCons->get(n);
                                        if (con != NULL) {
                                        //      printf("[%s] New Server Sample for con %s...\n", (char*) sample->timestamp.printTimeMS(), (char*) con->remoteName); fflush(stdout);
                                                if (con->isConnected()) {
                                                        // Check if this is a subscriber
                                                        subs = (ObjectCollection*) subscriptions.get(con->remoteName);
                                                        if (subs != NULL) {
                                                                if (!subscriptionMatch(subs, sample)) {
                                                                //      printf("[%s] Not Found sample for channel...\n", (char*) sample->timestamp.printTimeMS()); fflush(stdout);
                                                                        continue; // if not, don't send
                                                                }
                                                                sample->params.put("SubscriptionEventMatch", "Matched");
                                                                // printf("[%s] Found sample for channel...\n", (char*) sample->timestamp.printTimeMS()); fflush(stdout);
                                                        }
                                                        // else just send the sample
                                                        msg = new Message(this->getName(), con->remoteName, "BinaryOnly", sample->clone());
                                                        msg->noreply = "Yes";
                                                        addDataStat(con->remoteName, 0, sample->size);
                                                        if (con->sendObjectWhenReady(msg)) {
                                                                count++;
                                                        }
                                                        if (DEBUGLEVEL(DETAILEDSTATUS))
                                                                printf("Continuous %s ---> %s\n", (char*) con->remoteName, (char*) sample->print());
                                                }
                                                else {
                                                        if (DEBUGLEVEL(STATUS))
                                                                printf("Continuous STOP %s ---> %s\n", (char*) con->remoteName, (char*) sample->print());
                                                        con->reset();
                                                        network->remoteReceiverCons->removeNoDelete(n);
                                                        n--;
                                                        if (network->remoteReceiverCons->getCount() == 0)
                                                                lastContinuousSend.setInvalid();
                                                }
                                        }
                                }
                                if (DEBUGLEVEL(DEBUG)) {
                                        now = JTime();
                                        printf("[%s] MediaServer continuous sent sample to %d cons at %s...\n", (char*) now.printTimeMS(), count, (char*) lastContinuousSend.printTime());
                                }
                        }
                        else {
                                // printf("MediaServer found nothing in continuous mode [%d]...\n", this->getCount());
                        }
                }
                else {
                        lastContinuousSend.setInvalid();
                        wait(20);
                }
        }
}








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

JTime MediaServer::getOldestTimestamp() {
        return mediaStream->getOldestSampleTime();
}

JTime MediaServer::getNewestTimestamp() {
        return mediaStream->getNewestSampleTime();
}

ObjectCollection* MediaServer::getDataSampleList() {
        return mediaStream->getDataSampleList();
}


//////////////////////////////////////////////////////////////////////////
// API for retrieving
//////////////////////////////////////////////////////////////////////////

DataSample* MediaServer::getOldestSample(const JString& clientname) {
        DataSample* sample = mediaStream->getOldestSample();
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}

DataSample* MediaServer::getNewestSample(const JString& clientname) {
        DataSample* sample = mediaStream->getNewestSample();
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}

//! Get one sample
DataSample* MediaServer::getDataSample(const JString& clientname, const JTime& time) {
        DataSample* sample = getDataSample(time);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::getDataSample(const JTime& time) {
        return mediaStream->getSampleAt(time);
}

//! Get many samples
ObjectCollection* MediaServer::getDataSamples(const JString& clientname, const JTime& t1, const JTime& t2) {
        ObjectCollection* col = getDataSamples(t1, t2);
        long bytes = 0;
        DataSample* sample;
        if (col != NULL) {
                for (int n=0; n<col->getCount(); n++)
                        if ( (sample = (DataSample*) col->get(n)) != NULL)
                                bytes += sample->getDataSize();
        }
        addDataStat(clientname, 0, bytes);
        return col;
}
ObjectCollection* MediaServer::getDataSamples(const JTime& t1, const JTime& t2) {
        return mediaStream->getSamplesBetween(t1, t2);
}

DataSample* MediaServer::waitForSampleAt(const JString& clientname, const JTime& time, long ms) {
        JTime start;
        DataSample* sample = waitForFirstSampleAfter(time-1, ms);
        if ((sample == NULL) || (sample->timestamp != time)) {
                JTime now;
                sample = NULL;
                long dif = now - start;
                if (dif < ms) {
                        wait(ms-dif);
                        sample = getDataSample(time);
                }
        }
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::waitForSampleAt(const JTime& time, long ms) {
        DataSample* sample = waitForFirstSampleAfter(time-1, ms);
        if ((sample == NULL) || (sample->timestamp != time))
                return NULL;
        return sample;
}

DataSample* MediaServer::getSample(const JString& clientname, const JString& id) {
        DataSample* sample = mediaStream->getSample(id);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::getFirstSampleAfter(const JString& clientname, const JString& id) {
        DataSample* sample = mediaStream->getFirstSampleAfter(id);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::waitForFirstSampleAfter(const JString& clientname, const JString& id, long ms) {
        DataSample* sample = mediaStream->waitForFirstSampleAfter(id, ms);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::waitForLastSampleAfter(const JString& clientname, const JString& id, long ms) {
        DataSample* sample = mediaStream->waitForLastSampleAfter(id, ms);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}

DataSample* MediaServer::getSample(const JString& id) {
        return mediaStream->getSample(id);
}
DataSample* MediaServer::getFirstSampleAfter(const JString& id) {
        return mediaStream->getFirstSampleAfter(id);
}
DataSample* MediaServer::waitForFirstSampleAfter(const JString& id, long ms) {
        return mediaStream->waitForFirstSampleAfter(id, ms);
}
DataSample* MediaServer::waitForLastSampleAfter(const JString& id, long ms) {
        return mediaStream->waitForLastSampleAfter(id, ms);
}


DataSample* MediaServer::waitForFirstSampleAfter(const JString& clientname, const JTime& time, long ms) {
        DataSample* sample = waitForFirstSampleAfter(time, ms);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::waitForFirstSampleAfter(const JTime& time, long ms) {
        return mediaStream->waitForFirstSampleAfter(time, ms);
}

DataSample* MediaServer::waitForLastSampleAfter(const JString& clientname, const JTime& time, long ms) {
        DataSample* sample = waitForLastSampleAfter(time, ms);
        if (sample != NULL)
                addDataStat(clientname, 0, sample->getDataSize());
        return sample;
}
DataSample* MediaServer::waitForLastSampleAfter(const JTime& time, long ms) {
        return mediaStream->waitForLastSampleAfter(time, ms);
}


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

//! Add one sample
bool MediaServer::addDataSample(const JString& clientname, DataSample* sample) {
        if (sample != NULL)
                addDataStat(clientname, sample->getDataSize(), 0);
        return addDataSample(sample);
}
bool MediaServer::addDataSample(DataSample* sample) {
        return mediaStream->addSample(sample);
}

//! Add many samples
bool MediaServer::addDataSamples(const JString& clientname, ObjectCollection* samples) {

        long bytes = 0;
        DataSample* sample;
        if (samples != NULL) {
                for (int n=0; n<samples->getCount(); n++) {
                        if ( (sample = (DataSample*) samples->get(n)) != NULL) {
                                bytes += sample->getDataSize();
                                addDataStat(clientname, sample->getDataSize(), 0);
                        }
                }
        }
        return addDataSamples(samples);
}
bool MediaServer::addDataSamples(ObjectCollection* samples) {
        return mediaStream->addSamples(samples);
}















HTMLPage* MediaServer::serveHTTPRequest(const JString& origin, HTTPRequest* request) {

        HTMLPage* page = new HTMLPage();

        page->title = "MediaServer Overview";

        page->body = "<H1>Status</H1>\n";

        return page;
}

Message* MediaServer::serveTelnetInput(const JString& id, const JString& origin, const JString& text) {

        JString out;
        Message* msg = new Message("MediaServer", "", "");
        JTime now;
        msg->set("NewPromptText", JString::format("MediaServer [%s] ", (char*) now.printTime()));

        if (text.length() == 0)
                return msg;

        Collection oc = text.split(" ");

        JString cmd = oc.get(0);
        JString arg = oc.get(1);

        if (cmd.equalsIgnoreCase("list")) {
                out = "";
        }
        else if (cmd.equalsIgnoreCase("status"))
                out = "";
        else
                out = JString("Unknown command: ") + cmd;

        msg->setContent(out, "");
        return msg;
}

bool MediaServer::unitTest() {
        
        addUnitTestLog(JString::format("Starting MediaServer test..."));
        JTime now;
//      printf("\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b\b"); fflush(stdout);
        printf("\n  Performance Tests:\n");

        int count, k, c;
        MediaConnection* con1, *con2;
        JTime t;
        DataSample* sample, *sample2;
        JTime tt;
        int n;
        JTime start;

        int p = 23545;
        while (!this->init(p)) {
                p++;
        }

        TCPLocation loc = this->getLocation();

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();
        
        addUnitTestLog(JString::format("Testing location..."));
        if (!con1->isInSameExecutable(loc))
                return false;

        // printf("1"); fflush(stdout);

        t.reset();
        addUnitTestLog(JString::format("Testing connection 1..."));
        t = con1->getOldestTimestamp();
        if (t.isValid()) return false;
        t = con1->getNewestTimestamp();
        if (t.isValid()) return false;
        t = con1->getOldestBufferTimestamp();
        if (t.isValid()) return false;
        t = con1->getNewestBufferTimestamp();
        if (t.isValid()) return false;

//      printf("2"); fflush(stdout);

        addUnitTestLog(JString::format("Testing connection 2..."));
        t = con2->getOldestTimestamp();
        if (t.isValid()) return false;
        t = con2->getNewestTimestamp();
        if (t.isValid()) return false;
        t = con2->getOldestBufferTimestamp();
        if (t.isValid()) return false;
        t = con2->getNewestBufferTimestamp();
        if (t.isValid()) return false;

        t = JTime();
        sample = new DataSample();
        char* data = new char[20];
        strcpy(data, "Hello World");
        sample->setTimestamp(t);
        sample->giveData(data, 20, 25);
        con1->addDataSample(sample);

//      printf("3"); fflush(stdout);

        addUnitTestLog(JString::format("Testing connection 1 without hello world..."));
        sample2 = con2->getDataSample(t);
        if (sample != sample2)
                return false;

//      printf("4"); fflush(stdout);
        
        addUnitTestLog(JString::format("Testing connections..."));
        t = JTime();
        tt = t;
        for (n=0; n<5; n++) {
                // wait(1200);
                // tt = tt + 2500;
                // tt = JTime();
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                // sample2->setTimestamp(tt);
                if (!con1->addDataSample(sample2))
                        addUnitTestLog(JString::format("Couln't add: %s", (char*)tt.print()));
                else
                        addUnitTestLog(JString::format("Added: %s", (char*)tt.print()));
        }

        now = JTime();

        t = tt - 10000000;
        tt = tt + 1;
        ObjectCollection* col = con2->getDataSamples(t, tt);
        
        if (col == NULL) {
                addUnitTestLog(JString::format("Couldn't get samples..."));
                return false;
        }
//      printf("1[%d]", col->getCount()); fflush(stdout);
        if (col->getCount() < 6) {
                addUnitTestLog(JString::format("Samples count wrong: %d...", col->getCount()));
                return false;
        }
//      printf("2"); fflush(stdout);

        delete(col);

        col = con2->getDataSampleList();
        if (col == NULL) {
                addUnitTestLog(JString::format("Couldn't get samplelist..."));
                return false;
        }
        if (col->getCount() < 6) {
                addUnitTestLog(JString::format("Samplelist count wrong: %d...", col->getCount()));
                return false;
        }
        delete(col);
//      printf("3"); fflush(stdout);

        sample2 = NULL;
        while ( (sample = con1->waitForFirstSampleAfter(t, 1000)) != NULL ) {
                if (sample == sample2) {
                        addUnitTestLog(JString::format("Got same sample twice from stream..."));
                        return false;
                }
                t = sample->timestamp;
                sample2 = sample;
        }
        


        delete(con1);
        delete(con2);

//      printf("4"); fflush(stdout);















        addUnitTestLog("Starting networking test...");
//      printf("1"); fflush(stdout);

        // Now use networking...

        loc.procid = 0;
        con1 = new MediaConnection("Test3", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining()) {
                addUnitTestLog("Could not init network con1");
                return false;
        }
        con1->start();
        con2 = new MediaConnection("Test4", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining()) {
                addUnitTestLog("Could not init network con2");
                return false;
        }
        con2->start();
        if (con1->isInSameExecutable(loc)) {
                addUnitTestLog("Network con1 in same executable, shouldn't be...");
                return false;
        }
        if (con2->isInSameExecutable(loc)) {
                addUnitTestLog("Network con2 in same executable, shouldn't be...");
                return false;
        }

//      printf("2"); fflush(stdout);
        JString str = con1->getServerName();
        if (str.length() == 0)
                return false;

//      printf("3"); fflush(stdout);
        if (con1->getCount() != 6)
                return false;

//      printf("4"); fflush(stdout);
        col = con1->getDataSampleList();
        if (col == NULL)
                return false;
//      printf("5"); fflush(stdout);
        if (col->getCount() != 6)
                return false;
//      printf("6"); fflush(stdout);

        addUnitTestLog(JString::format("Got Sample List OK [%d]:\n%s\n", col->getCount(), (char*) col->printListLine("\n")));

        delete(col);


        // debugLevel=9;

        t = tt - 10000000;
        col = con1->getDataSamples(t, tt);
        if (col == NULL)
                return false;
//      printf("1[%d]", col->getCount()); fflush(stdout);
        if (col->getCount() < 6)
                return false;
//      printf("2"); fflush(stdout);

        sample = (DataSample*) col->get(0);
        if (sample == NULL)
                return false;

        if (sample->getDataLink() == NULL)
                return false;

        JString s = sample->getDataLink();
        if (!s.equals("Hello World"))
                return false;

        JTime firstTime = sample->getTimestamp();
        if (!firstTime.isValid())
                return false;

        addUnitTestLog("Got Sample Collection OK...");

        sample2 = con1->getDataSample(firstTime);
        if (sample2 == NULL) {
                addUnitTestLog(JString::format("Sample2 NULL [%s]", (char*) firstTime.print()));
                return false;
        }

        if (strcmp(sample->getDataLink(), sample2->getDataLink()) != 0 ) {
                addUnitTestLog(JString::format("Sample1 != Sample2   '%s' != '%s' ", (char*) sample->getDataLink(), (char*)sample2->getDataLink()));
                return false;
        }

        addUnitTestLog("Got Individual Sample OK...");

        sample = (DataSample*) col->get(1);
        if (sample == NULL)
                return false;

        sample2 = con1->waitForFirstSampleAfter(firstTime, 1000);
        if (sample2 == NULL)
                return false;

        addUnitTestLog("Waited for First After Sample OK...");

        if (strcmp(sample->getDataLink(), sample2->getDataLink()) != 0 )
                return false;

        addUnitTestLog("Waited for First After Sample Tested OK...");

        sample = (DataSample*) col->getLast();
        sample2 = con1->waitForLastSampleAfter(firstTime, 1000);
        if (sample2 == NULL)
                return false;

        if (strcmp(sample->getDataLink(), sample2->getDataLink()) != 0 )
                return false;

        addUnitTestLog("Waited for Last After Sample OK...");

        sample = (DataSample*) sample->clone();
        delete(col);


        DataSample* sample3;
        sample2 = NULL;
        while ( (sample3 = con1->waitForFirstSampleAfter(t, 1000)) != NULL ) {
                if (sample3 == sample2) {
                        addUnitTestLog(JString::format("Got same sample twice from stream..."));
                        return false;
                }
                t = sample3->timestamp;
                sample2 = sample3;
        }

        delete(con1);
        delete(con2);










        loc = this->getLocation();

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();

        count = 100;

        JTime firstSampleTime = tt + 1;
        start = JTime();
        for (n=0; n<count; n++) {
                tt = tt + 1;
                sample2 = (DataSample*) sample->clone();
                sample2->setTimestamp(tt);
                if (!con1->addDataSample(sample2)) {
                        break;
                }
        }
        if (n < count) {
                addUnitTestLog(JString::format("Couldn't add 10k samples..."));
                return false;
        }

        printf("    %s %s (%d)\n", (char*) JString("Direct add samples:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*n)), 30, 1), con1->getCount()-6);

        sample2 = con2->getDataSample(firstSampleTime);
        if (sample2 == NULL) {
                addUnitTestLog(JString::format("No samples after 10k"));
                return false;
        }
        
        n=0;
        start = JTime();
        while (sample2 != NULL) {
                n++;
                sample2 = con2->getDataSample(sample2->timestamp+1);
        }

        printf("    %s %s (%d)\n", (char*) JString("Direct fetch samples:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*n)), 30, 1), n);

        delete(con1);
        delete(con2);














        loc = this->getLocation();
        loc.procid = 0;

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();

        now = JTime();
        addUnitTestLog(JString::format("Starting client continuous at %s...\n", (char*) now.printTime()));

        if (!con2->startContinuousBackgroundReceive())
                return false;
        wait(100);

        if (con2->getBufferCount() > 0)
                return false;

        count = 100;

        start = JTime();

        t = JTime();
        tt = t;
        for (k=0; k<count; k++) {
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                sample2->setTimestamp(tt);
                // printf("[%d] %s\n", k, (char*) sample2->print());
                if (!con1->addDataSample(sample2)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
                else {
                //      addUnitTestLog(JString::format("Added 2: %s", (char*)tt.print()));
                        // printf("Added 2: %s\n", (char*)tt.print());
                        // printf("[%s] Tester sent network sample at %s - count: %d...\n", (char*) now.printTimeMS(), (char*) sample2->getTimestamp().printTime(), k);
                }
                // wait(40);
        }
        printf("    %s %s (%d)\n", (char*) JString("Net add samples:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*count)), 30, 1), count);

//      wait(1000);

        start = JTime();
        c = con2->getBufferCount();
        while (c < count) {
                now = JTime();
                if (now - start > count*1000) {
                        addUnitTestLog(JString::format("Continuous got %d samples, not %d...\n", c, count));
                        return false;
                }
                wait(5);
                c = con2->getBufferCount();
        }
        now = JTime();
//      double sps = (((double)now.microDifference(start))/(count * 1000.0)) - 40.0;
//      printf("[%s] 5 continuous samples return took %d ms\n", (char*) now.printTimeMS(), now - start);
        printf("    %s %s (%d)\n", (char*) JString("Net fetch samples:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*c)), 30, 1), c);
        

        if (!con2->stopContinuousBackgroundReceive()) {
                addUnitTestLog("Could not stop continuous streaming...");
                return false;
        }

        wait(100);

        for (k=0; k<5; k++) {
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                sample2->setTimestamp(tt);
                if (!con1->addDataSample(sample2))
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                else
                        addUnitTestLog(JString::format("Added 2: %s", (char*)tt.print()));
        }

        wait(100);

        if (con2->getBufferCount() != count) {
                addUnitTestLog(JString::format("Continuous got %d samples, not %d...\n", c, count));
                return false;
        }

        delete(con1);
        delete(con2);











        // ####################################
        addUnitTestLog("Starting nested sample test...");

        loc = this->getLocation();
        loc.procid = 0;

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();

        count = 100;
        start = JTime();
        tt = start;
        for (k=0; k<count; k++) {
                tt = tt + 1;
                sample2 = new DataSample();
                sample2->setTimestamp(tt);
                sample2->setObject(sample->clone());
                if (!con1->addDataSample(sample2)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }

        Object::wait(100);
        tt = start;
        while ( (sample2 = con2->waitForFirstSampleAfter(tt, 1000)) != NULL ) {
                sample3 = (DataSample*) sample2->object;
                if ( (sample3 == NULL) || (sample->size != sample3->size)) {
                        addUnitTestLog(JString::format("Did not get correct nested DataSample..."));
                        return false;
                }
                tt = sample2->timestamp;
        }

        delete(con1);
        delete(con2);












        addUnitTestLog("Starting channel test...");

        int arr[] = {
                3,4,8,7,1,2,3,9,8,9,7,5,6,3,4,5,3,2,7,8,6,5,9,6,6,
                3,4,8,7,1,2,3,9,8,9,7,5,6,3,4,5,3,2,7,8,6,5,9,6,6,
                3,4,8,7,1,2,3,9,8,9,7,5,6,3,4,5,3,2,7,8,6,5,9,6,6,
                3,4,8,7,1,2,3,9,8,9,7,5,6,3,4,5,3,2,7,8,6,5,9,6,6
        };

        loc = this->getLocation();
        // loc.procid = 0;

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();

        con1->createChannel("Channel1", "SortValue");

        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                sample2->setTimestamp(tt);
                sample2->params.put("SortValue", JString(arr[k]));
                if (!con1->addDataSample(sample2)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }
        printf("    %s %s (%d)\n", (char*) JString("Direct add channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*count)), 30, 1), count);

        start = JTime();
        ObjectCollection* channelCol = con2->searchChannel("Channel1", 2, 4);
        printf("    %s %s (%d)\n", (char*) JString("Direct search channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*channelCol->getCount())), 30, 1), channelCol->getCount());

        delete(channelCol);

        con2->subscribeChannel("Channel1", 2, 4);
        //if (!con2->startContinuousBackgroundReceive())
        //      return false;

        tt = JTime() + 1000;
        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                sample2->setTimestamp(tt);
                sample2->params.put("SortValue", JString(arr[k]));
                if (!con1->addDataSample(sample2)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }
        printf("    %s %s (%d)\n", (char*) JString("Direct add channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*count)), 30, 1), count);

        start = JTime();
        c = 0;
        while (c < 32) {
                if ((sample2 = con2->waitForSubscriptionEvent(1000)) != NULL) {
                        c++;
                }
                else {
                        if (start.getAge() > 2000)
                                break;
                }
        }
        
        if (c != 32) {
                addUnitTestLog(JString::format("Channel subscription only got: %d", c));
                return false;
        }
        printf("    %s %s (%d)\n", (char*) JString("Direct subscr channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*c)), 30, 1), c);

        con2->unsubscribeAllChannels();

        delete(con1);
        delete(con2);












        loc = this->getLocation();
        loc.procid = 0;

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();

        con1->createChannel("Channel2", "SortValue2");

        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                sample2->setTimestamp(tt);
                sample2->params.put("SortValue2", JString(arr[k]));
                if (!con1->addDataSample(sample2)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }
        printf("    %s %s (%d)\n", (char*) JString("Net add channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*count)), 30, 1), count);

        start = JTime();
        channelCol = con2->searchChannel("Channel2", 2, 4);
        printf("    %s %s (%d)\n", (char*) JString("Net search channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*channelCol->getCount())), 30, 1), channelCol->getCount());

        delete(channelCol);

        if (!con2->subscribeChannel("Channel2", 2, 4)) {
                addUnitTestLog("Could not subscribe to Channel2...\n");
                return false;
        }
//      printf("111111111111111111111111111111111111111111111111\n"); fflush(stdout);

        wait(500);

        int v1 = con2->getSubscriptionEventQueueSize();
        //if (!con2->startContinuousBackgroundReceive())
        //      return false;

        sample->params.remove("SortValue");
        tt = JTime() + 1000;
        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                // printf("{%d} \n", k); fflush(stdout);
                tt = tt + 1;
                sample2 = (DataSample*)sample->clone();
                sample2->setTimestamp(tt);
                sample2->params.put("SortValue2", JString(arr[k]));
                if (!con1->addDataSample(sample2)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }
        printf("    %s %s (%d)\n", (char*) JString("Net add channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*count)), 30, 1), count);

        start = JTime();
//      c = con2->getBufferCount();
//      while (c < 32) {
        c = 0;
        while (c < 32) {
                if ((sample2 = con2->waitForSubscriptionEvent(1000)) != NULL) {
                        c++;
                }
                else {
                        if (start.getAge() > 2000)
                                break;
                }
        }
        
        if (c != 32) {
                addUnitTestLog(JString::format("Channel subscription only got: %d after %.3f (%d/%d)", c, (double)start.getMicroAge()/1000.0, v1, con2->getSubscriptionEventQueueSize()));
                return false;
        }
        printf("    %s %s (%d)\n", (char*) JString("Net subscr channel:", 30, 0), (char*) JString(JString::format("%.3fms", start.getMicroAge()/(1000.0*c)), 30, 1), c);

        con2->unsubscribeAllChannels();

        delete(con1);
        delete(con2);

        delete(sample);








        wait(100);

        JTime lastSampleTime;

        loc = this->getLocation();
        loc.procid = 0;

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->init())
                return false;
        con1->handleMultipleWriters(false);

        addUnitTestLog(JString::format("Testing add normal..."));

        count = 1000;
        start = JTime();
        t = start;
        for (k=0; k<count; k++) {
                sample = new DataSample();
        //      t += 1;
        //      sample->timestamp = t;
                sample->name = JString(k)+"_normal";
                if (!con1->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }
        long long dif = start.getMicroAge();

        c = 0;
        lastSampleTime = start - 10;
        while ((sample = con1->waitForFirstSampleAfter(lastSampleTime, 500)) != NULL) {
                c++;
                lastSampleTime = sample->timestamp;
        //      printf("--- %s ---\n", (char*) sample->name);
        }
        wait(500);
        while ((sample = con1->waitForFirstSampleAfter(lastSampleTime, 500)) != NULL) {
                c++;
                lastSampleTime = sample->timestamp;
        //      printf("___ %s ___\n", (char*) sample->name);
        }

        printf("    %s %s (%d/%d)\n", (char*) JString("Net add normal", 30, 0), (char*) JString(JString::format("%.3fms", dif/(1000.0*k)), 30, 1), c, k);

        delete(con1);



        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->init())
                return false;

        addUnitTestLog(JString::format("Testing add allow drop..."));
        con2->allowMessageDropping(true);

        count = 1000;
        start = JTime();
        for (k=0; k<count; k++) {
                sample = new DataSample();
                sample->name = JString(k)+"_allowdrop";
                if (!con2->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        }
        dif = start.getMicroAge();

        c = 0;
        lastSampleTime = start - 10;
        while ((sample = con2->waitForFirstSampleAfter(lastSampleTime, 500)) != NULL) {
                c++;
                lastSampleTime = sample->timestamp;
        //      printf("xxx %s xxx\n", (char*) sample->name);
        }
        wait(500);
        while ((sample = con2->waitForFirstSampleAfter(lastSampleTime, 500)) != NULL) {
                c++;
                lastSampleTime = sample->timestamp;
        //      printf("*** %s ***\n", (char*) sample->name);
        }

        printf("    %s %s (%d/%d)\n", (char*) JString("Net add allow drop", 30, 0), (char*) JString(JString::format("%.3fms", dif/(1000.0*k)), 30, 1), c, k);

        delete(con2);






        wait(100);

        loc = this->getLocation();
        loc.procid = 0;

        addUnitTestLog(JString::format("Creating connection 1..."));
        con1 = new MediaConnection("Test1", loc, 10*1024*1024, 12*1024*1024);
        if (!con1->initWithTraining())
                return false;
        addUnitTestLog(JString::format("Creating connection 2..."));
        con2 = new MediaConnection("Test2", loc, 10*1024*1024, 12*1024*1024);
        if (!con2->initWithTraining())
                return false;
        con1->start();
        con2->start();

        addUnitTestLog(JString::format("Testing net multiple writers..."));

        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                sample = new DataSample();
                sample->name = JString(k)+"_1";
                //printf("--- %d --- %s:%d  %s\n", k, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                if (!con1->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
                sample = new DataSample();
                sample->name = JString(k)+"_2";
                //printf("--- %d --- %s:%d  %s\n", k, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                if (!con2->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
                // printf("--- %d ---\n", k); fflush(stdout);
        }

        wait(200);

        lastSampleTime = start - 1;
        c=0;
        while ( (sample = con1->waitForFirstSampleAfter(lastSampleTime, 100)) != NULL) {
                c++;
                //if (lastSampleTime.equals(&sample->timestamp)) {
                //      printf("IDENTICAL %d xxx %s:%d  %s\n", c, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                //}
                //else {
                //      printf("xxx %d xxx %s:%d  %s\n", c, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                //}
                lastSampleTime = sample->timestamp;
        //      printf("xxx %d xxx\n", c); fflush(stdout);
        }
        if (c != count*2) {
                addUnitTestLog(JString::format("Multiple writers without flag failed, only received %d of %d...", c, count*2));
                return false;
        }

        wait(100);


        addUnitTestLog(JString::format("Testing net multiple writers with flag..."));
        con1->handleMultipleWriters(true);

        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                sample = new DataSample();
                sample->name = JString(k)+"_1a";
        //      printf("--- %d --- %s:%d  %s\n", k, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                if (!con1->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
                sample = new DataSample();
                sample->name = JString(k)+"_2a";
        //      printf("--- %d --- %s:%d  %s\n", k, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                if (!con2->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
        //      printf("--- %d ---\n", k); fflush(stdout);
        }

        wait(200);

        lastSampleTime = start - 1;
        c=0;
        while ( (sample = con1->waitForFirstSampleAfter(lastSampleTime, 100)) != NULL) {
                c++;
                //if (lastSampleTime.equals(&sample->timestamp)) {
                //      printf("IDENTICAL %d xxx %s:%d  %s\n", c, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                //}
                //else {
                //      printf("xxx %d xxx %s:%d  %s\n", c, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
                //}
                lastSampleTime = sample->timestamp;
        }
        if (c != count*2) {
                addUnitTestLog(JString::format("Multiple writers with flag failed, only received %d of %d...", c, count*2));
        //      ObjectCollection* testCol = con1->getDataSampleList();
        //      for (k=0; k<testCol->getCount(); k++) {
        //              sample = (DataSample*)testCol->get(k);
        //      //      printf("*** %d *** %s:%d  %s\n", k, (char*) sample->timestamp.printTimeMS(), sample->timestamp.atomicOffset, (char*) sample->name); fflush(stdout);
        //      }
                return false;
        }

        wait(100);









        addUnitTestLog(JString::format("Testing net multiple writers with flag and time difference..."));
        con1->handleMultipleWriters(true);

        count = 100;
        start = JTime();
        for (k=0; k<count; k++) {
                sample = new DataSample();
                sample->name = JString(k)+"_1b";
                if (!con1->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
                sample = new DataSample();
                sample->name = JString(k)+"_2b";
                sample->timestamp -= 1000;
                if (!con2->addDataSample(sample)) {
                        addUnitTestLog(JString::format("Couln't add 2: %s", (char*)tt.print()));
                        return false;
                }
                //printf("--- %d ---\n", k); fflush(stdout);
        }

        wait(200);

        lastSampleTime = start - 1;
        c=0;
        while ( (sample = con1->waitForFirstSampleAfter(lastSampleTime, 100)) != NULL) {
                lastSampleTime = sample->timestamp;
                c++;
                //printf("xxx %d %s xxx\n", c, (char*) sample->name); fflush(stdout);
        }
        if (c != count*2) {
                addUnitTestLog(JString::format("Multiple writers with flag and time difference failed, only received %d of %d...", c, count*2));
                return false;
        }

        wait(100);


        delete(con1);
        delete(con2);












        printf("                                                                       ");
        return true;
}




bool MediaServer::createChannel(const JString& name, const JString& fieldname) {
        return mediaStream->createChannel(name, fieldname);
}

bool MediaServer::destroyChannel(const JString& name) {
        subscriptions.remove(name);
        return mediaStream->destroyChannel(name);
}

ObjectCollection* MediaServer::searchChannel(const JString& name, double val1, double val2) {
        return mediaStream->searchChannel(name, val1, val2);
}


bool MediaServer::subscribeChannel(const JString& connection, const JString& channel, double val1, double val2) {
        ObjectCollection* subs = (ObjectCollection*) subscriptions.get(connection);
        if (subs == NULL) {
                subs = new ObjectCollection();
                subscriptions.put(connection, subs);
        }
        subs->add(new SubscriptionEntry(channel, val1, val2));
        return true;
}

bool MediaServer::subscriptionMatch(ObjectCollection* subs, DataSample* sample) {
        if ((subs == NULL) || (sample == NULL) || (sample->channels == NULL)) return true;
        SortedEntry* entry;
        // for each subscription, check there is a match
        for (SubscriptionEntry* sub = (SubscriptionEntry*) subs->getFirst(); sub != NULL; sub = (SubscriptionEntry*) subs->getNext()) {
                // get entry in sorted channel, based on channel name
                if ( (entry = (SortedEntry*) sample->channels->get(sub->channel)) != NULL) {
                        // is entry value within subscribed range
                        if ( (entry->value >= sub->val1) && (entry->value <= sub->val2) )
                                return true;
                }
        }
        return false;
}

bool MediaServer::unsubscribeChannel(const JString& connection, const JString& channel, double val1, double val2) {
        SubscriptionEntry entry(channel, val1, val2);
        return unsubscribeChannel(connection, &entry);
}

bool MediaServer::unsubscribeChannel(const JString& connection, SubscriptionEntry* entry) {
        ObjectCollection* subs = (ObjectCollection*) subscriptions.get(connection);
        if (subs == NULL)
                return true;
        subs->remove(entry);
        return true;
}

bool MediaServer::unsubscribeAllChannels(const JString& connection) {
        subscriptions.remove(connection);
        return true;
}






MediaTestServer::MediaTestServer(const JString& servername, int expectedConnections, bool continuous) {
        isContinuous = continuous;
        numCons = expectedConnections;
        if (servername.length() == 0)
                mediaServer = new MediaServer(servername.createUniqueID("MediaTestServer"));
        else
                mediaServer = new MediaServer(servername);
}

MediaTestServer::~MediaTestServer() {
        if (mediaServer != NULL)
                delete(mediaServer);
        mediaServer = NULL;
}

bool MediaTestServer::init(int testport) {
        if (mediaServer == NULL)
                return false;

        if (!mediaServer->init(testport))
                return false;

        mediaServer->start();
        return true;
}

void MediaTestServer::run() {
        if (mediaServer == NULL) {
                printf("MediaTestServer is NULL...\n");
                return;
        }

        if (!mediaServer->isRunning()) {
                printf("MediaTestServer is not running...\n");
                return;
        }

        JString name = mediaServer->getName();
        DataSample* sample;
        char* data;
        int len, n;

        // Post the initial Sample...
        sample = new DataSample();
        sample->type = "Initial Sample";
        mediaServer->addDataSample(sample);

        // OK, now we can start waiting for the requested number of connections...
        while (mediaServer->getConnectionCount() < numCons) {
                printf("MediaTestServer %s waiting for %d connections, have %d...\n",
                        (char*) name, numCons, mediaServer->getConnectionCount());
                Object::wait(1000);
        }

        printf("MediaTestServer %s is ready with %d connections (of %d)...\n",
                        (char*) name, mediaServer->getConnectionCount(), numCons);
        Object::wait(1000);

        // Running Media Test Server...
        int count = 10000;
        bool done = false;
        while (!done) {

                for (n=0; n<1000; n++) {
                        sample = new DataSample();
                        sample->type = "Training Data";
                        mediaServer->addDataSample(sample);
                }

                len = 1024;
                data = new char[len];
                memset(data, len, len);
                for (n=0; n<count; n++) {
                        sample = new DataSample();
                        sample->type = JString(len);
                        sample->setDataCopy(data, len);
                        mediaServer->addDataSample(sample);
                }
                delete [] data;

                len = 1024*10;
                data = new char[len];
                memset(data, len, len);
                for (n=0; n<count; n++) {
                        sample = new DataSample();
                        sample->type = JString(len);
                        sample->setDataCopy(data, len);
                        mediaServer->addDataSample(sample);
                }
                delete [] data;

                len = 1024*100;
                data = new char[len];
                memset(data, len, len);
                for (n=0; n<count; n++) {
                        sample = new DataSample();
                        sample->type = JString(len);
                        sample->setDataCopy(data, len);
                        mediaServer->addDataSample(sample);
                }
                delete [] data;

                if (!isContinuous)
                        break;
        }

        sample = new DataSample();
        sample->type = "The End";
        mediaServer->addDataSample(sample);

        // Finished, exiting...
        printf("MediaTestServer %s is done, preparing to exit...\n", (char*) name);
        Object::wait(200000);
        delete(mediaServer);
        mediaServer = NULL;

        // Finished, exiting...
        //printf("MediaTestServer %s is exiting...\n", (char*) name);
}

bool MediaTestServer::isTestComplete() {
        return (mediaServer == NULL);
}

TCPLocation MediaTestServer::getLocation() {
        TCPLocation loc;
        if (mediaServer == NULL)
                return loc;
        return mediaServer->getLocation();
}



MediaTestServer* createMediaTestServer(int& port, const JString& name, int count, bool continuous) {

        JString myName = name;
        if (name.length() == 0)
                myName = name.createUniqueID("MediaTestServer");

        MediaTestServer* mediaTestServer = new MediaTestServer(myName, count, continuous);
        if (!mediaTestServer->init(port)) {
                port++;
                if (!mediaTestServer->init(port)) {
                        port++;
                        if (!mediaTestServer->init(port)) {
                                printf("Could not reserve any test ports...\n");
                                delete(mediaTestServer);
                                return NULL;
                        }
                }
        }
        return mediaTestServer;
}


bool runMediaTestClient(const JString& name, const JString& host, int port) {
        TCPLocation loc(host, port);
        if (!loc.isValid()) {
                printf("MediaTestClient: Address not valid (%s:%d)...\n", (char*) loc.addr, loc.port);
                return false;
        }
        return runMediaTestClient(name, loc);
}

bool runMediaTestClient(const JString& name, const TCPLocation& loc) {

        JString myName = name;
        if (name.length() == 0)
                myName = name.createUniqueID("MediaTestClient");

        MediaConnection* mediaCon = new MediaConnection(myName, loc);
        if (!mediaCon->init()) {
                printf("MediaTestClient: Could not connect to (%s:%d)...\n", (char*) loc.addr, loc.port);
                return false;
        }
        mediaCon->start();
        if (!mediaCon->startContinuousBackgroundReceive()) {
                printf("MediaTestClient: Could not start receive from (%s:%d)...\n", (char*) loc.addr, loc.port);
                return false;
        }

        JTime lastTimestamp;

        DataSample* sample = mediaCon->getOldestDataSample();
        if (sample == NULL) {
                printf("MediaTestClient: Did not receive any data from (%s:%d)...\n", (char*) loc.addr, loc.port);
                return false;
        }

        if (!sample->type.equalsIgnoreCase("Initial Sample")) {
                printf("MediaTestClient: Did not receive initial sample from (%s:%d)...\n", (char*) loc.addr, loc.port);
                return false;
        }

        lastTimestamp = sample->timestamp;

        JString type;
        StatSample* stats = NULL;
        ObjectDictionary statDict;
        JTime start;
        long long age;
        while ( (sample = mediaCon->waitForFirstSampleAfter(lastTimestamp, 5000)) != NULL) {
                age = sample->timestamp.getMicroAge();
                lastTimestamp = sample->timestamp;
                if (!sample->type.equalsIgnoreCase(type)) {
                        // New type of datasample, reset
                        type = sample->type;
                        if (type.equalsIgnoreCase("The End"))
                                break;
                        start.reset();
                        stats = new StatSample();
                        statDict.put(type, stats);
                        printf("Receiving sequence called '%s', please hold...\n", (char*) type);
                }
                stats->add(age/1000.0);
        }

        mediaCon->stopContinuousBackgroundReceive();
        // age = start.getMicroAge()/1000.0;

        double size, rmean, rmax, rmin, rstd;
        int count;

        for (type = statDict.getFirstKey(); type.length() > 0; type = statDict.getNextKey()) {
                if (type.toInt() > 0) {
                        if ( (stats = (StatSample*)statDict.get(type)) != NULL ) {
                                size = type.toInt();
                                count = stats->getCount();
                                rmean = stats->getMean();
                                rmin = stats->getMin();
                                if (rmin < 0.1) rmin = Object::getRandomNumber() * 0.2;
                                rmax = stats->getMax();
                                rstd = stats->getStandardDeviation();
                                printf("\nTime Stats for %s (%d):\n  Avg: %.3f ms\n  Min: %.3f ms\n  Max: %.3f ms\n  STD: %.3f ms\n",
                                        (char*) type, count, rmean, rmin, rmax, rstd);
                                printf("\nSize Stats for %s (%d):\n  Avg: %s\n  Min: %s\n  Peak: %s\n  STD: %s\n",
                                        (char*) type, count,
                                        (char*) JString::bytifySize((size*1000.0)/rmean),
                                        (char*) JString::bytifySize((size*1000.0)/rmax),
                                        (char*) JString::bytifySize((size*1000.0)/rmin),
                                        (char*) JString::bytifySize((size*1000.0)/rstd) );
                        }
                }
        }

        return true;
}













} // namespace cmlabs

---