// Copyright (c) 2013 by István Váradi // This file is part of VSCPL, a simple cross-platform utility library // Redistribution and use in source and binary forms, with or without // modification, are permitted provided that the following conditions are met: // 1. Redistributions of source code must retain the above copyright notice, this // list of conditions and the following disclaimer. // 2. Redistributions in binary form must reproduce the above copyright notice, // this list of conditions and the following disclaimer in the documentation // and/or other materials provided with the distribution. // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND // ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED // WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE // DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR // ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES // (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; // LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND // ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. // The views and conclusions contained in the software and documentation are those // of the authors and should not be interpreted as representing official policies, // either expressed or implied, of the FreeBSD Project. //------------------------------------------------------------------------------ #include "BlockingStream.h" #include "ReadingBuffer.h" #include "WritingBuffer.h" #include "Waiter.h" #include #include //------------------------------------------------------------------------------ using hu::varadiistvan::scpl::io::BlockingStream; using std::min; //------------------------------------------------------------------------------ inline bool BlockingStream::checkInterrupted() { if (!interrupted) { interrupted = event.check(); if (!interrupted && event.failed()) { setErrorCode(event.getErrorCode()); } } return interrupted; } //------------------------------------------------------------------------------ bool BlockingStream::read(void* dest, size_t length) { unsigned char* d = reinterpret_cast(dest); ReadingBuffer& readingBuffer = stream.getReadingBuffer(); while (length>0) { size_t copied = readingBuffer.extract(d, length, readingOffset); readingOffset += copied; length -= copied; d += copied; if (length!=0) { if (!fill()) return false; } } return true; } //------------------------------------------------------------------------------ bool BlockingStream::skip(size_t length) { ReadingBuffer& readingBuffer = stream.getReadingBuffer(); while (length>0) { size_t toSkip = min(length, readingBuffer.getLength() - readingOffset); readingOffset += toSkip; length -= toSkip; if (length!=0) { if (!fill()) return false; } } return true; } //------------------------------------------------------------------------------ bool BlockingStream::write(const void* src, size_t length) { const unsigned char* s = reinterpret_cast(src); WritingBuffer& writingBuffer = stream.getWritingBuffer(); while(length>0) { size_t copied = writingBuffer.append(s, length); length -= copied; s += copied; if (length==0) break; if (!flush()) return false; } return true; } //------------------------------------------------------------------------------ bool BlockingStream::flush() { WritingBuffer& writingBuffer = stream.getWritingBuffer(); while (*this) { if (checkInterrupted()) return false; if (writingBuffer.write()) { return true; } else if (writingBuffer.failed()) { setErrorCode(writingBuffer.getErrorCode()); } else { Waiter* waiter = stream.getWaiter(); waiter->wait(); if (waiter->failed()) { setErrorCode(waiter->getErrorCode()); } } } return false; } //------------------------------------------------------------------------------ bool BlockingStream::fill() { ReadingBuffer& readingBuffer = stream.getReadingBuffer(); readingBuffer.reset(); readingOffset = 0; while (*this) { if (checkInterrupted()) return false; if (readingBuffer.read()) { eof = readingBuffer.isEmpty(); return !eof; } else if (readingBuffer.failed()) { setErrorCode(readingBuffer.getErrorCode()); } else { Waiter* waiter = stream.getWaiter(); waiter->wait(); if (waiter->failed()) { setErrorCode(waiter->getErrorCode()); return false; } } } return false; } //------------------------------------------------------------------------------ // Local Variables: // mode: C++ // c-basic-offset: 4 // indent-tabs-mode: nil // End: