stan/jdk-24
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge

Browse Source
This commit is contained in:
Tim Bell 2009-05-04 22:16:02 -07:00
commit 7a94d5e47f
14 changed files with 471 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
jdk/src/share/classes/java/io/Console.java
View File

@ -503,20 +503,25 @@ public final class Console implements Flushable
// Set up JavaIOAccess in SharedSecrets
static {
// Add a shutdown hook to restore console's echo state should
// it be necessary.
sun.misc.SharedSecrets.getJavaLangAccess()
.registerShutdownHook(0 /* shutdown hook invocation order */,
new Runnable() {
public void run() {
try {
if (echoOff) {
echo(true);
}
} catch (IOException x) { }
}
});
try {
// Add a shutdown hook to restore console's echo state should
// it be necessary.
sun.misc.SharedSecrets.getJavaLangAccess()
.registerShutdownHook(0 /* shutdown hook invocation order */,
false /* only register if shutdown is not in progress */,
new Runnable() {
public void run() {
try {
if (echoOff) {
echo(true);
}
} catch (IOException x) { }
}
});
} catch (IllegalStateException e) {
// shutdown is already in progress and console is first used
// by a shutdown hook
}
sun.misc.SharedSecrets.setJavaIOAccess(new sun.misc.JavaIOAccess() {
public Console console() {

30
jdk/src/share/classes/java/io/DeleteOnExitHook.java
View File

@ -34,23 +34,31 @@ import java.io.File;
*/
class DeleteOnExitHook {
static {
sun.misc.SharedSecrets.getJavaLangAccess()
.registerShutdownHook(2 /* Shutdown hook invocation order */,
new Runnable() {
public void run() {
runHooks();
}
});
}
private static LinkedHashSet<String> files = new LinkedHashSet<String>();
static {
// DeleteOnExitHook must be the last shutdown hook to be invoked.
// Application shutdown hooks may add the first file to the
// delete on exit list and cause the DeleteOnExitHook to be
// registered during shutdown in progress. So set the
// registerShutdownInProgress parameter to true.
sun.misc.SharedSecrets.getJavaLangAccess()
.registerShutdownHook(2 /* Shutdown hook invocation order */,
true /* register even if shutdown in progress */,
new Runnable() {
public void run() {
runHooks();
}
}
);
}
private DeleteOnExitHook() {}
static synchronized void add(String file) {
if(files == null)
if(files == null) {
// DeleteOnExitHook is running. Too late to add a file
throw new IllegalStateException("Shutdown in progress");
}
files.add(file);
}

23
jdk/src/share/classes/java/lang/ApplicationShutdownHooks.java
View File

@ -35,17 +35,26 @@ import java.util.*;
*/
class ApplicationShutdownHooks {
/* The set of registered hooks */
private static IdentityHashMap<Thread, Thread> hooks;
static {
Shutdown.add(1 /* shutdown hook invocation order */,
new Runnable() {
public void run() {
runHooks();
try {
Shutdown.add(1 /* shutdown hook invocation order */,
false /* not registered if shutdown in progress */,
new Runnable() {
public void run() {
runHooks();
}
}
});
);
hooks = new IdentityHashMap<Thread, Thread>();
} catch (IllegalStateException e) {
// application shutdown hooks cannot be added if
// shutdown is in progress.
hooks = null;
}
}
/* The set of registered hooks */
private static IdentityHashMap<Thread, Thread> hooks = new IdentityHashMap<Thread, Thread>();
private ApplicationShutdownHooks() {}

47
jdk/src/share/classes/java/lang/Shutdown.java
View File

@ -53,6 +53,9 @@ class Shutdown {
private static final int MAX_SYSTEM_HOOKS = 10;
private static final Runnable[] hooks = new Runnable[MAX_SYSTEM_HOOKS];
// the index of the currently running shutdown hook to the hooks array
private static int currentRunningHook = 0;
/* The preceding static fields are protected by this lock */
private static class Lock { };
private static Object lock = new Lock();
@ -68,17 +71,39 @@ class Shutdown {
}
/* Add a new shutdown hook. Checks the shutdown state and the hook itself,
/**
* Add a new shutdown hook. Checks the shutdown state and the hook itself,
* but does not do any security checks.
*
* The registerShutdownInProgress parameter should be false except
* registering the DeleteOnExitHook since the first file may
* be added to the delete on exit list by the application shutdown
* hooks.
*
* @params slot the slot in the shutdown hook array, whose element
* will be invoked in order during shutdown
* @params registerShutdownInProgress true to allow the hook
* to be registered even if the shutdown is in progress.
* @params hook the hook to be registered
*
* @throw IllegalStateException
* if registerShutdownInProgress is false and shutdown is in progress; or
* if registerShutdownInProgress is true and the shutdown process
* already passes the given slot
*/
static void add(int slot, Runnable hook) {
static void add(int slot, boolean registerShutdownInProgress, Runnable hook) {
synchronized (lock) {
if (state > RUNNING)
throw new IllegalStateException("Shutdown in progress");
if (hooks[slot] != null)
throw new InternalError("Shutdown hook at slot " + slot + " already registered");
if (!registerShutdownInProgress) {
if (state > RUNNING)
throw new IllegalStateException("Shutdown in progress");
} else {
if (state > HOOKS || (state == HOOKS && slot <= currentRunningHook))
throw new IllegalStateException("Shutdown in progress");
}
hooks[slot] = hook;
}
}
@ -86,11 +111,15 @@ class Shutdown {
/* Run all registered shutdown hooks
*/
private static void runHooks() {
/* We needn't bother acquiring the lock just to read the hooks field,
* since the hooks can't be modified once shutdown is in progress
*/
for (Runnable hook : hooks) {
for (int i=0; i < MAX_SYSTEM_HOOKS; i++) {
try {
Runnable hook;
synchronized (lock) {
// acquire the lock to make sure the hook registered during
// shutdown is visible here.
currentRunningHook = i;
hook = hooks[i];
}
if (hook != null) hook.run();
} catch(Throwable t) {
if (t instanceof ThreadDeath) {

4
jdk/src/share/classes/java/lang/System.java
View File

@ -1171,8 +1171,8 @@ public final class System {
public void blockedOn(Thread t, Interruptible b) {
t.blockedOn(b);
}
public void registerShutdownHook(int slot, Runnable r) {
Shutdown.add(slot, r);
public void registerShutdownHook(int slot, boolean registerShutdownInProgress, Runnable hook) {
Shutdown.add(slot, registerShutdownInProgress, hook);
}
});
}

6
jdk/src/share/classes/java/util/zip/ZipFile.java
View File

@ -154,7 +154,7 @@ class ZipFile implements ZipConstants {
* @param file the ZIP file to be opened for reading
* @param mode the mode in which the file is to be opened
* @param charset
* the {@link java.nio.charset.Charset {@code charset}} to
* the {@linkplain java.nio.charset.Charset charset} to
* be used to decode the ZIP entry name and comment that are not
* encoded by using UTF-8 encoding (indicated by entry's general
* purpose flag).
@ -206,7 +206,7 @@ class ZipFile implements ZipConstants {
*
* @param name the name of the zip file
* @param charset
* the {@link java.nio.charset.Charset {@code charset}} to
* the {@linkplain java.nio.charset.Charset charset} to
* be used to decode the ZIP entry name and comment that are not
* encoded by using UTF-8 encoding (indicated by entry's general
* purpose flag).
@ -230,7 +230,7 @@ class ZipFile implements ZipConstants {
* Opens a ZIP file for reading given the specified File object.
* @param file the ZIP file to be opened for reading
* @param charset
* The {@link java.nio.charset.Charset {@code charset}} to be
* The {@linkplain java.nio.charset.Charset charset} to be
* used to decode the ZIP entry name and comment (ignored if
* the <a href="package-summary.html#lang_encoding"> language
* encoding bit</a> of the ZIP entry's general purpose bit

2
jdk/src/share/classes/java/util/zip/ZipInputStream.java
View File

@ -84,7 +84,7 @@ class ZipInputStream extends InflaterInputStream implements ZipConstants {
* @param in the actual input stream
*
* @param charset
* The {@link java.nio.charset.Charset {@code charset}} to be
* The {@linkplain java.nio.charset.Charset charset} to be
* used to decode the ZIP entry name (ignored if the
* <a href="package-summary.html#lang_encoding"> language
* encoding bit</a> of the ZIP entry's general purpose bit

2
jdk/src/share/classes/java/util/zip/ZipOutputStream.java
View File

@ -108,7 +108,7 @@ class ZipOutputStream extends DeflaterOutputStream implements ZipConstants {
*
* @param out the actual output stream
*
* @param charset the {@link java.nio.charset.Charset </code>charset<code>}
* @param charset the {@linkplain java.nio.charset.Charset charset}
* to be used to encode the entry names and comments
*
* @since 1.7

20
jdk/src/share/classes/sun/misc/JavaLangAccess.java
View File

@ -55,6 +55,22 @@ public interface JavaLangAccess {
/** Set thread's blocker field. */
void blockedOn(Thread t, Interruptible b);
/** register shutdown hook */
void registerShutdownHook(int slot, Runnable r);
/**
* Registers a shutdown hook.
*
* It is expected that this method with registerShutdownInProgress=true
* is only used to register DeleteOnExitHook since the first file
* may be added to the delete on exit list by the application shutdown
* hooks.
*
* @params slot the slot in the shutdown hook array, whose element
* will be invoked in order during shutdown
* @params registerShutdownInProgress true to allow the hook
* to be registered even if the shutdown is in progress.
* @params hook the hook to be registered
*
* @throw IllegalStateException if shutdown is in progress and
* the slot is not valid to register.
*/
void registerShutdownHook(int slot, boolean registerShutdownInProgress, Runnable hook);
}

112
jdk/src/windows/classes/sun/nio/ch/WindowsAsynchronousSocketChannelImpl.java
View File

@ -475,49 +475,40 @@ class WindowsAsynchronousSocketChannelImpl
// get an OVERLAPPED structure (from the cache or allocate)
overlapped = ioCache.add(result);
// synchronize on result to allow this thread handle the case
// where the read completes immediately.
synchronized (result) {
int n = read0(handle, numBufs, readBufferArray, overlapped);
if (n == IOStatus.UNAVAILABLE) {
// I/O is pending
pending = true;
return;
}
// read completed immediately:
// 1. update buffer position
// 2. reset read flag
// 3. release waiters
if (n == 0) {
n = -1;
} else {
updateBuffers(n);
}
// initiate read
int n = read0(handle, numBufs, readBufferArray, overlapped);
if (n == IOStatus.UNAVAILABLE) {
// I/O is pending
pending = true;
return;
}
if (n == IOStatus.EOF) {
// input shutdown
enableReading();
if (scatteringRead) {
result.setResult((V)Long.valueOf(n));
result.setResult((V)Long.valueOf(-1L));
} else {
result.setResult((V)Integer.valueOf(n));
result.setResult((V)Integer.valueOf(-1));
}
} else {
throw new InternalError("Read completed immediately");
}
} catch (Throwable x) {
// failed to initiate read:
// 1. reset read flag
// 2. free resources
// 3. release waiters
// failed to initiate read
// reset read flag before releasing waiters
enableReading();
if (overlapped != 0L)
ioCache.remove(overlapped);
if (x instanceof ClosedChannelException)
x = new AsynchronousCloseException();
if (!(x instanceof IOException))
x = new IOException(x);
result.setFailure(x);
} finally {
if (prepared && !pending) {
// return direct buffer(s) to cache if substituted
releaseBuffers();
// release resources if I/O not pending
if (!pending) {
if (overlapped != 0L)
ioCache.remove(overlapped);
if (prepared)
releaseBuffers();
}
end();
}
@ -721,7 +712,6 @@ class WindowsAsynchronousSocketChannelImpl
@Override
@SuppressWarnings("unchecked")
public void run() {
int n = -1;
long overlapped = 0L;
boolean prepared = false;
boolean pending = false;
@ -736,56 +726,34 @@ class WindowsAsynchronousSocketChannelImpl
// get an OVERLAPPED structure (from the cache or allocate)
overlapped = ioCache.add(result);
// synchronize on result to allow this thread handle the case
// where the read completes immediately.
synchronized (result) {
n = write0(handle, numBufs, writeBufferArray, overlapped);
if (n == IOStatus.UNAVAILABLE) {
// I/O is pending
pending = true;
return;
}
enableWriting();
if (n == IOStatus.EOF) {
// special case for shutdown output
shutdown = true;
throw new ClosedChannelException();
}
// write completed immediately:
// 1. enable writing
// 2. update buffer position
// 3. release waiters
updateBuffers(n);
// result is a Long or Integer
if (gatheringWrite) {
result.setResult((V)Long.valueOf(n));
} else {
result.setResult((V)Integer.valueOf(n));
}
int n = write0(handle, numBufs, writeBufferArray, overlapped);
if (n == IOStatus.UNAVAILABLE) {
// I/O is pending
pending = true;
return;
}
if (n == IOStatus.EOF) {
// special case for shutdown output
shutdown = true;
throw new ClosedChannelException();
}
// write completed immediately
throw new InternalError("Write completed immediately");
} catch (Throwable x) {
// write failed. Enable writing before releasing waiters.
enableWriting();
// failed to initiate read:
if (!shutdown && (x instanceof ClosedChannelException))
x = new AsynchronousCloseException();
if (!(x instanceof IOException))
x = new IOException(x);
result.setFailure(x);
// release resources
if (overlapped != 0L)
ioCache.remove(overlapped);
} finally {
if (prepared && !pending) {
// return direct buffer(s) to cache if substituted
releaseBuffers();
// release resources if I/O not pending
if (!pending) {
if (overlapped != 0L)
ioCache.remove(overlapped);
if (prepared)
releaseBuffers();
}
end();
}

17
jdk/src/windows/native/sun/nio/ch/WindowsAsynchronousSocketChannelImpl.c
View File

@ -157,14 +157,13 @@ Java_sun_nio_ch_WindowsAsynchronousSocketChannelImpl_read0(JNIEnv* env, jclass t
WSABUF* lpWsaBuf = (WSABUF*) jlong_to_ptr(address);
OVERLAPPED* lpOverlapped = (OVERLAPPED*) jlong_to_ptr(ov);
BOOL res;
DWORD nread = 0;
DWORD flags = 0;
ZeroMemory((PVOID)lpOverlapped, sizeof(OVERLAPPED));
res = WSARecv(s,
lpWsaBuf,
(DWORD)count,
&nread,
NULL,
&flags,
lpOverlapped,
NULL);
@ -175,17 +174,12 @@ Java_sun_nio_ch_WindowsAsynchronousSocketChannelImpl_read0(JNIEnv* env, jclass t
return IOS_UNAVAILABLE;
}
if (error == WSAESHUTDOWN) {
return 0; // input shutdown
return IOS_EOF; // input shutdown
}
JNU_ThrowIOExceptionWithLastError(env, "WSARecv failed");
return IOS_THROWN;
}
if (nread == 0) {
// Handle graceful close or bytes not yet available cases
// via completion port notification.
return IOS_UNAVAILABLE;
}
return (jint)nread;
return IOS_UNAVAILABLE;
}
JNIEXPORT jint JNICALL
@ -196,13 +190,12 @@ Java_sun_nio_ch_WindowsAsynchronousSocketChannelImpl_write0(JNIEnv* env, jclass
WSABUF* lpWsaBuf = (WSABUF*) jlong_to_ptr(address);
OVERLAPPED* lpOverlapped = (OVERLAPPED*) jlong_to_ptr(ov);
BOOL res;
DWORD nwritten;
ZeroMemory((PVOID)lpOverlapped, sizeof(OVERLAPPED));
res = WSASend(s,
lpWsaBuf,
(DWORD)count,
&nwritten,
NULL,
0,
lpOverlapped,
NULL);
@ -218,5 +211,5 @@ Java_sun_nio_ch_WindowsAsynchronousSocketChannelImpl_write0(JNIEnv* env, jclass
JNU_ThrowIOExceptionWithLastError(env, "WSASend failed");
return IOS_THROWN;
}
return (jint)nwritten;
return IOS_UNAVAILABLE;
}

69
jdk/test/java/lang/Runtime/shutdown/ShutdownHooks.java Normal file
View File

@ -0,0 +1,69 @@
/*
* Copyright 2009 Sun Microsystems, Inc. All Rights Reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
* CA 95054 USA or visit www.sun.com if you need additional information or
* have any questions.
*/
/*
* @bug 6829503
* @summary 1) Test Console and DeleteOnExitHook can be initialized
* while shutdown is in progress
* 2) Test if files that are added by the application shutdown
* hook are deleted on exit during shutdown
*/
import java.io.*;
public class ShutdownHooks {
private static File file;
public static void main(String[] args) throws Exception {
if (args.length != 2) {
throw new IllegalArgumentException("Usage: ShutdownHooks <dir> <filename>");
}
// Add a shutdown hook
Runtime.getRuntime().addShutdownHook(new Cleaner());
File dir = new File(args[0]);
file = new File(dir, args[1]);
// write to file
System.out.println("writing to "+ file);
PrintWriter pw = new PrintWriter(file);
pw.println("Shutdown begins");
pw.close();
}
public static class Cleaner extends Thread {
public void run() {
// register the Console's shutdown hook while the application
// shutdown hook is running
Console cons = System.console();
// register the DeleteOnExitHook while the application
// shutdown hook is running
file.deleteOnExit();
try {
PrintWriter pw = new PrintWriter(file);
pw.println("file is being deleted");
pw.close();
} catch (FileNotFoundException e) {
throw new RuntimeException(e);
}
}
}
}

57
jdk/test/java/lang/Runtime/shutdown/ShutdownHooks.sh Normal file
View File

@ -0,0 +1,57 @@
#!/bin/sh
#
# Copyright 2009 Sun Microsystems, Inc. All Rights Reserved.
# DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
#
# This code is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 2 only, as
# published by the Free Software Foundation.
#
# This code is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
# version 2 for more details (a copy is included in the LICENSE file that
# accompanied this code).
#
# You should have received a copy of the GNU General Public License version
# 2 along with this work; if not, write to the Free Software Foundation,
# Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
# CA 95054 USA or visit www.sun.com if you need additional information or
# have any questions.
#
# @test
# @bug 6829503
# @summary 1) Test Console and DeleteOnExitHook can be initialized
# while shutdown is in progress
# 2) Test if files that are added by the application shutdown
# hook are deleted on exit during shutdown
#
# @build ShutdownHooks
# @run shell ShutdownHooks.sh
if [ "${TESTJAVA}" = "" ]
then
echo "TESTJAVA not set. Test cannot execute. Failed."
exit 1
fi
FILENAME=fileToBeDeleted
rm -f ${TESTCLASSES}/${FILENAME}
# create the file to be deleted on exit
echo "testing shutdown" > ${TESTCLASSES}/${FILENAME}
${TESTJAVA}/bin/java ${TESTVMOPTS} -classpath ${TESTCLASSES} ShutdownHooks ${TESTCLASSES} $FILENAME
if [ $? != 0 ] ; then
echo "Test Failed"; exit 1
fi
if [ -f ${TESTCLASSES}/${FILENAME} ]; then
echo "Test Failed: ${TESTCLASSES}/${FILENAME} not deleted"; exit 2
fi
echo "ShutdownHooks test passed.";

183
jdk/test/java/nio/channels/AsynchronousSocketChannel/StressLoopback.java Normal file
View File

@ -0,0 +1,183 @@
/*
* Copyright 2008-2009 Sun Microsystems, Inc. All Rights Reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
* CA 95054 USA or visit www.sun.com if you need additional information or
* have any questions.
*/
/* @test
* @bug 6834246
* @summary Stress test connections through the loopback interface
*/
import java.nio.ByteBuffer;
import java.net.*;
import java.nio.channels.*;
import java.util.Random;
import java.io.IOException;
public class StressLoopback {
static final Random rand = new Random();
public static void main(String[] args) throws Exception {
// setup listener
AsynchronousServerSocketChannel listener =
AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(0));
int port =((InetSocketAddress)(listener.getLocalAddress())).getPort();
InetAddress lh = InetAddress.getLocalHost();
SocketAddress remote = new InetSocketAddress(lh, port);
// create sources and sinks
int count = 2 + rand.nextInt(9);
Source[] source = new Source[count];
Sink[] sink = new Sink[count];
for (int i=0; i<count; i++) {
AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
ch.connect(remote).get();
source[i] = new Source(ch);
sink[i] = new Sink(listener.accept().get());
}
// start the sinks and sources
for (int i=0; i<count; i++) {
sink[i].start();
source[i].start();
}
// let the test run for a while
Thread.sleep(20*1000);
// wait until everyone is done
boolean failed = false;
long total = 0L;
for (int i=0; i<count; i++) {
long nwrote = source[i].finish();
long nread = sink[i].finish();
if (nread != nwrote)
failed = true;
System.out.format("%d -> %d (%s)\n",
nwrote, nread, (failed) ? "FAIL" : "PASS");
total += nwrote;
}
if (failed)
throw new RuntimeException("Test failed - see log for details");
System.out.format("Total sent %d MB\n", total / (1024L * 1024L));
}
/**
* Writes bytes to a channel until "done". When done the channel is closed.
*/
static class Source {
private final AsynchronousByteChannel channel;
private final ByteBuffer sentBuffer;
private volatile long bytesSent;
private volatile boolean finished;
Source(AsynchronousByteChannel channel) {
this.channel = channel;
int size = 1024 + rand.nextInt(10000);
this.sentBuffer = (rand.nextBoolean()) ?
ByteBuffer.allocateDirect(size) : ByteBuffer.allocate(size);
}
void start() {
sentBuffer.position(0);
sentBuffer.limit(sentBuffer.capacity());
channel.write(sentBuffer, null, new CompletionHandler<Integer,Void> () {
public void completed(Integer nwrote, Void att) {
bytesSent += nwrote;
if (finished) {
closeUnchecked(channel);
} else {
sentBuffer.position(0);
sentBuffer.limit(sentBuffer.capacity());
channel.write(sentBuffer, null, this);
}
}
public void failed(Throwable exc, Void att) {
exc.printStackTrace();
closeUnchecked(channel);
}
public void cancelled(Void att) {
}
});
}
long finish() {
finished = true;
waitUntilClosed(channel);
return bytesSent;
}
}
/**
* Read bytes from a channel until EOF is received.
*/
static class Sink {
private final AsynchronousByteChannel channel;
private final ByteBuffer readBuffer;
private volatile long bytesRead;
Sink(AsynchronousByteChannel channel) {
this.channel = channel;
int size = 1024 + rand.nextInt(10000);
this.readBuffer = (rand.nextBoolean()) ?
ByteBuffer.allocateDirect(size) : ByteBuffer.allocate(size);
}
void start() {
channel.read(readBuffer, null, new CompletionHandler<Integer,Void> () {
public void completed(Integer nread, Void att) {
if (nread < 0) {
closeUnchecked(channel);
} else {
bytesRead += nread;
readBuffer.clear();
channel.read(readBuffer, null, this);
}
}
public void failed(Throwable exc, Void att) {
exc.printStackTrace();
closeUnchecked(channel);
}
public void cancelled(Void att) {
}
});
}
long finish() {
waitUntilClosed(channel);
return bytesRead;
}
}
static void waitUntilClosed(Channel c) {
while (c.isOpen()) {
try {
Thread.sleep(100);
} catch (InterruptedException ignore) { }
}
}
static void closeUnchecked(Channel c) {
try {
c.close();
} catch (IOException ignore) { }
}
}