archive/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2025-09-21 11:34:38 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

6993126: (aio) remove AsynchronousDatagramChannel

Browse source 
Reviewed-by: chegar
This commit is contained in:
Alan Bateman 2010-11-02 10:05:42 +00:00
parent aab01a90d7
commit c51ac2a369
12 changed files with 4 additions and 1733 deletions
Download patch file Download diff file Expand all files Collapse all files

2
jdk/make/java/nio/FILES_java.gmk
View file

@ -33,7 +33,6 @@ FILES_src = \
java/nio/channels/AsynchronousByteChannel.java \
java/nio/channels/AsynchronousChannel.java \
java/nio/channels/AsynchronousChannelGroup.java \
java/nio/channels/AsynchronousDatagramChannel.java \
java/nio/channels/AsynchronousFileChannel.java \
java/nio/channels/AsynchronousServerSocketChannel.java \
java/nio/channels/AsynchronousSocketChannel.java \
@ -207,7 +206,6 @@ FILES_src = \
sun/nio/ch/SelChImpl.java \
sun/nio/ch/ServerSocketAdaptor.java \
sun/nio/ch/ServerSocketChannelImpl.java \
sun/nio/ch/SimpleAsynchronousDatagramChannelImpl.java \
sun/nio/ch/SinkChannelImpl.java \
sun/nio/ch/SocketAdaptor.java \
sun/nio/ch/SocketChannelImpl.java \

572
jdk/src/share/classes/java/nio/channels/AsynchronousDatagramChannel.java
View file

