blob: 2ec092c0ff2f6483398e4b8820a60fa56fcf0449 [file] [log] [blame]
/*
* 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;
}