| /* |
| * 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(); |
| } |
| } |