@ -1,572 +0,0 @@
/*
* Copyright (c) 2007, 2009, Oracle and/or its affiliates. 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. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* 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 Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package java.nio.channels;
import java.nio.channels.spi.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.Future;
import java.io.IOException;
import java.net.SocketOption;
import java.net.SocketAddress;
import java.net.ProtocolFamily;
import java.nio.ByteBuffer;
/**
* An asynchronous channel for datagram-oriented sockets.
*
* <p> An asynchronous datagram channel is created by invoking one of the {@link
* #open open} methods defined by this class. It is not possible to create a channel
* for an arbitrary, pre-existing datagram socket. A newly-created asynchronous
* datagram channel is open but not connected. It need not be connected in order
* for the {@link #send send} and {@link #receive receive} methods to be used.
* A datagram channel may be connected, by invoking its {@link #connect connect}
* method, in order to avoid the overhead of the security checks that are otherwise
* performed as part of every send and receive operation when a security manager
* is set. The channel must be connected in order to use the {@link #read read}
* and {@link #write write} methods, since those methods do not accept or return
* socket addresses. Once connected, an asynchronous datagram channel remains
* connected until it is disconnected or closed.
*
* <p> Socket options are configured using the {@link #setOption(SocketOption,Object)
* setOption} method. An asynchronous datagram channel to an Internet Protocol
* (IP) socket supports the following options:
* <blockquote>
* <table border>
* <tr>
* <th>Option Name</th>
* <th>Description</th>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#SO_SNDBUF SO_SNDBUF} </td>
* <td> The size of the socket send buffer </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#SO_RCVBUF SO_RCVBUF} </td>
* <td> The size of the socket receive buffer </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#SO_REUSEADDR SO_REUSEADDR} </td>
* <td> Re-use address </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#SO_BROADCAST SO_BROADCAST} </td>
* <td> Allow transmission of broadcast datagrams </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#IP_TOS IP_TOS} </td>
* <td> The Type of Service (ToS) octet in the Internet Protocol (IP) header </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#IP_MULTICAST_IF IP_MULTICAST_IF} </td>
* <td> The network interface for Internet Protocol (IP) multicast datagrams </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#IP_MULTICAST_TTL
* IP_MULTICAST_TTL} </td>
* <td> The <em>time-to-live</em> for Internet Protocol (IP) multicast
* datagrams </td>
* </tr>
* <tr>
* <td> {@link java.net.StandardSocketOption#IP_MULTICAST_LOOP
* IP_MULTICAST_LOOP} </td>
* <td> Loopback for Internet Protocol (IP) multicast datagrams </td>
* </tr>
* </table>
* </blockquote>
* Additional (implementation specific) options may also be supported.
*
* <p> Asynchronous datagram channels allow more than one read/receive and
* write/send to be oustanding at any given time.
*
* <p> <b>Usage Example:</b>
* <pre>
* final AsynchronousDatagramChannel dc = AsynchronousDatagramChannel.open()
* .bind(new InetSocketAddress(4000));
*
* // print the source address of all packets that we receive
* dc.receive(buffer, buffer, new CompletionHandler&lt;SocketAddress,ByteBuffer&gt;() {
* public void completed(SocketAddress sa, ByteBuffer buffer) {
* System.out.println(sa);
* buffer.clear();
* dc.receive(buffer, buffer, this);
* }
* public void failed(Throwable exc, ByteBuffer buffer) {
* ...
* }
* });
* </pre>
*
* @since 1.7
*/
public abstract class AsynchronousDatagramChannel
implements AsynchronousByteChannel, MulticastChannel
{
private final AsynchronousChannelProvider provider;
/**
* Initializes a new instance of this class.
*/
protected AsynchronousDatagramChannel(AsynchronousChannelProvider provider) {
this.provider = provider;
}
/**
* Returns the provider that created this channel.
*/
public final AsynchronousChannelProvider provider() {
return provider;
}
/**
* Opens an asynchronous datagram channel.
*
* <p> The new channel is created by invoking the {@link
* java.nio.channels.spi.AsynchronousChannelProvider#openAsynchronousDatagramChannel
* openAsynchronousDatagramChannel} method on the {@link
* java.nio.channels.spi.AsynchronousChannelProvider} object that created
* the given group (or the default provider where {@code group} is {@code
* null}).
*
* <p> The {@code family} parameter is used to specify the {@link ProtocolFamily}.
* If the datagram channel is to be used for Internet Protocol {@link
* MulticastChannel multicasting} then this parameter should correspond to
* the address type of the multicast groups that this channel will join.
*
* @param family
* The protocol family, or {@code null} to use the default protocol
* family
* @param group
* The group to which the newly constructed channel should be bound,
* or {@code null} for the default group
*
* @return A new asynchronous datagram channel
*
* @throws UnsupportedOperationException
* If the specified protocol family is not supported. For example,
* suppose the parameter is specified as {@link
* java.net.StandardProtocolFamily#INET6 INET6} but IPv6 is not
* enabled on the platform.
* @throws ShutdownChannelGroupException
* The specified group is shutdown
* @throws IOException
* If an I/O error occurs
*/
public static AsynchronousDatagramChannel open(ProtocolFamily family,
AsynchronousChannelGroup group)
throws IOException
{
AsynchronousChannelProvider provider = (group == null) ?
AsynchronousChannelProvider.provider() : group.provider();
return provider.openAsynchronousDatagramChannel(family, group);
}
/**
* Opens an asynchronous datagram channel.
*
* <p> This method returns an asynchronous datagram channel that is
* bound to the <em>default group</em>. This method is equivalent to evaluating
* the expression:
* <blockquote><pre>
* open((ProtocolFamily)null,&nbsp;(AsynchronousChannelGroup)null);
* </pre></blockquote>
*
* @return A new asynchronous datagram channel
*
* @throws IOException
* If an I/O error occurs
*/
public static AsynchronousDatagramChannel open()
throws IOException
{
return open(null, null);
}
// -- Socket-specific operations --
/**
* @throws AlreadyBoundException {@inheritDoc}
* @throws UnsupportedAddressTypeException {@inheritDoc}
* @throws ClosedChannelException {@inheritDoc}
* @throws IOException {@inheritDoc}
* @throws SecurityException
* If a security manager has been installed and its {@link
* SecurityManager#checkListen checkListen} method denies the
* operation
*/
@Override
public abstract AsynchronousDatagramChannel bind(SocketAddress local)
throws IOException;
/**
* @throws IllegalArgumentException {@inheritDoc}
* @throws ClosedChannelException {@inheritDoc}
* @throws IOException {@inheritDoc}
*/
@Override
public abstract <T> AsynchronousDatagramChannel setOption(SocketOption<T> name, T value)
throws IOException;
/**
* Returns the remote address to which this channel is connected.
*
* <p> Where the channel is connected to an Internet Protocol socket address
* then the return value from this method is of type {@link
* java.net.InetSocketAddress}.
*
* @return The remote address; {@code null} if the channel's socket is not
* connected
*
* @throws ClosedChannelException
* If the channel is closed
* @throws IOException
* If an I/O error occurs
*/
public abstract SocketAddress getRemoteAddress() throws IOException;
/**
* Connects this channel's socket.
*
* <p> The channel's socket is configured so that it only receives
* datagrams from, and sends datagrams to, the given remote <i>peer</i>
* address. Once connected, datagrams may not be received from or sent to
* any other address. A datagram socket remains connected until it is
* explicitly disconnected or until it is closed.
*
* <p> This method performs exactly the same security checks as the {@link
* java.net.DatagramSocket#connect connect} method of the {@link
* java.net.DatagramSocket} class. That is, if a security manager has been
* installed then this method verifies that its {@link
* java.lang.SecurityManager#checkAccept checkAccept} and {@link
* java.lang.SecurityManager#checkConnect checkConnect} methods permit
* datagrams to be received from and sent to, respectively, the given
* remote address.
*
* <p> This method may be invoked at any time. Whether it has any effect
* on outstanding read or write operations is implementation specific and
* therefore not specified.
*
* @param remote
* The remote address to which this channel is to be connected
*
* @return This datagram channel
*
* @throws ClosedChannelException
* If this channel is closed
*
* @throws SecurityException
* If a security manager has been installed
* and it does not permit access to the given remote address
*
* @throws IOException
* If some other I/O error occurs
*/
public abstract AsynchronousDatagramChannel connect(SocketAddress remote)
throws IOException;
/**
* Disconnects this channel's socket.
*
* <p> The channel's socket is configured so that it can receive datagrams
* from, and sends datagrams to, any remote address so long as the security
* manager, if installed, permits it.
*
* <p> This method may be invoked at any time. Whether it has any effect
* on outstanding read or write operations is implementation specific and
* therefore not specified.
*
* @return This datagram channel
*
* @throws IOException
* If some other I/O error occurs
*/
public abstract AsynchronousDatagramChannel disconnect() throws IOException;
/**
* Receives a datagram via this channel.
*
* <p> This method initiates the receiving of a datagram into the given
* buffer. The {@code handler} parameter is a completion handler that is
* invoked when the receive operation completes (or fails). The result
* passed to the completion handler is the datagram's source address.
*
* <p> The datagram is transferred into the given byte buffer starting at
* its current position, as if by a regular {@link AsynchronousByteChannel#read
* read} operation. If there are fewer bytes remaining in the buffer
* than are required to hold the datagram then the remainder of the datagram
* is silently discarded.
*
* <p> If a timeout is specified and the timeout elapses before the operation
* completes then the operation completes with the exception {@link
* InterruptedByTimeoutException}. When a timeout elapses then the state of
* the {@link ByteBuffer} is not defined. The buffers should be discarded or
* at least care must be taken to ensure that the buffer is not accessed
* while the channel remains open.
*
* <p> When a security manager has been installed and the channel is not
* connected, then it verifies that the source's address and port number are
* permitted by the security manager's {@link SecurityManager#checkAccept
* checkAccept} method. The permission check is performed with privileges that
* are restricted by the calling context of this method. If the permission
* check fails then the operation completes with a {@link SecurityException}.
* The overhead of this security check can be avoided by first connecting the
* socket via the {@link #connect connect} method.
*
* @param dst
* The buffer into which the datagram is to be transferred
* @param timeout
* The timeout, or {@code 0L} for no timeout
* @param unit
* The time unit of the {@code timeout} argument
* @param attachment
* The object to attach to the I/O operation; can be {@code null}
* @param handler
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the timeout is negative or the buffer is read-only
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
public abstract <A> void receive(ByteBuffer dst,
long timeout,
TimeUnit unit,
A attachment,
CompletionHandler<SocketAddress,? super A> handler);
/**
* Receives a datagram via this channel.
*
* <p> This method initiates the receiving of a datagram into the given
* buffer. The {@code handler} parameter is a completion handler that is
* invoked when the receive operation completes (or fails). The result
* passed to the completion handler is the datagram's source address.
*
* <p> This method is equivalent to invoking {@link
* #receive(ByteBuffer,long,TimeUnit,Object,CompletionHandler)} with a
* timeout of {@code 0L}.
*
* @param dst
* The buffer into which the datagram is to be transferred
* @param attachment
* The object to attach to the I/O operation; can be {@code null}
* @param handler
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the buffer is read-only
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
public final <A> void receive(ByteBuffer dst,
A attachment,
CompletionHandler<SocketAddress,? super A> handler)
{
receive(dst, 0L, TimeUnit.MILLISECONDS, attachment, handler);
}
/**
* Receives a datagram via this channel.
*
* <p> This method initiates the receiving of a datagram into the given
* buffer. The method behaves in exactly the same manner as the {@link
* #receive(ByteBuffer,Object,CompletionHandler)
* receive(ByteBuffer,Object,CompletionHandler)} method except that instead
* of specifying a completion handler, this method returns a {@code Future}
* representing the pending result. The {@code Future}'s {@link Future#get()
* get} method returns the datagram's source address.
*
* @param dst
* The buffer into which the datagram is to be transferred
*
* @return a {@code Future} object representing the pending result
*
* @throws IllegalArgumentException
* If the buffer is read-only
*/
public abstract Future<SocketAddress> receive(ByteBuffer dst);
/**
* Sends a datagram via this channel.
*
* <p> This method initiates sending of a datagram from the given buffer to
* the given address. The {@code handler} parameter is a completion handler
* that is invoked when the send completes (or fails). The result passed to
* the completion handler is the number of bytes sent.
*
* <p> Otherwise this method works in the same manner as the {@link
* AsynchronousByteChannel#write(ByteBuffer,Object,CompletionHandler)}
* method.
*
* @param src
* The buffer containing the datagram to be sent
* @param target
* The address to which the datagram is to be sent
* @param attachment
* The object to attach to the I/O operation; can be {@code null}
* @param handler
* The handler for consuming the result
*
* @throws UnresolvedAddressException
* If the given remote address is not fully resolved
* @throws UnsupportedAddressTypeException
* If the type of the given remote address is not supported
* @throws IllegalArgumentException
* If the channel's socket is connected and is connected to an
* address that is not equal to {@code target}
* @throws SecurityException
* If a security manager has been installed and it does not permit
* datagrams to be sent to the given address
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
public abstract <A> void send(ByteBuffer src,
SocketAddress target,
A attachment,
CompletionHandler<Integer,? super A> handler);
/**
* Sends a datagram via this channel.
*
* <p> This method initiates sending of a datagram from the given buffer to
* the given address. The method behaves in exactly the same manner as the
* {@link #send(ByteBuffer,SocketAddress,Object,CompletionHandler)
* send(ByteBuffer,SocketAddress,Object,CompletionHandler)} method except
* that instead of specifying a completion handler, this method returns a
* {@code Future} representing the pending result. The {@code Future}'s
* {@link Future#get() get} method returns the number of bytes sent.
*
* @param src
* The buffer containing the datagram to be sent
* @param target
* The address to which the datagram is to be sent
*
* @return a {@code Future} object representing the pending result
*
* @throws UnresolvedAddressException
* If the given remote address is not fully resolved
* @throws UnsupportedAddressTypeException
* If the type of the given remote address is not supported
* @throws IllegalArgumentException
* If the channel's socket is connected and is connected to an
* address that is not equal to {@code target}
* @throws SecurityException
* If a security manager has been installed and it does not permit
* datagrams to be sent to the given address
*/
public abstract Future<Integer> send(ByteBuffer src, SocketAddress target);
/**
* Receives a datagram via this channel.
*
* <p> This method initiates the receiving of a datagram into the given
* buffer. The {@code handler} parameter is a completion handler that is
* invoked when the receive operation completes (or fails). The result
* passed to the completion handler is number of bytes read.
*
* <p> This method may only be invoked if this channel is connected, and it
* only accepts datagrams from the peer that the channel is connected too.
* The datagram is transferred into the given byte buffer starting at
* its current position and exactly as specified in the {@link
* AsynchronousByteChannel} interface. If there are fewer bytes
* remaining in the buffer than are required to hold the datagram then the
* remainder of the datagram is silently discarded.
*
* <p> If a timeout is specified and the timeout elapses before the operation
* completes then the operation completes with the exception {@link
* InterruptedByTimeoutException}. When a timeout elapses then the state of
* the {@link ByteBuffer} is not defined. The buffers should be discarded or
* at least care must be taken to ensure that the buffer is not accessed
* while the channel remains open.
*
* @param dst
* The buffer into which the datagram is to be transferred
* @param timeout
* The timeout, or {@code 0L} for no timeout
* @param unit
* The time unit of the {@code timeout} argument
* @param attachment
* The object to attach to the I/O operation; can be {@code null}
* @param handler
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the timeout is negative or buffer is read-only
* @throws NotYetConnectedException
* If this channel is not connected
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
public abstract <A> void read(ByteBuffer dst,
long timeout,
TimeUnit unit,
A attachment,
CompletionHandler<Integer,? super A> handler);
/**
* @throws NotYetConnectedException
* If this channel is not connected
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
@Override
public final <A> void read(ByteBuffer dst,
A attachment,
CompletionHandler<Integer,? super A> handler)
{
read(dst, 0L, TimeUnit.MILLISECONDS, attachment, handler);
}
/**
* @throws NotYetConnectedException
* If this channel is not connected
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
@Override
public abstract Future<Integer> read(ByteBuffer dst);
/**
* @throws NotYetConnectedException
* If this channel is not connected
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
@Override
public abstract <A> void write(ByteBuffer src,
A attachment,
CompletionHandler<Integer,? super A> handler);
/**
* @throws NotYetConnectedException
* If this channel is not connected
* @throws ShutdownChannelGroupException
* If the channel group has terminated
*/
@Override
public abstract Future<Integer> write(ByteBuffer src);
}

