/*
 * Copyright (c) 2004-2007 Marcus Overhagen <marcus@overhagen.de>
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without restriction, 
 * including without limitation the rights to use, copy, modify, 
 * merge, publish, distribute, sublicense, and/or sell copies of 
 * the Software, and to permit persons to whom the Software is 
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be 
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT 
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR 
 * OTHER DEALINGS IN THE SOFTWARE.
 */

#include <stdio.h>
#include <OS.h>

#include "Packet.h"
#include "PacketQueue.h"

#define TRACE_PACKET_QUEUE
#ifdef TRACE_PACKET_QUEUE
  #define TRACE printf
#else
  #define TRACE(a...)
#endif


PacketQueue::PacketQueue(int max_packets)
 :      fQueue(new Packet* [max_packets])
 ,      fSem(create_sem(0, "packet queue sem"))
 ,      fTerminate(false)
 ,      fWriteIndex(0)
 ,      fReadIndex(0)
 ,      fMaxPackets(max_packets)
 ,      fPacketCount(0)
{
}


PacketQueue::~PacketQueue()
{
        Empty();
        delete_sem(fSem);
        delete [] fQueue;
}


void
PacketQueue::Empty()
{
        while (fPacketCount--) {
                delete fQueue[fReadIndex];
                fReadIndex = (fReadIndex + 1) % fMaxPackets;
        }
}
                        

status_t
PacketQueue::Insert(Packet *packet)
{
        if (fTerminate) {
                return B_NOT_ALLOWED;
        }
        if (atomic_add(&fPacketCount, 1) == fMaxPackets) {
                atomic_add(&fPacketCount, -1);
                return B_WOULD_BLOCK;
        }
        fQueue[fWriteIndex] = packet;
        fWriteIndex = (fWriteIndex + 1) % fMaxPackets;
        release_sem(fSem);
        return B_OK;
}


status_t
PacketQueue::Remove(Packet **packet)
{
        if (fTerminate) {
                return B_NOT_ALLOWED;
        }
        if (acquire_sem(fSem) != B_OK)
                return B_ERROR;
        if (fTerminate) {
                return B_NOT_ALLOWED;
        }
        *packet = fQueue[fReadIndex];
        atomic_add(&fPacketCount, -1);
        fReadIndex = (fReadIndex + 1) % fMaxPackets;
        return B_OK;
}


void
PacketQueue::Flush(bigtime_t timeout)
{
        if (fTerminate) {
                return;
        }

        timeout += system_time();
        
        while (acquire_sem_etc(fSem, 1, B_ABSOLUTE_TIMEOUT, timeout) == B_OK) {
                if (fTerminate) {
                        return;
                }
                Packet *packet = fQueue[fReadIndex];
                atomic_add(&fPacketCount, -1);
                fReadIndex = (fReadIndex + 1) % fMaxPackets;
                delete packet;
        }
}


// peeks into queue and delivers a copy
status_t
PacketQueue::Peek(Packet **packet)
{
        if (fTerminate) {
                return B_NOT_ALLOWED;
        }
        if (acquire_sem(fSem) != B_OK)
                return B_ERROR;
        if (fTerminate) {
                return B_NOT_ALLOWED;
        }
        *packet = new Packet(*fQueue[fReadIndex]);
        release_sem(fSem);
        return B_OK;
}


void
PacketQueue::Terminate()
{
        fTerminate = true;
        release_sem(fSem);
}


void
PacketQueue::Restart()
{
        Empty();

        delete_sem(fSem);
        fSem = create_sem(0, "packet queue sem");
        fTerminate = false;
        fWriteIndex = 0;
        fReadIndex = 0;
        fPacketCount = 0;
}