Logo Search packages:      
Sourcecode: earth3d version File versions  Download package

connectNetworkServiceRequestQueue.cpp

#include "connectNetworkServiceRequestQueue.h"
#include <qdom.h>
#include "mutex.h"
#include "connectNetworkService.h"
#include <iostream>

using namespace std;

/** The message ID for new data requests. Every request gets an unique number
 *  and is saved into mapHandler.
 */
long currentMessageID = 0;

ConnectNetworkServiceRequestQueue::ConnectNetworkServiceRequestQueue(int maxRunningRequests) {
  this->maxRunningRequests = maxRunningRequests;
  //  this->currentMessageID=0;
  this->runningRequests = 0;
}

ConnectNetworkServiceRequestQueue::~ConnectNetworkServiceRequestQueue() {
}

void ConnectNetworkServiceRequestQueue::setMaxRunningRequests(int maxRunningRequest) {
  this->maxRunningRequests = maxRunningRequests;
}

void ConnectNetworkServiceRequestQueue::push(ConnectNetworkServiceRequestQueueObject *obj) {
  queueMutex.lock();
  queue.push_front(obj);
//   queue.push_back(obj);
  queueMutex.unlock();

  obj->getDataReceivedListener()->addDownloader(this);

  if (runningRequests<maxRunningRequests) dequeue();
}

void ConnectNetworkServiceRequestQueue::dequeue() {
  bool result= false;

  {
    queueMutex.lock();
    
    /* Get the first object and then remove it from the request queue */
    if (queue.size()<=0) {
      queueMutex.unlock();
      return;
    }
    
    ConnectNetworkServiceRequestQueueObject *request = queue.front();
    queue.pop_front();

    runningRequests++;
    
    /* add DataReceiverListener to list */
    QTime timeoutval = QTime::currentTime();
    timeoutval = timeoutval.addSecs(request->getTimeout());
    
    mapHandlerMutex.lock();
    mapHandler[currentMessageID]=new TimedDataReceivedListener(this, timeoutval, request);
    request->setMessageID(currentMessageID);
    mapHandlerMutex.unlock();
    
    /* increase message id */
    currentMessageID++;

    queueMutex.unlock();

    /* request the data */
    result = requestData(request);
    
  }

  /* see if we can dequeue another one */
  if (result) {
    runningRequests--;
    dequeue();
  }
}

void ConnectNetworkServiceRequestQueue::dataReceived(const char *response, int size, const char *sender, int countParts, struct part_t *parts, void *userdata, ConnectNetworkService *cns, StopableDownload *download) {
  /* a request has finished. Call the original listener and dequeue another task */
  long messageID = (long) userdata;
  mapHandlerMutex.lock();
  TimedDataReceivedListener *listener = NULL;
  if (mapHandler.count(messageID)>0) {
    listener = mapHandler[messageID];
  }
  mapHandlerMutex.unlock();

  if (listener) {
    QMutexLocker mlock2(&treelock);

    /* get original listener */
    ConnectNetworkServiceRequestQueueObject *request = (ConnectNetworkServiceRequestQueueObject *) listener->userdata;
      
    if (!listener->downloadStopped) {
      /* call it with the received data */
      request->getDataReceivedListener()->dataReceived(response, size, sender, countParts, parts, request->getUserData(), cns, download);
      request->getDataReceivedListener()->removeDownloader(this);
    }
      
    /* remove from mapHandler */
    mapHandlerMutex.lock();
    delete(request);
    delete(listener);
    mapHandler.erase(messageID);
    mapHandlerMutex.unlock();
  }

  /* dequeue another request */
  runningRequests--;
  dequeue();
}

TimedDataReceivedListener *ConnectNetworkServiceRequestQueue::getMapHandler(long id) {
  QMutexLocker qml(&mapHandlerMutex);
  return(mapHandler[id]);
}

void ConnectNetworkServiceRequestQueue::eraseMapHandler(long id) {
  QMutexLocker qml(&mapHandlerMutex);
  mapHandler.erase(id);
}

void ConnectNetworkServiceRequestQueue::stopDownload(Downloadable *downloadable) {
  /* remove downloadable from queue and stop download if it is already requested */
  QMutexLocker qml(&queueMutex);

  /* remove from queue */
  std::list<ConnectNetworkServiceRequestQueueObject *>::iterator iq;
  for(iq=queue.begin(); iq!=queue.end(); iq++) {
    if ((*iq)->getDataReceivedListener()==downloadable) {
      queue.erase(iq);
      break;
    }
  }

  /* add stopped information to mapHandler */
  std::map<long, TimedDataReceivedListener *>::iterator it;
  for(it=mapHandler.begin(); it!=mapHandler.end(); it++) {
    TimedDataReceivedListener *listener = (*it).second;

    /* get original listener */
    ConnectNetworkServiceRequestQueueObject *request = (ConnectNetworkServiceRequestQueueObject *) listener->userdata;

    if (request->getDataReceivedListener()==downloadable) {
      /* stop download */
      listener->downloadStopped = true;
    }
  }
}


Generated by  Doxygen 1.6.0   Back to index