2
jdk/src/share/classes/java/nio/channels/package-info.java
View file

@ -232,8 +232,6 @@
* <td>An asynchronous channel to a stream-oriented connecting socket</td></tr>
* <tr><td valign=top><tt>{@link java.nio.channels.AsynchronousServerSocketChannel}&nbsp;&nbsp;</tt></td>
* <td>An asynchronous channel to a stream-oriented listening socket</td></tr>
* <tr><td valign=top><tt>{@link java.nio.channels.AsynchronousDatagramChannel}</tt></td>
* <td>An asynchronous channel to a datagram-oriented socket</td></tr>
* <tr><td valign=top><tt>{@link java.nio.channels.CompletionHandler}</tt></td>
* <td>A handler for consuming the result of an asynchronous operation</td></tr>
* <tr><td valign=top><tt>{@link java.nio.channels.AsynchronousChannelGroup}</tt></td>

23
jdk/src/share/classes/java/nio/channels/spi/AsynchronousChannelProvider.java
View file

@ -26,7 +26,6 @@
package java.nio.channels.spi;
import java.nio.channels.*;
import java.net.ProtocolFamily;
import java.io.IOException;
import java.util.Iterator;
import java.util.ServiceLoader;
@ -239,26 +238,4 @@ public abstract class AsynchronousChannelProvider {
*/
public abstract AsynchronousSocketChannel openAsynchronousSocketChannel
(AsynchronousChannelGroup group) throws IOException;
/**
* Opens an asynchronous datagram channel.
*
* @param family
* The protocol family, or {@code null} for the default protocol
* family
* @param group
* The group to which the channel is bound, or {@code null} to
* bind to the default group
*
* @return The new channel
*
* @throws IllegalChannelGroupException
* If the provider that created the group differs from this provider
* @throws ShutdownChannelGroupException
* The group is shutdown
* @throws IOException
* If an I/O error occurs
*/
public abstract AsynchronousDatagramChannel openAsynchronousDatagramChannel
(ProtocolFamily family, AsynchronousChannelGroup group) throws IOException;
}

