// TJP COPYRIGHT HEADER #include "Pipe.h" #include namespace tjp { namespace core { template Pipe::Pipe () : offset(0), size_(0), open(true) { } template Pipe::~Pipe () { } template void Pipe::close() { auto l = lock_of(mutex); open = false; } template bool Pipe::isOpen() const { auto l = lock_of(mutex); return open; } template size_t Pipe::write_no_lock (Packet &&packet) { debug_assert(mutex.locked_by_caller()); if (packet.size() == 0) return size_; size_ += packet.size(); pipe.emplace_back(std::move(packet)); return size_; } template size_t Pipe::write (Packet &&packet) { auto l = lock_of(mutex); return write_no_lock(std::move(packet)); } template Packet Pipe::read_no_lock () { debug_assert(!pipe.empty()); debug_assert(offset == 0); debug_assert(mutex.locked_by_caller()); auto p = std::move(pipe.front()); pipe.pop_front(); debug_assert(size_ >= p.size()); size_ -= p.size(); offset = 0; return p; } template Packet Pipe::read () { auto l = lock_of(mutex); return read_no_lock(); } template bool Pipe::empty_no_lock() const { return pipe.empty(); } template bool Pipe::empty() const { auto l = lock_of(mutex); return empty_no_lock(); } // --------------------- template size_t Pipe::peek_no_lock (char *buffer, size_t size) const { debug_assert(mutex.locked_by_caller()); size_t ptr = 0; size_t offset = this->offset; auto &buffers = pipe; typename PacketList::const_iterator i=buffers.begin(); while (i!=buffers.end() && (size > 0)) { auto chunksize = i->size() - offset; size_t newoffset = 0; if (chunksize > size) { newoffset = offset + size; chunksize = size; } if (buffer) memcpy (buffer + ptr, i->data() + offset, chunksize); ptr += chunksize; size -= chunksize; offset = newoffset; i++; } return ptr; } template size_t Pipe::peek (char *buffer, size_t size) const { auto l = lock_of(mutex); return peek_no_lock(buffer, size); } template size_t Pipe::read_no_lock (char *buffer, size_t requestedSize) { auto size = requestedSize; size_t ptr = 0; auto &buffers = pipe; while (!buffers.empty() && (size > 0)) { auto chunksize = buffers.front().size() - offset; size_t newoffset = 0; if (chunksize > size) { newoffset = offset + size; chunksize = size; } if (buffer) { debug_assert(ptr + chunksize <= requestedSize); memcpy (buffer + ptr, buffers.front().data() + offset, chunksize); } ptr += chunksize; size -= chunksize; offset = newoffset; if (offset == 0) { size_ -= buffers.front().size(); buffers.pop_front(); } } sizeReadRequested += requestedSize; sizeRead += ptr; return ptr; } ; template void Pipe::clearSizeRead () { sizeReadRequested = sizeRead = 0; } template size_t Pipe::read (char *buffer, size_t requestedSize) { auto l = lock_of(mutex); return read_no_lock(buffer, requestedSize); } template size_t Pipe::write (const char *buffer, size_t size) { if (size == 0) return size; write(Packet(buffer, buffer+size)); return size; } template size_t Pipe::skip_no_lock(size_t size) { return read_no_lock(nullptr, size); } template size_t Pipe::skip(size_t size) { auto l = lock_of(mutex); return skip_no_lock(size); } template size_t Pipe::size () const { auto l = lock_of(mutex); return size_ - offset; } template void Pipe::clear () { auto l = lock_of(mutex); pipe.clear(); size_ = offset = 0; flush_ = false; } template size_t Pipe::resize (size_t maximumSize) { auto l = lock_of(mutex); if (size_ > maximumSize) { skip_no_lock(size_ - maximumSize); } return size_; } template const Packet *Pipe::peek() const { auto l = lock_of(mutex); if (pipe.empty()) return nullptr; return &pipe.front(); } template bool Pipe::eos() const { return !open && empty(); } template bool Pipe::finishing() const { return !open && !empty(); } template bool Pipe::flushing() const { return flush_ && !empty(); } template bool Pipe::flushed() { if (!flush_) return false; if (!empty()) return false; flush_ = false; return true; } } // namespace } // namespace