| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| #include "BlockingByteArrayInputStream.h" |
| #include <algorithm> |
| |
| using namespace std; |
| using namespace decaf; |
| using namespace decaf::io; |
| using namespace decaf::lang; |
| using namespace decaf::lang::exceptions; |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| BlockingByteArrayInputStream::BlockingByteArrayInputStream() : |
| InputStream(), buffer(), pos(buffer.end()), closing(false) { |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| BlockingByteArrayInputStream::BlockingByteArrayInputStream(const unsigned char* buffer, int bufferSize) : |
| InputStream(), buffer(), pos(), closing(false) { |
| |
| setByteArray(buffer, bufferSize); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| BlockingByteArrayInputStream::~BlockingByteArrayInputStream() { |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void BlockingByteArrayInputStream::setByteArray(const unsigned char* lbuffer, int lbufferSize) { |
| |
| synchronized(this) { |
| |
| // Remove old data |
| this->buffer.clear(); |
| this->buffer.reserve(lbufferSize); |
| |
| // Copy data to internal buffer. |
| this->buffer.insert(this->buffer.begin(), lbuffer, lbuffer + lbufferSize); |
| |
| // Begin at the Beginning. |
| this->pos = this->buffer.begin(); |
| |
| // Notify any listening threads that there is now data available. |
| notifyAll(); |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| int BlockingByteArrayInputStream::available() const { |
| return (int) std::distance(pos, buffer.end()); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void BlockingByteArrayInputStream::close() { |
| |
| synchronized(this) { |
| |
| // Indicate that we're shutting down. |
| closing = true; |
| |
| // Clear out the buffer. |
| buffer.clear(); |
| |
| // Notify that this stream is shutting down. |
| notifyAll(); |
| } |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| int BlockingByteArrayInputStream::doReadByte() { |
| |
| try { |
| |
| synchronized(this) { |
| |
| while (!closing) { |
| |
| if (pos != buffer.end()) { |
| return *(pos++); |
| } |
| |
| // Wait for more data |
| wait(); |
| } |
| |
| throw IOException(__FILE__, __LINE__, "close occurred during read"); |
| } |
| |
| return 0; |
| } |
| DECAF_CATCH_RETHROW(IOException) |
| DECAF_CATCHALL_THROW(IOException) |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| int BlockingByteArrayInputStream::doReadArrayBounded(unsigned char* buffer, int size, int offset, int length) { |
| |
| if (length == 0) { |
| return 0; |
| } |
| |
| if (buffer == NULL) { |
| throw NullPointerException(__FILE__, __LINE__, "BlockingByteArrayInputStream::read - Passed buffer is Null"); |
| } |
| |
| if (size < 0) { |
| throw IndexOutOfBoundsException(__FILE__, __LINE__, "size parameter out of Bounds: %d.", size); |
| } |
| |
| if (offset > size || offset < 0) { |
| throw IndexOutOfBoundsException(__FILE__, __LINE__, "offset parameter out of Bounds: %d.", offset); |
| } |
| |
| if (length < 0 || length > size - offset) { |
| throw IndexOutOfBoundsException(__FILE__, __LINE__, "length parameter out of Bounds: %d.", length); |
| } |
| |
| try { |
| |
| synchronized(this) { |
| |
| int ix = 0; |
| |
| for (; ix < length && !closing; ++ix) { |
| |
| if (pos == this->buffer.end()) { |
| // Wait for more data to come in. |
| wait(); |
| } |
| |
| if (!closing && pos != this->buffer.end()) { |
| buffer[ix + offset] = *(pos); |
| ++pos; |
| } |
| } |
| |
| if (closing) { |
| throw IOException(__FILE__, __LINE__, "BlockingByteArrayInputStream::read - close occurred during read"); |
| } |
| |
| return (int) ix; |
| } |
| |
| return 0; |
| } |
| DECAF_CATCH_RETHROW(IOException) |
| DECAF_CATCH_RETHROW(NullPointerException) |
| DECAF_CATCH_RETHROW(IndexOutOfBoundsException) |
| DECAF_CATCHALL_THROW(IOException) |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| long long BlockingByteArrayInputStream::skip(long long num) { |
| |
| long long ix = 0; |
| |
| synchronized(this) { |
| // Increment the pos until we'v skipped the desired num |
| // or we've hit the end of the buffer. |
| for (; ix < num && !closing && pos != buffer.end(); ++ix, ++pos) {} |
| } |
| |
| return ix; |
| } |