667
jdk/src/share/classes/sun/nio/ch/SimpleAsynchronousDatagramChannelImpl.java
View file

@ -1,667 +0,0 @@
/*
* Copyright (c) 2008, 2009, Oracle and/or its affiliates. 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. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* 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 Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package sun.nio.ch;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.net.*;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.security.AccessController;
import java.security.AccessControlContext;
import java.security.PrivilegedExceptionAction;
import java.security.PrivilegedActionException;
/**
* A prototype implementation of AsynchronousDatagramChannel, used to aid
* test and spec development.
*/
class SimpleAsynchronousDatagramChannelImpl
extends AsynchronousDatagramChannel implements Groupable, Cancellable
{
private final DatagramChannel dc;
private final AsynchronousChannelGroupImpl group;
private final Object attachKey;
private boolean closed;
// used to coordinate timed and blocking reads
private final Object readLock = new Object();
// channel blocking mode (requires readLock)
private boolean isBlocking = true;
// number of blocking readers (requires readLock)
private int blockingReaderCount;
// true if timed read attempted while blocking read in progress (requires readLock)
private boolean transitionToNonBlocking;
// true if a blocking read is cancelled (requires readLock)
private boolean blockingReadKilledByCancel;
// temporary Selectors used by timed reads (requires readLock)
private Selector firstReader;
private Set<Selector> otherReaders;
SimpleAsynchronousDatagramChannelImpl(ProtocolFamily family,
AsynchronousChannelGroupImpl group)
throws IOException
{
super(group.provider());
this.dc = (family == null) ?
DatagramChannel.open() : DatagramChannel.open(family);
this.group = group;
// attach this channel to the group as foreign channel
boolean registered = false;
try {
if (!(dc instanceof DatagramChannelImpl))
throw new UnsupportedOperationException();
attachKey = group
.attachForeignChannel(this, ((DatagramChannelImpl)dc).getFD());
registered = true;
} finally {
if (!registered)
dc.close();
}
}
// throws RuntimeException if blocking read has been cancelled
private void ensureBlockingReadNotKilled() {
assert Thread.holdsLock(readLock);
if (blockingReadKilledByCancel)
throw new RuntimeException("Reading not allowed due to cancellation");
}
// invoke prior to non-timed read/receive
private void beginNoTimeoutRead() {
synchronized (readLock) {
ensureBlockingReadNotKilled();
if (isBlocking)
blockingReaderCount++;
}
}
// invoke after non-timed read/receive has completed
private void endNoTimeoutRead() {
synchronized (readLock) {
if (isBlocking) {
if (--blockingReaderCount == 0 && transitionToNonBlocking) {
// notify any threads waiting to make channel non-blocking
readLock.notifyAll();
}
}
}
}
// invoke prior to timed read
// returns the timeout remaining
private long prepareForTimedRead(PendingFuture<?,?> result, long timeout)
throws IOException
{
synchronized (readLock) {
ensureBlockingReadNotKilled();
if (isBlocking) {
transitionToNonBlocking = true;
while (blockingReaderCount > 0 &&
timeout > 0L &&
!result.isCancelled())
{
long st = System.currentTimeMillis();
try {
readLock.wait(timeout);
} catch (InterruptedException e) { }
timeout -= System.currentTimeMillis() - st;
}
if (blockingReaderCount == 0) {
// re-check that blocked read wasn't cancelled
ensureBlockingReadNotKilled();
// no blocking reads so change channel to non-blocking
dc.configureBlocking(false);
isBlocking = false;
}
}
return timeout;
}
}
// returns a temporary Selector
private Selector getSelector() throws IOException {
Selector sel = Util.getTemporarySelector(dc);
synchronized (readLock) {
if (firstReader == null) {
firstReader = sel;
} else {
if (otherReaders == null)
otherReaders = new HashSet<Selector>();
otherReaders.add(sel);
}
}
return sel;
}
// releases a temporary Selector
private void releaseSelector(Selector sel) throws IOException {
synchronized (readLock) {
if (firstReader == sel) {
firstReader = null;
} else {
otherReaders.remove(sel);
}
}
Util.releaseTemporarySelector(sel);
}
// wakeup all Selectors currently in use
private void wakeupSelectors() {
synchronized (readLock) {
if (firstReader != null)
firstReader.wakeup();
if (otherReaders != null) {
for (Selector sel: otherReaders) {
sel.wakeup();
}
}
}
}
@Override
public AsynchronousChannelGroupImpl group() {
return group;
}
@Override
public boolean isOpen() {
return dc.isOpen();
}
@Override
public void onCancel(PendingFuture<?,?> task) {
synchronized (readLock) {
if (blockingReaderCount > 0) {
blockingReadKilledByCancel = true;
readLock.notifyAll();
return;
}
}
wakeupSelectors();
}
@Override
public void close() throws IOException {
synchronized (dc) {
if (closed)
return;
closed = true;
}
// detach from group and close underlying channel
group.detachForeignChannel(attachKey);
dc.close();
// wakeup any threads blocked in timed read/receives
wakeupSelectors();
}
@Override
public AsynchronousDatagramChannel connect(SocketAddress remote)
throws IOException
{
dc.connect(remote);
return this;
}
@Override
public AsynchronousDatagramChannel disconnect() throws IOException {
dc.disconnect();
return this;
}
private static class WrappedMembershipKey extends MembershipKey {
private final MulticastChannel channel;
private final MembershipKey key;
WrappedMembershipKey(MulticastChannel channel, MembershipKey key) {
this.channel = channel;
this.key = key;
}
@Override
public boolean isValid() {
return key.isValid();
}
@Override
public void drop() {
key.drop();
}
@Override
public MulticastChannel channel() {
return channel;
}
@Override
public InetAddress group() {
return key.group();
}
@Override
public NetworkInterface networkInterface() {
return key.networkInterface();
}
@Override
public InetAddress sourceAddress() {
return key.sourceAddress();
}
@Override
public MembershipKey block(InetAddress toBlock) throws IOException {
key.block(toBlock);
return this;
}
@Override
public MembershipKey unblock(InetAddress toUnblock) {
key.unblock(toUnblock);
return this;
}
@Override
public String toString() {
return key.toString();
}
}
@Override
public MembershipKey join(InetAddress group,
NetworkInterface interf)
throws IOException
{
MembershipKey key = ((MulticastChannel)dc).join(group, interf);
return new WrappedMembershipKey(this, key);
}
@Override
public MembershipKey join(InetAddress group,
NetworkInterface interf,
InetAddress source)
throws IOException
{
MembershipKey key = ((MulticastChannel)dc).join(group, interf, source);
return new WrappedMembershipKey(this, key);
}
private <A> Future<Integer> implSend(ByteBuffer src,
SocketAddress target,
A attachment,
CompletionHandler<Integer,? super A> handler)
{
int n = 0;
Throwable exc = null;
try {
n = dc.send(src, target);
} catch (IOException ioe) {
exc = ioe;
}
if (handler == null)
return CompletedFuture.withResult(n, exc);
Invoker.invoke(this, handler, attachment, n, exc);
return null;
}
@Override
public Future<Integer> send(ByteBuffer src, SocketAddress target) {
return implSend(src, target, null, null);
}
@Override
public <A> void send(ByteBuffer src,
SocketAddress target,
A attachment,
CompletionHandler<Integer,? super A> handler)
{
if (handler == null)
throw new NullPointerException("'handler' is null");
implSend(src, target, attachment, handler);
}
private <A> Future<Integer> implWrite(ByteBuffer src,
A attachment,
CompletionHandler<Integer,? super A> handler)
{
int n = 0;
Throwable exc = null;
try {
n = dc.write(src);
} catch (IOException ioe) {
exc = ioe;
}
if (handler == null)
return CompletedFuture.withResult(n, exc);
Invoker.invoke(this, handler, attachment, n, exc);
return null;
}
@Override
public Future<Integer> write(ByteBuffer src) {
return implWrite(src, null, null);
}
@Override
public <A> void write(ByteBuffer src,
A attachment,
CompletionHandler<Integer,? super A> handler)
{
if (handler == null)
throw new NullPointerException("'handler' is null");
implWrite(src, attachment, handler);
}
/**
* Receive into the given buffer with privileges enabled and restricted by
* the given AccessControlContext (can be null).
*/
private SocketAddress doRestrictedReceive(final ByteBuffer dst,
AccessControlContext acc)
throws IOException
{
if (acc == null) {
return dc.receive(dst);
} else {
try {
return AccessController.doPrivileged(
new PrivilegedExceptionAction<SocketAddress>() {
public SocketAddress run() throws IOException {
return dc.receive(dst);
}}, acc);
} catch (PrivilegedActionException pae) {
Exception cause = pae.getException();
if (cause instanceof SecurityException)
throw (SecurityException)cause;
throw (IOException)cause;
}
}
}
private <A> Future<SocketAddress> implReceive(final ByteBuffer dst,
final long timeout,
final TimeUnit unit,
A attachment,
final CompletionHandler<SocketAddress,? super A> handler)
{
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (timeout < 0L)
throw new IllegalArgumentException("Negative timeout");
if (unit == null)
throw new NullPointerException();
// complete immediately if channel closed
if (!isOpen()) {
Throwable exc = new ClosedChannelException();
if (handler == null)
return CompletedFuture.withFailure(exc);
Invoker.invoke(this, handler, attachment, null, exc);
return null;
}
final AccessControlContext acc = (System.getSecurityManager() == null) ?
null : AccessController.getContext();
final PendingFuture<SocketAddress,A> result =
new PendingFuture<SocketAddress,A>(this, handler, attachment);
Runnable task = new Runnable() {
public void run() {
try {
SocketAddress remote = null;
long to;
if (timeout == 0L) {
beginNoTimeoutRead();
try {
remote = doRestrictedReceive(dst, acc);
} finally {
endNoTimeoutRead();
}
to = 0L;
} else {
to = prepareForTimedRead(result, unit.toMillis(timeout));
if (to <= 0L)
throw new InterruptedByTimeoutException();
remote = doRestrictedReceive(dst, acc);
}
if (remote == null) {
Selector sel = getSelector();
SelectionKey sk = null;
try {
sk = dc.register(sel, SelectionKey.OP_READ);
for (;;) {
if (!dc.isOpen())
throw new AsynchronousCloseException();
if (result.isCancelled())
break;
long st = System.currentTimeMillis();
int ns = sel.select(to);
if (ns > 0) {
remote = doRestrictedReceive(dst, acc);
if (remote != null)
break;
}
sel.selectedKeys().remove(sk);
if (timeout != 0L) {
to -= System.currentTimeMillis() - st;
if (to <= 0)
throw new InterruptedByTimeoutException();
}
}
} finally {
if (sk != null)
sk.cancel();
releaseSelector(sel);
}
}
result.setResult(remote);
} catch (Throwable x) {
if (x instanceof ClosedChannelException)
x = new AsynchronousCloseException();
result.setFailure(x);
}
Invoker.invokeUnchecked(result);
}
};
try {
group.executeOnPooledThread(task);
} catch (RejectedExecutionException ree) {
throw new ShutdownChannelGroupException();
}
return result;
}
@Override
public Future<SocketAddress> receive(ByteBuffer dst) {
return implReceive(dst, 0L, TimeUnit.MILLISECONDS, null, null);
}
@Override
public <A> void receive(ByteBuffer dst,
long timeout,
TimeUnit unit,
A attachment,
CompletionHandler<SocketAddress,? super A> handler)
{
if (handler == null)
throw new NullPointerException("'handler' is null");
implReceive(dst, timeout, unit, attachment, handler);
}
private <A> Future<Integer> implRead(final ByteBuffer dst,
final long timeout,
final TimeUnit unit,
A attachment,
final CompletionHandler<Integer,? super A> handler)
{
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (timeout < 0L)
throw new IllegalArgumentException("Negative timeout");
if (unit == null)
throw new NullPointerException();
// complete immediately if channel closed
if (!isOpen()) {
Throwable exc = new ClosedChannelException();
if (handler == null)
return CompletedFuture.withFailure(exc);
Invoker.invoke(this, handler, attachment, null, exc);
return null;
}
// another thread may disconnect before read is initiated
if (!dc.isConnected())
throw new NotYetConnectedException();
final PendingFuture<Integer,A> result =
new PendingFuture<Integer,A>(this, handler, attachment);
Runnable task = new Runnable() {
public void run() {
try {
int n = 0;
long to;
if (timeout == 0L) {
beginNoTimeoutRead();
try {
n = dc.read(dst);
} finally {
endNoTimeoutRead();
}
to = 0L;
} else {
to = prepareForTimedRead(result, unit.toMillis(timeout));
if (to <= 0L)
throw new InterruptedByTimeoutException();
n = dc.read(dst);
}
if (n == 0) {
Selector sel = getSelector();
SelectionKey sk = null;
try {
sk = dc.register(sel, SelectionKey.OP_READ);
for (;;) {
if (!dc.isOpen())
throw new AsynchronousCloseException();
if (result.isCancelled())
break;
long st = System.currentTimeMillis();
int ns = sel.select(to);
if (ns > 0) {
if ((n = dc.read(dst)) != 0)
break;
}
sel.selectedKeys().remove(sk);
if (timeout != 0L) {
to -= System.currentTimeMillis() - st;
if (to <= 0)
throw new InterruptedByTimeoutException();
}
}
} finally {
if (sk != null)
sk.cancel();
releaseSelector(sel);
}
}
result.setResult(n);
} catch (Throwable x) {
if (x instanceof ClosedChannelException)
x = new AsynchronousCloseException();
result.setFailure(x);
}
Invoker.invokeUnchecked(result);
}
};
try {
group.executeOnPooledThread(task);
} catch (RejectedExecutionException ree) {
throw new ShutdownChannelGroupException();
}
return result;
}
@Override
public Future<Integer> read(ByteBuffer dst) {
return implRead(dst, 0L, TimeUnit.MILLISECONDS, null, null);
}
@Override
public <A> void read(ByteBuffer dst,
long timeout,
TimeUnit unit,
A attachment,
CompletionHandler<Integer,? super A> handler)
{
if (handler == null)
throw new NullPointerException("'handler' is null");
implRead(dst, timeout, unit, attachment, handler);
}
@Override
public AsynchronousDatagramChannel bind(SocketAddress local)
throws IOException
{
dc.bind(local);
return this;
}
@Override
public SocketAddress getLocalAddress() throws IOException {
return dc.getLocalAddress();
}
@Override
public <T> AsynchronousDatagramChannel setOption(SocketOption<T> name, T value)
throws IOException
{
dc.setOption(name, value);
return this;
}
@Override
public <T> T getOption(SocketOption<T> name) throws IOException {
return dc.getOption(name);
}
@Override
public Set<SocketOption<?>> supportedOptions() {
return dc.supportedOptions();
}
@Override
public SocketAddress getRemoteAddress() throws IOException {
return dc.getRemoteAddress();
}
}

