Java/File Input Output/FilterInputStream
A FilterInputStream with a limited bandwith
<source lang="java">
import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; /*********************************************
Copyright (c) 2001 by Daniel Matuschek
- /
/**
* A FilterInputStream with a limited bandwith * * This implements an filter for an existing input stream that allows * it to limit the read bandwidth. This can be useful for network * streams that should be limited to a specified bandwidth. * * @author * @version $id$ */
public class LimitedBandwidthStream
extends FilterInputStream {
/** usable bandwidth in bytes/second **/
private int bandwidth = 0;
/** bandwidth limit will be calculated form the start time **/
private boolean isReading = false;
/** number of bytes read **/
private int count = 0;
/** check bandwidth every n bytes **/
private static int CHECK_INTERVAL = 100;
/** start time **/
long starttime = 0;
/** used time **/
long usedtime = 0;
/**
* initializes the LimitedBandWidth stream
*/
public LimitedBandwidthStream (InputStream in, int bandwidth)
throws IOException
{
super(in);
if (bandwidth > 0) {
this.bandwidth=bandwidth;
} else {
this.bandwidth=0;
}
count = 0;
}
/**
* Reads the next byte.
*
* Reads the next byte of data from this input stream. The value byte
* is returned as an int in the range 0 to 255. If no byte is available
* because the end of the stream has been reached, the value -1 is
* returned. This method blocks until input data is available, the end
* of the stream is detected, or an exception is thrown.
* If the bandwidth consumption exceeds the defined limit, read will block
* until the bandwidth is in the limit again.
*
* @return the next byte from the stream or -1 if end-of-stream
*/
public int read()
throws IOException
{
long currentBandwidth;
if (! isReading) {
starttime = System.currentTimeMillis();
isReading = true;
}
// do bandwidth check only if bandwidth
if ((bandwidth > 0) &&
((count % CHECK_INTERVAL) == 0)) {
do {
usedtime = System.currentTimeMillis()-starttime;
if (usedtime > 0) {
currentBandwidth = (count*1000) / usedtime;
} else {
currentBandwidth = 0;
}
if (currentBandwidth > bandwidth) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {}
}
} while (currentBandwidth > bandwidth);
}
count++;
return super.read();
}
/**
* Shortcut for read(b,0,b.length)
*
* @see #read(byte[], int, int)
*/
public int read(byte[] b) throws IOException {
return read(b, 0, b.length);
}
/**
* Reads a block of bytes from the stream.
*
* If the bandwith is not limited, it simply used the
* read(byte[], int, int) method of the input stream, otherwise it
* uses multiple read() request to enforce bandwith limitation (this
* is easier to implement using byte reads).
*
* @return the number of bytes read or -1 at end of stream
*/
public int read(byte[] b, int off, int len) throws IOException {
int mycount = 0;
int current = 0;
// limit bandwidth ?
if (bandwidth > 0) {
for (int i=off; i < off+len; i++) {
current = read();
if (current == -1) {
return mycount;
} else {
b[i]=(byte)current;
count++;
mycount++;
}
}
return mycount;
} else {
return in.read(b, off, len);
}
}
} // LimitedBandwidthStream
</source>
introduce a protocol for reading arbitrary length data in a uniform way
<source lang="java">
/*
* 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. */
import java.io.DataInputStream; import java.io.FilterInputStream; import java.io.InputStream; import java.io.IOException; /**
* This input stream works in conjunction with the WrappedOutputStream * to introduce a protocol for reading arbitrary length data in a * uniform way.*
* Note: See the javadoc for WrappedOutputStream for
* more information.
*
* @see WrappedOutputStream
*
* @author Andy Clark, IBM
*
* @version $Id: WrappedInputStream.java 447688 2006-09-19 02:39:49Z mrglavas $
*/
public class WrappedInputStream
extends FilterInputStream {
//
// Data
//
/** Bytes left on input stream for current packet. */
protected int fPacketCount;
/**
* Data input stream. This stream is used to input the block sizes
* from the data stream that are written by the WrappedOutputStream.
* <p>
* Note: The data input stream is only used for
* reading the byte count for performance reasons. We avoid the
* method indirection for reading the byte data.
*/
protected DataInputStream fDataInputStream;
/** To mark that the stream is "closed". */
protected boolean fClosed;
//
// Constructors
//
/** Constructs a wrapper for the given an input stream. */
public WrappedInputStream(InputStream stream) {
super(stream);
fDataInputStream = new DataInputStream(stream);
} // <init>(InputStream)
//
// InputStream methods
//
/** Reads a single byte. */
public int read() throws IOException {
// ignore, if already closed
if (fClosed) {
return -1;
}
// read packet header
if (fPacketCount == 0) {
fPacketCount = fDataInputStream.readInt() & 0x7FFFFFFF;
if (fPacketCount == 0) {
fClosed = true;
return -1;
}
}
// read a byte from the packet
fPacketCount--;
return super.in.read();
} // read():int
/**
* Reads a block of bytes and returns the total number of bytes read.
*/
public int read(byte[] b, int offset, int length) throws IOException {
// ignore, if already closed
if (fClosed) {
return -1;
}
// read packet header
if (fPacketCount == 0) {
fPacketCount = fDataInputStream.readInt() & 0x7FFFFFFF;
if (fPacketCount == 0) {
fClosed = true;
return -1;
}
}
// read bytes from packet
if (length > fPacketCount) {
length = fPacketCount;
}
int count = super.in.read(b, offset, length);
if (count == -1) {
// NOTE: This condition should not happen. The end of
// the stream should always be designated by a
// byte count header of 0. -Ac
fClosed = true;
return -1;
}
fPacketCount -= count;
// return total bytes read
return count;
} // read(byte[],int,int):int
/** Skips the specified number of bytes from the input stream. */
public long skip(long n) throws IOException {
if (!fClosed) {
// NOTE: This should be rewritten to be more efficient. -Ac
for (long i = 0; i < n; i++) {
int b = read();
if (b == -1) {
return i + 1;
}
}
return n;
}
return 0;
} // skip(long):long
/**
* Closes the input stream. This method will search for the end of
* the wrapped input, positioning the stream at after the end packet.
* <p>
* Note: This method does not close the underlying
* input stream.
*/
public void close() throws IOException {
if (!fClosed) {
fClosed = true;
do {
super.in.skip(fPacketCount);
fPacketCount = fDataInputStream.readInt() & 0x7FFFFFFF;
} while (fPacketCount > 0);
}
} // close()
} // class WrappedInputStream
/*
* 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.
*/
import java.io.DataOutputStream;
import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.OutputStream;
/**
* This output stream works in conjunction with the WrappedInputStream
* to introduce a protocol for sending arbitrary length data in a
* uniform way. This output stream allows variable length data to be
* inserted into an existing output stream so that it can be read by
* an input stream without reading too many bytes (in case of buffering
* by the input stream).
* <p>
* This output stream is used like any normal output stream. The protocol
* is introduced by the WrappedOutputStream and does not need to be known
* by the user of this class. However, for those that are interested, the
* method is described below.
* <p>
* The output stream writes the requested bytes as packets of binary
* information. The packet consists of a header and payload. The header
* is two bytes of a single unsigned short (written in network order)
* that specifies the length of bytes in the payload. A header value of
* 0 indicates that the stream is "closed".
* <p>
* Note: For this wrapped output stream to be used,
* the application must call close()
* to end the output.
*
* @see WrappedInputStream
*
* @author Andy Clark, IBM
*
* @version $Id: WrappedOutputStream.java 447688 2006-09-19 02:39:49Z mrglavas $
*/
public class WrappedOutputStream
extends FilterOutputStream {
//
// Constants
//
/** Default buffer size (1024). */
public static final int DEFAULT_BUFFER_SIZE = 1024;
//
// Data
//
/** Buffer. */
protected byte[] fBuffer;
/** Buffer position. */
protected int fPosition;
/**
* Data output stream. This stream is used to output the block sizes
* into the data stream that are read by the WrappedInputStream.
* <p>
* Note: The data output stream is only used for
* writing the byte count for performance reasons. We avoid the
* method indirection for writing the byte data.
*/
protected DataOutputStream fDataOutputStream;
//
// Constructors
//
/** Constructs a wrapper for the given output stream. */
public WrappedOutputStream(OutputStream stream) {
this(stream, DEFAULT_BUFFER_SIZE);
} // <init>(OutputStream)
/**
* Constructs a wrapper for the given output stream with the
* given buffer size.
*/
public WrappedOutputStream(OutputStream stream, int bufferSize) {
super(stream);
fBuffer = new byte[bufferSize];
fDataOutputStream = new DataOutputStream(stream);
} // <init>(OutputStream)
//
// OutputStream methods
//
/**
* Writes a single byte to the output.
* <p>
* Note: Single bytes written to the output stream
* will be buffered
*/
public void write(int b) throws IOException {
fBuffer[fPosition++] = (byte)b;
if (fPosition == fBuffer.length) {
fPosition = 0;
fDataOutputStream.writeInt(fBuffer.length);
super.out.write(fBuffer, 0, fBuffer.length);
}
} // write(int)
/** Writes an array of bytes to the output. */
public void write(byte[] b, int offset, int length)
throws IOException {
// flush existing buffer
if (fPosition > 0) {
flush0();
}
// write header followed by actual bytes
fDataOutputStream.writeInt(length);
super.out.write(b, offset, length);
} // write(byte[])
/**
* Flushes the output buffer, writing all bytes currently in
* the buffer to the output.
*/
public void flush() throws IOException {
flush0();
super.out.flush();
} // flush()
/**
* Closes the output stream. This method must be
* called when done writing all data to the output stream.
* <p>
* Note: This method does not close the
* actual output stream, only makes the input stream see the stream
* closed. Do not write bytes after closing the output stream.
*/
public void close() throws IOException {
flush0();
fDataOutputStream.writeInt(0);
super.out.flush();
} // close()
//
// Protected methods
//
/**
* Flushes the output buffer, writing all bytes currently in
* the buffer to the output. This method does not call the
* flush() method of the output stream; it merely writes the
* remaining bytes in the buffer.
*/
public void flush0() throws IOException {
int length = fPosition;
fPosition = 0;
if (length > 0) {
fDataOutputStream.writeInt(length);
super.out.write(fBuffer, 0, length);
}
} // flush0()
} // class WrappedOutputStream
</source>
