/*
* Copyright (c) 2009, 2010, 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.net.InetAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.InetSocketAddress;
import java.io.FileDescriptor;
import java.io.IOException;
import java.util.Collections;
import java.util.Set;
import java.util.HashSet;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ConnectionPendingException;
import java.nio.channels.NoConnectionPendingException;
import java.nio.channels.AlreadyConnectedException;
import java.nio.channels.NotYetBoundException;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.spi.SelectorProvider;
import com.sun.nio.sctp.AbstractNotificationHandler;
import com.sun.nio.sctp.Association;
import com.sun.nio.sctp.AssociationChangeNotification;
import com.sun.nio.sctp.HandlerResult;
import com.sun.nio.sctp.IllegalReceiveException;
import com.sun.nio.sctp.InvalidStreamException;
import com.sun.nio.sctp.IllegalUnbindException;
import com.sun.nio.sctp.MessageInfo;
import com.sun.nio.sctp.NotificationHandler;
import com.sun.nio.sctp.SctpChannel;
import com.sun.nio.sctp.SctpSocketOption;
import sun.nio.ch.PollArrayWrapper;
import sun.nio.ch.SelChImpl;
import static com.sun.nio.sctp.SctpStandardSocketOptions.*;
import static sun.nio.ch.SctpResultContainer.SEND_FAILED;
import static sun.nio.ch.SctpResultContainer.ASSOCIATION_CHANGED;
import static sun.nio.ch.SctpResultContainer.PEER_ADDRESS_CHANGED;
import static sun.nio.ch.SctpResultContainer.SHUTDOWN;
/**
* An implementation of an SctpChannel
*/
public class SctpChannelImpl extends SctpChannel
implements SelChImpl
{
private final FileDescriptor fd;
private final int fdVal;
/* IDs of native threads doing send and receivess, for signalling */
private volatile long receiverThread = 0;
private volatile long senderThread = 0;
/* Lock held by current receiving or connecting thread */
private final Object receiveLock = new Object();
/* Lock held by current sending or connecting thread */
private final Object sendLock = new Object();
private final ThreadLocal<Boolean> receiveInvoked =
new ThreadLocal<Boolean>() {
@Override protected Boolean initialValue() {
return Boolean.FALSE;
}
};
/* Lock held by any thread that modifies the state fields declared below
DO NOT invoke a blocking I/O operation while holding this lock! */
private final Object stateLock = new Object();
private enum ChannelState {
UNINITIALIZED,
UNCONNECTED,
PENDING,
CONNECTED,
KILLPENDING,
KILLED,
}
/* -- The following fields are protected by stateLock -- */
private ChannelState state = ChannelState.UNINITIALIZED;
/* Binding; Once bound the port will remain constant. */
int port = -1;
private HashSet<InetSocketAddress> localAddresses = new HashSet<InetSocketAddress>();
/* Has the channel been bound to the wildcard address */
private boolean wildcard; /* false */
//private InetSocketAddress remoteAddress = null;
/* Input/Output open */
private boolean readyToConnect;
/* Shutdown */
private boolean isShutdown;
private Association association;
private Set<SocketAddress> remoteAddresses = Collections.EMPTY_SET;
/* -- End of fields protected by stateLock -- */
/**
* Constructor for normal connecting sockets
*/
public SctpChannelImpl(SelectorProvider provider) throws IOException {
//TODO: update provider remove public modifier
super(provider);
this.fd = SctpNet.socket(true);
this.fdVal = IOUtil.fdVal(fd);
this.state = ChannelState.UNCONNECTED;
}
/**
* Constructor for sockets obtained from server sockets
*/
public SctpChannelImpl(SelectorProvider provider, FileDescriptor fd)
throws IOException {
this(provider, fd, null);
}
/**
* Constructor for sockets obtained from branching
*/
public SctpChannelImpl(SelectorProvider provider,
FileDescriptor fd,
Association association)
throws IOException {
super(provider);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
this.state = ChannelState.CONNECTED;
port = (Net.localAddress(fd)).getPort();
if (association != null) { /* branched */
this.association = association;
} else { /* obtained from server channel */
/* Receive COMM_UP */
ByteBuffer buf = Util.getTemporaryDirectBuffer(50);
try {
receive(buf, null, null, true);
} finally {
Util.releaseTemporaryDirectBuffer(buf);
}
}
}
/**
* Binds the channel's socket to a local address.
*/
@Override
public SctpChannel bind(SocketAddress local) throws IOException {
synchronized (receiveLock) {
synchronized (sendLock) {
synchronized (stateLock) {
ensureOpenAndUnconnected();
if (isBound())
SctpNet.throwAlreadyBoundException();
InetSocketAddress isa = (local == null) ?
new InetSocketAddress(0) : Net.checkAddress(local);
Net.bind(fd, isa.getAddress(), isa.getPort());
InetSocketAddress boundIsa = Net.localAddress(fd);
port = boundIsa.getPort();
localAddresses.add(isa);
if (isa.getAddress().isAnyLocalAddress())
wildcard = true;
}
}
}
return this;
}
@Override
public SctpChannel bindAddress(InetAddress address)
throws IOException {
bindUnbindAddress(address, true);
localAddresses.add(new InetSocketAddress(address, port));
return this;
}
@Override
public SctpChannel unbindAddress(InetAddress address)
throws IOException {
bindUnbindAddress(address, false);
localAddresses.remove(new InetSocketAddress(address, port));
return this;
}
private SctpChannel bindUnbindAddress(InetAddress address, boolean add)
throws IOException {
if (address == null)
throw new IllegalArgumentException();
synchronized (receiveLock) {
synchronized (sendLock) {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (!isBound())
throw new NotYetBoundException();
if (wildcard)
throw new IllegalStateException(
"Cannot add or remove addresses from a channel that is bound to the wildcard address");
if (address.isAnyLocalAddress())
throw new IllegalArgumentException(
"Cannot add or remove the wildcard address");
if (add) {
for (InetSocketAddress addr : localAddresses) {
if (addr.getAddress().equals(address)) {
SctpNet.throwAlreadyBoundException();
}
}
} else { /*removing */
/* Verify that there is more than one address
* and that address is already bound */
if (localAddresses.size() <= 1)
throw new IllegalUnbindException("Cannot remove address from a channel with only one address bound");
boolean foundAddress = false;
for (InetSocketAddress addr : localAddresses) {
if (addr.getAddress().equals(address)) {
foundAddress = true;
break;
}
}
if (!foundAddress )
throw new IllegalUnbindException("Cannot remove address from a channel that is not bound to that address");
}
SctpNet.bindx(fdVal, new InetAddress[]{address}, port, add);
/* Update our internal Set to reflect the addition/removal */
if (add)
localAddresses.add(new InetSocketAddress(address, port));
else {
for (InetSocketAddress addr : localAddresses) {
if (addr.getAddress().equals(address)) {
localAddresses.remove(addr);
break;
}
}
}
}
}
}
return this;
}
private boolean isBound() {
synchronized (stateLock) {
return port == -1 ? false : true;
}
}
private boolean isConnected() {
synchronized (stateLock) {
return (state == ChannelState.CONNECTED);
}
}
private void ensureOpenAndUnconnected() throws IOException {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (isConnected())
throw new AlreadyConnectedException();
if (state == ChannelState.PENDING)
throw new ConnectionPendingException();
}
}
private boolean ensureReceiveOpen() throws ClosedChannelException {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (!isConnected())
throw new NotYetConnectedException();
else
return true;
}
}
private void ensureSendOpen() throws ClosedChannelException {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (isShutdown)
throw new ClosedChannelException();
if (!isConnected())
throw new NotYetConnectedException();
}
}
private void receiverCleanup() throws IOException {
synchronized (stateLock) {
receiverThread = 0;
if (state == ChannelState.KILLPENDING)
kill();
}
}
private void senderCleanup() throws IOException {
synchronized (stateLock) {
senderThread = 0;
if (state == ChannelState.KILLPENDING)
kill();
}
}
@Override
public Association association() throws ClosedChannelException {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (!isConnected())
return null;
return association;
}
}
@Override
public boolean connect(SocketAddress endpoint) throws IOException {
synchronized (receiveLock) {
synchronized (sendLock) {
ensureOpenAndUnconnected();
InetSocketAddress isa = Net.checkAddress(endpoint);
SecurityManager sm = System.getSecurityManager();
if (sm != null)
sm.checkConnect(isa.getAddress().getHostAddress(),
isa.getPort());
synchronized (blockingLock()) {
int n = 0;
try {
try {
begin();
synchronized (stateLock) {
if (!isOpen()) {
return false;
}
receiverThread = NativeThread.current();
}
for (;;) {
InetAddress ia = isa.getAddress();
if (ia.isAnyLocalAddress())
ia = InetAddress.getLocalHost();
n = SctpNet.connect(fdVal, ia, isa.getPort());
if ( (n == IOStatus.INTERRUPTED)
&& isOpen())
continue;
break;
}
} finally {
receiverCleanup();
end((n > 0) || (n == IOStatus.UNAVAILABLE));
assert IOStatus.check(n);
}
} catch (IOException x) {
/* If an exception was thrown, close the channel after
* invoking end() so as to avoid bogus
* AsynchronousCloseExceptions */
close();
throw x;
}
if (n > 0) {
synchronized (stateLock) {
/* Connection succeeded */
state = ChannelState.CONNECTED;
if (!isBound()) {
InetSocketAddress boundIsa =
Net.localAddress(fd);
port = boundIsa.getPort();
}
/* Receive COMM_UP */
ByteBuffer buf = Util.getTemporaryDirectBuffer(50);
try {
receive(buf, null, null, true);
} finally {
Util.releaseTemporaryDirectBuffer(buf);
}
/* cache remote addresses */
try {
remoteAddresses = getRemoteAddresses();
} catch (IOException unused) { /* swallow exception */ }
return true;
}
} else {
synchronized (stateLock) {
/* If nonblocking and no exception then connection
* pending; disallow another invocation */
if (!isBlocking())
state = ChannelState.PENDING;
else
assert false;
}
}
}
return false;
}
}
}
@Override
public boolean connect(SocketAddress endpoint,
int maxOutStreams,
int maxInStreams)
throws IOException {
ensureOpenAndUnconnected();
return setOption(SCTP_INIT_MAXSTREAMS, InitMaxStreams.
create(maxInStreams, maxOutStreams)).connect(endpoint);
}
@Override
public boolean isConnectionPending() {
synchronized (stateLock) {
return (state == ChannelState.PENDING);
}
}
@Override
public boolean finishConnect() throws IOException {
synchronized (receiveLock) {
synchronized (sendLock) {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (isConnected())
return true;
if (state != ChannelState.PENDING)
throw new NoConnectionPendingException();
}
int n = 0;
try {
try {
begin();
synchronized (blockingLock()) {
synchronized (stateLock) {
if (!isOpen()) {
return false;
}
receiverThread = NativeThread.current();
}
if (!isBlocking()) {
for (;;) {
n = checkConnect(fd, false, readyToConnect);
if ( (n == IOStatus.INTERRUPTED)
&& isOpen())
continue;
break;
}
} else {
for (;;) {
n = checkConnect(fd, true, readyToConnect);
if (n == 0) {
// Loop in case of
// spurious notifications
continue;
}
if ( (n == IOStatus.INTERRUPTED)
&& isOpen())
continue;
break;
}
}
}
} finally {
synchronized (stateLock) {
receiverThread = 0;
if (state == ChannelState.KILLPENDING) {
kill();
/* poll()/getsockopt() does not report
* error (throws exception, with n = 0)
* on Linux platform after dup2 and
* signal-wakeup. Force n to 0 so the
* end() can throw appropriate exception */
n = 0;
}
}
end((n > 0) || (n == IOStatus.UNAVAILABLE));
assert IOStatus.check(n);
}
} catch (IOException x) {
/* If an exception was thrown, close the channel after
* invoking end() so as to avoid bogus
* AsynchronousCloseExceptions */
close();
throw x;
}
if (n > 0) {
synchronized (stateLock) {
state = ChannelState.CONNECTED;
if (!isBound()) {
InetSocketAddress boundIsa =
Net.localAddress(fd);
port = boundIsa.getPort();
}
/* Receive COMM_UP */
ByteBuffer buf = Util.getTemporaryDirectBuffer(50);
try {
receive(buf, null, null, true);
} finally {
Util.releaseTemporaryDirectBuffer(buf);
}
/* cache remote addresses */
try {
remoteAddresses = getRemoteAddresses();
} catch (IOException unused) { /* swallow exception */ }
return true;
}
}
}
}
return false;
}
@Override
protected void implConfigureBlocking(boolean block) throws IOException {
IOUtil.configureBlocking(fd, block);
}
@Override
public void implCloseSelectableChannel() throws IOException {
synchronized (stateLock) {
SctpNet.preClose(fdVal);
if (receiverThread != 0)
NativeThread.signal(receiverThread);
if (senderThread != 0)
NativeThread.signal(senderThread);
if (!isRegistered())
kill();
}
}
@Override
public FileDescriptor getFD() {
return fd;
}
@Override
public int getFDVal() {
return fdVal;
}
/**
* Translates native poll revent ops into a ready operation ops
*/
private boolean translateReadyOps(int ops, int initialOps, SelectionKeyImpl sk) {
int intOps = sk.nioInterestOps();
int oldOps = sk.nioReadyOps();
int newOps = initialOps;
if ((ops & PollArrayWrapper.POLLNVAL) != 0) {
/* This should only happen if this channel is pre-closed while a
* selection operation is in progress
* ## Throw an error if this channel has not been pre-closed */
return false;
}
if ((ops & (PollArrayWrapper.POLLERR
| PollArrayWrapper.POLLHUP)) != 0) {
newOps = intOps;
sk.nioReadyOps(newOps);
/* No need to poll again in checkConnect,
* the error will be detected there */
readyToConnect = true;
return (newOps & ~oldOps) != 0;
}
if (((ops & PollArrayWrapper.POLLIN) != 0) &&
((intOps & SelectionKey.OP_READ) != 0) &&
isConnected())
newOps |= SelectionKey.OP_READ;
if (((ops & PollArrayWrapper.POLLCONN) != 0) &&
((intOps & SelectionKey.OP_CONNECT) != 0) &&
((state == ChannelState.UNCONNECTED) || (state == ChannelState.PENDING))) {
newOps |= SelectionKey.OP_CONNECT;
readyToConnect = true;
}
if (((ops & PollArrayWrapper.POLLOUT) != 0) &&
((intOps & SelectionKey.OP_WRITE) != 0) &&
isConnected())
newOps |= SelectionKey.OP_WRITE;
sk.nioReadyOps(newOps);
return (newOps & ~oldOps) != 0;
}
@Override
public boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl sk) {
return translateReadyOps(ops, sk.nioReadyOps(), sk);
}
@Override
@SuppressWarnings("all")
public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl sk) {
return translateReadyOps(ops, 0, sk);
}
@Override
public void translateAndSetInterestOps(int ops, SelectionKeyImpl sk) {
int newOps = 0;
if ((ops & SelectionKey.OP_READ) != 0)
newOps |= PollArrayWrapper.POLLIN;
if ((ops & SelectionKey.OP_WRITE) != 0)
newOps |= PollArrayWrapper.POLLOUT;
if ((ops & SelectionKey.OP_CONNECT) != 0)
newOps |= PollArrayWrapper.POLLCONN;
sk.selector.putEventOps(sk, newOps);
}
@Override
public void kill() throws IOException {
synchronized (stateLock) {
if (state == ChannelState.KILLED)
return;
if (state == ChannelState.UNINITIALIZED) {
state = ChannelState.KILLED;
return;
}
assert !isOpen() && !isRegistered();
/* Postpone the kill if there is a waiting reader
* or writer thread. */
if (receiverThread == 0 && senderThread == 0) {
SctpNet.close(fdVal);
state = ChannelState.KILLED;
} else {
state = ChannelState.KILLPENDING;
}
}
}
@Override
public <T> SctpChannel setOption(SctpSocketOption<T> name, T value)
throws IOException {
if (name == null)
throw new NullPointerException();
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
SctpNet.setSocketOption(fdVal, name, value, 0 /*oneToOne*/);
}
return this;
}
@Override
@SuppressWarnings("unchecked")
public <T> T getOption(SctpSocketOption<T> name) throws IOException {
if (name == null)
throw new NullPointerException();
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
return (T)SctpNet.getSocketOption(fdVal, name, 0 /*oneToOne*/);
}
}
private static class DefaultOptionsHolder {
static final Set<SctpSocketOption<?>> defaultOptions = defaultOptions();
private static Set<SctpSocketOption<?>> defaultOptions() {
HashSet<SctpSocketOption<?>> set = new HashSet<SctpSocketOption<?>>(10);
set.add(SCTP_DISABLE_FRAGMENTS);
set.add(SCTP_EXPLICIT_COMPLETE);
set.add(SCTP_FRAGMENT_INTERLEAVE);
set.add(SCTP_INIT_MAXSTREAMS);
set.add(SCTP_NODELAY);
set.add(SCTP_PRIMARY_ADDR);
set.add(SCTP_SET_PEER_PRIMARY_ADDR);
set.add(SO_SNDBUF);
set.add(SO_RCVBUF);
set.add(SO_LINGER);
return Collections.unmodifiableSet(set);
}
}
@Override
public final Set<SctpSocketOption<?>> supportedOptions() {
return DefaultOptionsHolder.defaultOptions;
}
@Override
public <T> MessageInfo receive(ByteBuffer buffer,
T attachment,
NotificationHandler<T> handler)
throws IOException {
return receive(buffer, attachment, handler, false);
}
private <T> MessageInfo receive(ByteBuffer buffer,
T attachment,
NotificationHandler<T> handler,
boolean fromConnect)
throws IOException {
if (buffer == null)
throw new IllegalArgumentException("buffer cannot be null");
if (buffer.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (receiveInvoked.get())
throw new IllegalReceiveException(
"cannot invoke receive from handler");
receiveInvoked.set(Boolean.TRUE);
try {
SctpResultContainer resultContainer = new SctpResultContainer();
do {
resultContainer.clear();
synchronized (receiveLock) {
if (!ensureReceiveOpen())
return null;
int n = 0;
try {
begin();
synchronized (stateLock) {
if(!isOpen())
return null;
receiverThread = NativeThread.current();
}
do {
n = receive(fdVal, buffer, resultContainer, fromConnect);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
} finally {
receiverCleanup();
end((n > 0) || (n == IOStatus.UNAVAILABLE));
assert IOStatus.check(n);
}
if (!resultContainer.isNotification()) {
/* message or nothing */
if (resultContainer.hasSomething()) {
/* Set the association before returning */
SctpMessageInfoImpl info =
resultContainer.getMessageInfo();
synchronized (stateLock) {
assert association != null;
info.setAssociation(association);
}
return info;
} else
/* Non-blocking may return null if nothing available*/
return null;
} else { /* notification */
synchronized (stateLock) {
handleNotificationInternal(
resultContainer);
}
}
if (fromConnect) {
/* If we reach here, then it was connect that invoked
* receive and received the COMM_UP. We have already
* handled the COMM_UP with the internal notification
* handler. Simply return. */
return null;
}
} /* receiveLock */
} while (handler == null ? true :
(invokeNotificationHandler(resultContainer, handler, attachment)
== HandlerResult.CONTINUE));
return null;
} finally {
receiveInvoked.set(Boolean.FALSE);
}
}
private int receive(int fd,
ByteBuffer dst,
SctpResultContainer resultContainer,
boolean peek)
throws IOException {
int pos = dst.position();
int lim = dst.limit();
assert (pos <= lim);
int rem = (pos <= lim ? lim - pos : 0);
if (dst instanceof DirectBuffer && rem > 0)
return receiveIntoNativeBuffer(fd, resultContainer, dst, rem, pos, peek);
/* Substitute a native buffer */
int newSize = Math.max(rem, 1);
ByteBuffer bb = Util.getTemporaryDirectBuffer(newSize);
try {
int n = receiveIntoNativeBuffer(fd, resultContainer, bb, newSize, 0, peek);
bb.flip();
if (n > 0 && rem > 0)
dst.put(bb);
return n;
} finally {
Util.releaseTemporaryDirectBuffer(bb);
}
}
private int receiveIntoNativeBuffer(int fd,
SctpResultContainer resultContainer,
ByteBuffer bb,
int rem,
int pos,
boolean peek)
throws IOException
{
int n = receive0(fd, resultContainer, ((DirectBuffer)bb).address() + pos, rem, peek);
if (n > 0)
bb.position(pos + n);
return n;
}
private InternalNotificationHandler<?> internalNotificationHandler =
new InternalNotificationHandler();
private void handleNotificationInternal(SctpResultContainer resultContainer)
{
invokeNotificationHandler(resultContainer,
internalNotificationHandler, null);
}
private class InternalNotificationHandler<T>
extends AbstractNotificationHandler<T>
{
@Override
public HandlerResult handleNotification(
AssociationChangeNotification not, T unused) {
if (not.event().equals(
AssociationChangeNotification.AssocChangeEvent.COMM_UP) &&
association == null) {
SctpAssocChange sac = (SctpAssocChange) not;
association = new SctpAssociationImpl
(sac.assocId(), sac.maxInStreams(), sac.maxOutStreams());
}
return HandlerResult.CONTINUE;
}
}
private <T> HandlerResult invokeNotificationHandler
(SctpResultContainer resultContainer,
NotificationHandler<T> handler,
T attachment) {
SctpNotification notification = resultContainer.notification();
synchronized (stateLock) {
notification.setAssociation(association);
}
if (!(handler instanceof AbstractNotificationHandler)) {
return handler.handleNotification(notification, attachment);
}
/* AbstractNotificationHandler */
AbstractNotificationHandler absHandler =
(AbstractNotificationHandler)handler;
switch(resultContainer.type()) {
case ASSOCIATION_CHANGED :
return absHandler.handleNotification(
resultContainer.getAssociationChanged(), attachment);
case PEER_ADDRESS_CHANGED :
return absHandler.handleNotification(
resultContainer.getPeerAddressChanged(), attachment);
case SEND_FAILED :
return absHandler.handleNotification(
resultContainer.getSendFailed(), attachment);
case SHUTDOWN :
return absHandler.handleNotification(
resultContainer.getShutdown(), attachment);
default :
/* implementation specific handlers */
return absHandler.handleNotification(
resultContainer.notification(), attachment);
}
}
private void checkAssociation(Association sendAssociation) {
synchronized (stateLock) {
if (sendAssociation != null && !sendAssociation.equals(association)) {
throw new IllegalArgumentException(
"Cannot send to another association");
}
}
}
private void checkStreamNumber(int streamNumber) {
synchronized (stateLock) {
if (association != null) {
if (streamNumber < 0 ||
streamNumber >= association.maxOutboundStreams())
throw new InvalidStreamException();
}
}
}
/* TODO: Add support for ttl and isComplete to both 121 12M
* SCTP_EOR not yet supported on reference platforms
* TTL support limited...
*/
@Override
public int send(ByteBuffer buffer, MessageInfo messageInfo)
throws IOException {
if (buffer == null)
throw new IllegalArgumentException("buffer cannot be null");
if (messageInfo == null)
throw new IllegalArgumentException("messageInfo cannot be null");
checkAssociation(messageInfo.association());
checkStreamNumber(messageInfo.streamNumber());
synchronized (sendLock) {
ensureSendOpen();
int n = 0;
try {
begin();
synchronized (stateLock) {
if(!isOpen())
return 0;
senderThread = NativeThread.current();
}
do {
n = send(fdVal, buffer, messageInfo);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
return IOStatus.normalize(n);
} finally {
senderCleanup();
end((n > 0) || (n == IOStatus.UNAVAILABLE));
assert IOStatus.check(n);
}
}
}
private int send(int fd, ByteBuffer src, MessageInfo messageInfo)
throws IOException {
int streamNumber = messageInfo.streamNumber();
SocketAddress target = messageInfo.address();
boolean unordered = messageInfo.isUnordered();
int ppid = messageInfo.payloadProtocolID();
if (src instanceof DirectBuffer)
return sendFromNativeBuffer(fd, src, target, streamNumber,
unordered, ppid);
/* Substitute a native buffer */
int pos = src.position();
int lim = src.limit();
assert (pos <= lim && streamNumber >= 0);
int rem = (pos <= lim ? lim - pos : 0);
ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
try {
bb.put(src);
bb.flip();
/* Do not update src until we see how many bytes were written */
src.position(pos);
int n = sendFromNativeBuffer(fd, bb, target, streamNumber,
unordered, ppid);
if (n > 0) {
/* now update src */
src.position(pos + n);
}
return n;
} finally {
Util.releaseTemporaryDirectBuffer(bb);
}
}
private int sendFromNativeBuffer(int fd,
ByteBuffer bb,
SocketAddress target,
int streamNumber,
boolean unordered,
int ppid)
throws IOException {
InetAddress addr = null; // no preferred address
int port = 0;
if (target != null) {
InetSocketAddress isa = Net.checkAddress(target);
addr = isa.getAddress();
port = isa.getPort();
}
int pos = bb.position();
int lim = bb.limit();
assert (pos <= lim);
int rem = (pos <= lim ? lim - pos : 0);
int written = send0(fd, ((DirectBuffer)bb).address() + pos, rem, addr,
port, -1 /*121*/, streamNumber, unordered, ppid);
if (written > 0)
bb.position(pos + written);
return written;
}
@Override
public SctpChannel shutdown() throws IOException {
synchronized(stateLock) {
if (isShutdown)
return this;
ensureSendOpen();
SctpNet.shutdown(fdVal, -1);
if (senderThread != 0)
NativeThread.signal(senderThread);
isShutdown = true;
}
return this;
}
@Override
public Set<SocketAddress> getAllLocalAddresses()
throws IOException {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (!isBound())
return Collections.EMPTY_SET;
return SctpNet.getLocalAddresses(fdVal);
}
}
@Override
public Set<SocketAddress> getRemoteAddresses()
throws IOException {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
if (!isConnected() || isShutdown)
return Collections.EMPTY_SET;
try {
return SctpNet.getRemoteAddresses(fdVal, 0/*unused*/);
} catch (SocketException unused) {
/* an open connected channel should always have remote addresses */
return remoteAddresses;
}
}
}
/* Native */
private static native void initIDs();
static native int receive0(int fd, SctpResultContainer resultContainer,
long address, int length, boolean peek) throws IOException;
static native int send0(int fd, long address, int length,
InetAddress addr, int port, int assocId, int streamNumber,
boolean unordered, int ppid) throws IOException;
private static native int checkConnect(FileDescriptor fd, boolean block,
boolean ready) throws IOException;
static {
Util.load(); /* loads nio & net native libraries */
java.security.AccessController.doPrivileged(
new sun.security.action.LoadLibraryAction("sctp"));
initIDs();
}
}