9
jdk/src/solaris/classes/sun/nio/ch/LinuxAsynchronousChannelProvider.java
View file

@ -29,7 +29,6 @@ import java.nio.channels.*;
import java.nio.channels.spi.AsynchronousChannelProvider;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.net.ProtocolFamily;
import java.io.IOException;
public class LinuxAsynchronousChannelProvider
@ -88,12 +87,4 @@ public class LinuxAsynchronousChannelProvider
{
return new UnixAsynchronousSocketChannelImpl(toPort(group));
}
@Override
public AsynchronousDatagramChannel openAsynchronousDatagramChannel(ProtocolFamily family,
AsynchronousChannelGroup group)
throws IOException
{
return new SimpleAsynchronousDatagramChannelImpl(family, toPort(group));
}
}

9
jdk/src/solaris/classes/sun/nio/ch/SolarisAsynchronousChannelProvider.java
View file

@ -29,7 +29,6 @@ import java.nio.channels.*;
import java.nio.channels.spi.AsynchronousChannelProvider;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.net.ProtocolFamily;
import java.io.IOException;
public class SolarisAsynchronousChannelProvider
@ -91,12 +90,4 @@ public class SolarisAsynchronousChannelProvider
{
return new UnixAsynchronousSocketChannelImpl(toEventPort(group));
}
@Override
public AsynchronousDatagramChannel openAsynchronousDatagramChannel(ProtocolFamily family,
AsynchronousChannelGroup group)
throws IOException
{
return new SimpleAsynchronousDatagramChannelImpl(family, toEventPort(group));
}
}

