- public byte[] get(int bufferSize) throws EOFException, IOException {
- byte[] buffer = new byte[bufferSize];
- int read = pipedInputStream.read(buffer);
- if (read == -1) {
- throw new EOFException();
- }
- return Arrays.copyOf(buffer, read);
- }
-
- @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);
+ public void process(byte[] buffer) throws IOException {
+ super.process(buffer);
+ /* delay. */
+ counter += buffer.length;
+ long waitTime = (long) (counter / (rate / 1000.0));
+ while ((System.currentTimeMillis() - startTime) < waitTime) {
+ try {
+ long limitDelay = waitTime - (System.currentTimeMillis() - startTime);
+ logger.finest(String.format("Waiting %d ms...", limitDelay));
+ Thread.sleep(limitDelay);
+ } catch (InterruptedException ie1) {
+ /* ignore, keep looping. */
+ }