source: source/ariba/communication/modules/transport/protlib/fqueue.cpp@ 4987

/// ----------------------------------------*- mode: C++; -*--

/// @file fqueue.cpp

/// wrapper class for fastqueue

/// ----------------------------------------------------------

/// $Id: fqueue.cpp 2549 2007-04-02 22:17:37Z bless $

/// $HeadURL: https://svn.ipv6.tm.uka.de/nsis/protlib/trunk/src/fqueue.cpp $

// ===========================================================

//                      

// Copyright (C) 2005-2007, all rights reserved by

// - Institute of Telematics, Universitaet Karlsruhe (TH)

//

// More information and contact:

// https://projekte.tm.uka.de/trac/NSIS

//                      

// This program is free software; you can redistribute it and/or modify

// it under the terms of the GNU General Public License as published by

// the Free Software Foundation; version 2 of the License

//

// This program 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 General Public License for more details.

//

// You should have received a copy of the GNU General Public License along

// with this program; if not, write to the Free Software Foundation, Inc.,

// 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

//

// ===========================================================



#include "fqueue.h"

#include "stdio.h"

#include "logfile.h"



#include <string>



namespace protlib {



/** @addtogroup fastqueue Fast Queue

 * @{

 */



  using namespace log;



/**

 * Constructor.

 *

 * Initialize a FastQueue with a queue name and enable/disable expedited

 * data.

 *

 * @param qname the queue's name, or NULL

 * @param exp if true, expedited data support is enabled

 */

FastQueue::FastQueue(const char *qname, bool exp)

    : queue_name((qname == 0) ? "" : (char*)qname), shutdownflag(false)

{

  if ((queue = create_queue(qname)) == NULL)

  {

    Log(ERROR_LOG, LOG_ALERT, "FastQueue", "Could not create queue " << queue_name);

    throw FQError();

  } else queue_enable_expedited(queue,exp);

}





/**

 * Add a message to the queue.

 *

 * If exp is true and the queue allows expedited data, the message will

 * pass all normal messages in the queue and thus will be delivered earlier.

 * If there are other expedited messages in the queue already, it will be

 * added after the already existing expedited messages.

 *

 * This method may fail (and return false) if the queue is in shutdown mode,

 * there is a problem aquiring locks, or some other threading problem.

 *

 * In case the queue is deleted before this message has been removed, this

 * message is deleted using the delete operator. Because of this, the same

 * message may only appear once in a queue.

 * 

 * @param element a pointer to the message to add

 * @param exp true if this is expedited data

 * 

 * @return true if the element was enqueued successfully

 */

bool FastQueue::enqueue(message *element, bool exp)

{

  if (shutdownflag) return false;

  if (enqueue_element_expedited_signal(queue, (void*)element, exp) < 0)

  {

    Log(ERROR_LOG, LOG_ALERT, "FastQueue", "Could not enqueue element in queue " << queue_name);

    return false;

  }

  return true;

}





/**

 * Wait for a message for a given time.

 *

 * If no message arrives in the given time period, NULL is returned.

 *

 * @param msec the time to wait in milliseconds

 *

 * @return the message, or NULL

 */

message *FastQueue::dequeue_timedwait(const long int msec)

{

  struct timespec tspec = {0,0};

  tspec.tv_sec = msec/1000;

  tspec.tv_nsec = (msec%1000)*1000000;

  return (message*)dequeue_element_timedwait(queue, &tspec);

}





/**

 * Destructor.

 * 

 * Destroys the queue. All messages which are still in the queue are deleted

 * using the delete operator.

 */

FastQueue::~FastQueue()

{

  if (queue) 

  {

    cleanup();

    if ((destroy_queue(queue)) < 0)

    {

      Log(ERROR_LOG, LOG_ALERT, "FastQueue", "Could not destroy queue " << queue_name);

    }

  }

  DLog("FastQueue", "~FastQueue() - done for queue " << queue_name);

}



/**

 * Test if the queue is empty.

 *

 * @return true if the queue is empty

 */

bool FastQueue::is_empty() const

{

  if (queue_nr_of_elements(queue)==0)

    return true;

  else

    return false;

}





/**

 * Return the number of messages in the queue.

 *

 * @return the number of enqueued messages

 */

unsigned long FastQueue::size() const

{

  return queue_nr_of_elements(queue);

}





/**

 * Test if expedited message support is enabled.

 * 

 * @return true if expedited message support is enabled

 */

bool FastQueue::is_expedited_enabled() const

{

  if (queue_is_expedited_enabled(queue))

    return true;

  else

    return false;

}



/**

 * Enable or disable expedited messages.

 *

 * This also returns the previous value of this flag.

 *

 * @return true, if expedited messages were previously enabled, false otherwise

 */

bool FastQueue::enable_expedited(bool exp)

{

  if (queue_enable_expedited(queue,exp))

    return true;

  else

    return false;

}





/**

 * Disable enqueueing of new messages.

 *

 * A queue in shutdown mode does not accept messages any more.

 */

void FastQueue::shutdown() { shutdownflag = true; }





/**

 * Put queue into shutdown mode and delete all stored messages..

 *

 * @return the number of messages that were in the queue

 */

unsigned long FastQueue::cleanup()

{

  unsigned long count = 0;

  message* m = NULL;

  shutdownflag = true;

  while (!is_empty())

    if ((m = dequeue(false))) {

      delete m;

      m = NULL;

      count++;

    }

  return count;

}



//@}



} // end namespace protlib