9
jdk/src/windows/classes/sun/nio/ch/WindowsAsynchronousChannelProvider.java
View file

@ -29,7 +29,6 @@ import java.nio.channels.*;
import java.nio.channels.spi.AsynchronousChannelProvider;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.net.ProtocolFamily;
import java.io.IOException;
public class WindowsAsynchronousChannelProvider
@ -90,12 +89,4 @@ public class WindowsAsynchronousChannelProvider
{
return new WindowsAsynchronousSocketChannelImpl(toIocp(group));
}
@Override
public AsynchronousDatagramChannel openAsynchronousDatagramChannel(ProtocolFamily family,
AsynchronousChannelGroup group)
throws IOException
{
return new SimpleAsynchronousDatagramChannelImpl(family, toIocp(group));
}
}

18
jdk/test/java/nio/channels/AsynchronousChannelGroup/Basic.java
View file

@ -89,10 +89,9 @@ public class Basic {
}
// create channel that is bound to group
AsynchronousChannel ch;
switch (rand.nextInt(3)) {
switch (rand.nextInt(2)) {
case 0 : ch = AsynchronousSocketChannel.open(group); break;
case 1 : ch = AsynchronousServerSocketChannel.open(group); break;
case 2 : ch = AsynchronousDatagramChannel.open(null, group); break;
default : throw new AssertionError();
}
group.shutdown();
@ -128,18 +127,9 @@ public class Basic {
}
// I/O in progress
AsynchronousChannel ch;
if (rand.nextBoolean()) {
AsynchronousServerSocketChannel listener = AsynchronousServerSocketChannel
.open(group).bind(new InetSocketAddress(0));
listener.accept();
ch = listener;
} else {
AsynchronousDatagramChannel adc =
AsynchronousDatagramChannel.open(null, group);
adc.receive(ByteBuffer.allocate(100));
ch = adc;
}
AsynchronousServerSocketChannel ch = AsynchronousServerSocketChannel
.open(group).bind(new InetSocketAddress(0));
ch.accept();
// forceful shutdown
group.shutdownNow();

410
jdk/test/java/nio/channels/AsynchronousDatagramChannel/Basic.java
View file

@ -1,410 +0,0 @@
/*
* Copyright (c) 2008, 2009, Oracle and/or its affiliates. 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 Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/* @test
* @bug 4527345 6842687
* @summary Unit test for AsynchronousDatagramChannel
*/
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.net.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
public class Basic {
public static void main(String[] args) throws Exception {
doReceiveTests();
doReadTests();
doSendTests();
doWriteTests();
doCancelTests();
doMulticastTests();
}
// basic receive tests
static void doReceiveTests() throws Exception {
final byte[] msg = "hello".getBytes();
AsynchronousDatagramChannel ch = AsynchronousDatagramChannel.open()
.bind(new InetSocketAddress(0));
int port = ((InetSocketAddress)(ch.getLocalAddress())).getPort();
InetAddress rh = InetAddress.getLocalHost();
final SocketAddress sa = new InetSocketAddress(rh, port);
DatagramChannel sender = DatagramChannel.open();
ByteBuffer dst = ByteBuffer.allocateDirect(100);
// Test: datagram packet received immediately
sender.send(ByteBuffer.wrap(msg), sa);
dst.clear();
ch.receive(dst).get(1, TimeUnit.SECONDS);
if (dst.flip().remaining() != msg.length)
throw new RuntimeException("Unexpected number of bytes read");
// Test: datagram packet not received immediately
dst.clear();
final CountDownLatch latch = new CountDownLatch(1);
ch.receive(dst, (Void)null, new CompletionHandler<SocketAddress,Void>() {
public void completed(SocketAddress source, Void att) {
latch.countDown();
}
public void failed (Throwable exc, Void att) {
}
});
Thread.sleep(2000);
sender.send(ByteBuffer.wrap(msg), sa);
latch.await(2, TimeUnit.SECONDS); // wait for completion handler
// Test: timeout
dst.clear();
final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
ch.receive(dst, 2, TimeUnit.SECONDS, (Void)null, new CompletionHandler<SocketAddress,Void>() {
public void completed(SocketAddress source, Void att) {
}
public void failed (Throwable exc, Void att) {
exception.set(exc);
}
});
Throwable result;
while ((result = exception.get()) == null) {
Thread.sleep(100);
}
if (!(result instanceof InterruptedByTimeoutException))
throw new RuntimeException("InterruptedByTimeoutException expected");
// AsynchronousCloseException
dst = ByteBuffer.allocateDirect(100);
exception.set(null);
ch.receive(dst, (Void)null, new CompletionHandler<SocketAddress,Void>() {
public void completed(SocketAddress source, Void att) {
}
public void failed (Throwable exc, Void att) {
exception.set(exc);
}
});
ch.close();
while ((result = exception.get()) == null) {
Thread.sleep(100);
}
if (!(result instanceof AsynchronousCloseException))
throw new RuntimeException("AsynchronousCloseException expected");
// done
sender.close();
}
// basic read tests
static void doReadTests() throws Exception {
final byte[] msg = "hello".getBytes();
AsynchronousDatagramChannel ch = AsynchronousDatagramChannel.open()
.bind(new InetSocketAddress(0));
int port = ((InetSocketAddress)(ch.getLocalAddress())).getPort();
InetAddress lh = InetAddress.getLocalHost();
final SocketAddress sa = new InetSocketAddress(lh, port);
DatagramChannel sender = DatagramChannel.open();
ByteBuffer dst = ByteBuffer.allocateDirect(100);
// Test: not connected
try {
ch.read(dst);
throw new RuntimeException("NotYetConnectedException expected");
} catch (NotYetConnectedException e) {
}
// connect the channel
sender.bind(new InetSocketAddress(0));
ch.connect(new InetSocketAddress(lh,
((InetSocketAddress)(sender.getLocalAddress())).getPort()));
// Test: datagram packet received immediately
sender.send(ByteBuffer.wrap(msg), sa);
dst.clear();
ch.read(dst).get(1, TimeUnit.SECONDS);
if (dst.flip().remaining() != msg.length)
throw new RuntimeException("Unexpected number of bytes read");
// Test: datagram packet not received immediately
dst.clear();
final CountDownLatch l1 = new CountDownLatch(1);
ch.read(dst, (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesRead, Void att) {
l1.countDown();
}
public void failed (Throwable exc, Void att) {
}
});
Thread.sleep(2000);
sender.send(ByteBuffer.wrap(msg), sa);
l1.await(2, TimeUnit.SECONDS);
// Test: timeout
dst.clear();
final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
ch.read(dst, 2, TimeUnit.SECONDS, (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesRead, Void att) {
}
public void failed (Throwable exc, Void att) {
exception.set(exc);
}
});
Throwable result;
while ((result = exception.get()) == null) {
Thread.sleep(100);
}
if (!(result instanceof InterruptedByTimeoutException))
throw new RuntimeException("InterruptedByTimeoutException expected");
// AsynchronousCloseException
dst.clear();
exception.set(null);
ch.read(dst, (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesRead, Void att) {
}
public void failed (Throwable exc, Void att) {
exception.set(exc);
}
});
ch.close();
while ((result = exception.get()) == null) {
Thread.sleep(100);
}
if (!(result instanceof AsynchronousCloseException))
throw new RuntimeException("AsynchronousCloseException expected");
// done
sender.close();
}
// basic send tests
static void doSendTests() throws Exception {
final byte[] msg = "hello".getBytes();
DatagramChannel reader = DatagramChannel.open()
.bind(new InetSocketAddress(0));
int port = ((InetSocketAddress)(reader.getLocalAddress())).getPort();
InetAddress rh = InetAddress.getLocalHost();
SocketAddress sa = new InetSocketAddress(rh, port);
AsynchronousDatagramChannel ch = AsynchronousDatagramChannel.open();
// Test: send datagram packet to reader
int bytesSent = ch.send(ByteBuffer.wrap(msg), sa).get();
if (bytesSent != msg.length)
throw new RuntimeException("Unexpected number of bytes sent");
// check received
ByteBuffer dst = ByteBuffer.allocateDirect(100);
reader.receive(dst);
dst.flip();
if (dst.remaining() != msg.length)
throw new RuntimeException("Unexpected number of bytes received");
// Test: send datagram packet to reader and check completion handler
// is invoked
final CountDownLatch l2 = new CountDownLatch(1);
ch.send(ByteBuffer.wrap(msg), sa, (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesSent, Void att) {
if (bytesSent != msg.length)
throw new RuntimeException("Unexpected number of bytes received");
l2.countDown();
}
public void failed (Throwable exc, Void att) {
}
});
l2.await(5, TimeUnit.SECONDS);
// check received
dst.clear();
reader.receive(dst);
dst.flip();
if (dst.remaining() != msg.length)
throw new RuntimeException("Unexpected number of bytes received");
// Test: check that failed method is invoked
ch.close();
final CountDownLatch l3 = new CountDownLatch(1);
ch.send(ByteBuffer.wrap(msg), sa, (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesSent, Void att) {
throw new RuntimeException("completed method invoked");
}
public void failed (Throwable exc, Void att) {
if (exc instanceof ClosedChannelException) {
l3.countDown();
} else {
throw new RuntimeException(exc);
}
}
});
l3.await(5, TimeUnit.SECONDS);
// done
reader.close();
}
// basic write tests
static void doWriteTests() throws Exception {
final byte[] msg = "hello".getBytes();
DatagramChannel reader = DatagramChannel.open()
.bind(new InetSocketAddress(0));
int port = ((InetSocketAddress)(reader.getLocalAddress())).getPort();
InetAddress rh = InetAddress.getLocalHost();
SocketAddress sa = new InetSocketAddress(rh, port);
AsynchronousDatagramChannel ch = AsynchronousDatagramChannel.open();
// Test: unconnected
try {
ch.write(ByteBuffer.wrap(msg)).get();
throw new RuntimeException("NotYetConnectedException expected");
} catch (NotYetConnectedException e) {
}
// Test: connect, and write datagram
ch.connect(sa);
int bytesSent = ch.write(ByteBuffer.wrap(msg)).get();
if (bytesSent != msg.length)
throw new RuntimeException("Unexpected number of bytes sent");
// check received
ByteBuffer dst = ByteBuffer.allocateDirect(100);
reader.receive(dst);
dst.flip();
if (dst.remaining() != msg.length)
throw new RuntimeException("Unexpected number of bytes received");
// Test: write datagram and check completion handler is invoked
final CountDownLatch l2 = new CountDownLatch(1);
ch.write(ByteBuffer.wrap(msg), (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesSent, Void att) {
if (bytesSent != msg.length)
throw new RuntimeException("Unexpected number of bytes received");
l2.countDown();
}
public void failed (Throwable exc, Void att) {
}
});
l2.await(5, TimeUnit.SECONDS);
// check received
dst.clear();
reader.receive(dst);
dst.flip();
if (dst.remaining() != msg.length)
throw new RuntimeException("Unexpected number of bytes received");
// done
ch.close();
reader.close();
}
static void cancelAndCheck(Future<?> result)
throws InterruptedException
{
boolean cancelled = result.cancel(false);
if (!cancelled)
throw new RuntimeException("Not cancelled");
if (!result.isDone())
throw new RuntimeException("Should be done");
try {
result.get();
throw new RuntimeException("Result not expected");
} catch (CancellationException e) {
// expected
} catch (ExecutionException e) {
throw new RuntimeException("Should not fail");
}
}
// basic cancel tests
static void doCancelTests() throws Exception {
InetAddress lh = InetAddress.getLocalHost();
// receive
for (int i=0; i<2; i++) {
AsynchronousDatagramChannel ch =
AsynchronousDatagramChannel.open().bind(new InetSocketAddress(0));
Future<SocketAddress> remote = ch.receive(ByteBuffer.allocate(100));
cancelAndCheck(remote);
ch.close();
}
// read
for (int i=0; i<2; i++) {
AsynchronousDatagramChannel ch =
AsynchronousDatagramChannel.open().bind(new InetSocketAddress(0));
ch.connect(new InetSocketAddress(lh,
((InetSocketAddress)(ch.getLocalAddress())).getPort()));
final CountDownLatch latch = new CountDownLatch(1);
long timeout = (i == 0) ? 0L : 60L;
Future<Integer> result = ch.read(ByteBuffer.allocate(100));
cancelAndCheck(result);
ch.close();
}
}
// basic multicast test
static void doMulticastTests() throws Exception {
final byte[] msg = "hello".getBytes();
InetAddress lh = InetAddress.getLocalHost();
NetworkInterface interf = NetworkInterface.getByInetAddress(lh);
if (interf.isLoopback() || !interf.supportsMulticast()) {
System.out.println("Multicasting not tested");
return;
}
AsynchronousDatagramChannel ch = AsynchronousDatagramChannel
.open(StandardProtocolFamily.INET, null)
.setOption(StandardSocketOption.SO_REUSEADDR, true)
.bind(new InetSocketAddress(0));
int port = ((InetSocketAddress)(ch.getLocalAddress())).getPort();
// join group
InetAddress group = InetAddress.getByName("225.4.5.6");
MembershipKey key = ch.join(group, interf);
// check key
if (key.channel() != ch)
throw new RuntimeException("Not the expected channel");
// send message to group
DatagramChannel sender = DatagramChannel.open();
sender.send(ByteBuffer.wrap(msg), new InetSocketAddress(group, port));
sender.close();
// check message received
ByteBuffer dst = ByteBuffer.allocate(200);
SocketAddress source = ch.receive(dst).get(2, TimeUnit.SECONDS);
if (!((InetSocketAddress)source).getAddress().equals(lh))
throw new RuntimeException("Unexpected source");
// done
ch.close();
}
}

8
jdk/test/java/nio/channels/spi/AsynchronousChannelProvider/Provider1.java
View file

@ -23,7 +23,6 @@
import java.nio.channels.spi.AsynchronousChannelProvider;
import java.nio.channels.*;
import java.net.ProtocolFamily;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.io.IOException;
@ -59,11 +58,4 @@ public class Provider1 extends AsynchronousChannelProvider {
{
throw new RuntimeException();
}
@Override
public AsynchronousDatagramChannel openAsynchronousDatagramChannel
(ProtocolFamily family, AsynchronousChannelGroup group) throws IOException
{
throw new RuntimeException();
}
}

8
jdk/test/java/nio/channels/spi/AsynchronousChannelProvider/Provider2.java
View file

@ -23,7 +23,6 @@
import java.nio.channels.spi.AsynchronousChannelProvider;
import java.nio.channels.*;
import java.net.ProtocolFamily;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.io.IOException;
@ -59,11 +58,4 @@ public class Provider2 extends AsynchronousChannelProvider {
{
throw new RuntimeException();
}
@Override
public AsynchronousDatagramChannel openAsynchronousDatagramChannel
(ProtocolFamily family, AsynchronousChannelGroup group) throws IOException
{
throw new RuntimeException();
}
}