package net.pterodactylus.sonitus.data.filter;
-import static com.google.common.io.Closeables.close;
-
import java.io.EOFException;
import java.io.IOException;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
-import java.util.Arrays;
import java.util.logging.Logger;
import net.pterodactylus.sonitus.data.ConnectException;
-import net.pterodactylus.sonitus.data.Connection;
import net.pterodactylus.sonitus.data.Filter;
-import net.pterodactylus.sonitus.data.Format;
+import net.pterodactylus.sonitus.data.Metadata;
import net.pterodactylus.sonitus.data.Source;
import com.google.common.base.Preconditions;
/** The limiting rate in bytes/second. */
private final int rate;
- /** The source’s format. */
- private Format format;
+ /** The fast start time. */
+ private final long fastStartTime;
+
+ /** The source. */
+ private Source source;
- /** The input stream to read from. */
- private PipedInputStream pipedInputStream = new PipedInputStream();
+ /** The remaining fast start time. */
+ private long remainingFastStartTime;
/**
* Creates a new rate limiting filter.
* The limiting rate (in bytes/second)
*/
public RateLimitingFilter(int rate) {
+ this(rate, 0);
+ }
+
+ /**
+ * Creates a new rate limiting filter.
+ *
+ * @param rate
+ * The limiting rate (in bytes/second)
+ * @param fastStartTime
+ * The amount of time at the start of the filtering during which no delay
+ * will occur (in milliseconds)
+ */
+ public RateLimitingFilter(int rate, long fastStartTime) {
this.rate = rate;
+ this.fastStartTime = fastStartTime;
+ remainingFastStartTime = fastStartTime;
}
//
//
@Override
- public Format format() {
- return format;
+ public Metadata metadata() {
+ return source.metadata();
}
@Override
public byte[] get(int bufferSize) throws EOFException, IOException {
- byte[] buffer = new byte[bufferSize];
- int read = pipedInputStream.read(buffer);
- if (read == -1) {
- throw new EOFException();
+ long now = System.currentTimeMillis();
+ byte[] buffer = source.get(bufferSize);
+ /* delay. */
+ long waitTime = 1000 * buffer.length / rate;
+ remainingFastStartTime = Math.max(remainingFastStartTime - waitTime, 0);
+ while ((remainingFastStartTime == 0) && (System.currentTimeMillis() - now) < waitTime) {
+ try {
+ long limitDelay = waitTime - (System.currentTimeMillis() - now);
+ logger.finest(String.format("Waiting %d ms...", limitDelay));
+ Thread.sleep(limitDelay);
+ } catch (InterruptedException ie1) {
+ /* ignore, keep looping. */
+ }
}
- return Arrays.copyOf(buffer, read);
+ return buffer;
}
@Override
public void connect(Source source) throws ConnectException {
Preconditions.checkNotNull(source, "source must not be null");
- format = source.format();
- final long start = System.currentTimeMillis();
- try {
- pipedInputStream = new PipedInputStream();
- final PipedOutputStream pipedOutputStream = new PipedOutputStream(pipedInputStream);
- new Thread(new Connection(source) {
-
- @Override
- protected int bufferSize() {
- return rate;
- }
-
- @Override
- protected void feed(byte[] buffer) throws IOException {
- long waitTime = 1000 * buffer.length / rate;
- long now = System.currentTimeMillis();
- pipedOutputStream.write(buffer);
- pipedOutputStream.flush();
- while ((System.currentTimeMillis() - now) < waitTime) {
- try {
- long limitDelay = waitTime - (System.currentTimeMillis() - now);
- logger.finest(String.format("Waiting %d ms...", limitDelay));
- Thread.sleep(limitDelay);
- } catch (InterruptedException ie1) {
- /* ignore, keep looping. */
- }
- }
- }
-
- @Override
- protected void finish() throws IOException {
- close(pipedInputStream, true);
- close(pipedOutputStream, true);
- }
- }).start();
- } catch (IOException ioe1) {
- throw new ConnectException(ioe1);
- }
+ this.source = source;
+ }
+
+ @Override
+ public void metadataUpdated() {
+ /* ignore. */
}
}