Author: mturk
Date: Mon Jul 18 04:51:11 2011
New Revision: 1147743
URL: http://svn.apache.org/viewvc?rev=1147743&view=rev
Log:
Implement methods and test for SocketEndpoint
Added:
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestLocalEndpoint.java
- copied, changed from r1147618,
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestServerEndpoint.java
(with props)
Removed:
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Reader.java
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Writer.java
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/LocalServerEndpoint.java
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketAddress.java
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketDescriptor.java
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketEndpoint.java
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketServerEndpoint.java
commons/sandbox/runtime/trunk/src/main/native/os/unix/inetsock.c
commons/sandbox/runtime/trunk/src/main/native/os/unix/localsock.c
commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c
commons/sandbox/runtime/trunk/src/main/native/shared/netaddr.c
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Reader.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Reader.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Reader.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Reader.java
Mon Jul 18 04:51:11 2011
@@ -20,6 +20,7 @@ package org.apache.commons.runtime.io;
import java.io.IOException;
import java.nio.ByteBuffer;
+import org.apache.commons.runtime.DirectByteBuffer;
import org.apache.commons.runtime.Pointer;
/**
@@ -191,7 +192,7 @@ public interface Reader
* blocking mode.
* <p>
* {@code ByteBuffer} must be allocated using {@code allocatedirect()}
- * or obtained from {@link NioByteBuffer}.
+ * or obtained from {@link DirectByteBuffer}.
* </p>
*
* @param buffer
@@ -220,7 +221,7 @@ public interface Reader
* is reached or an exception is thrown.
* <p>
* {@code ByteBuffer} must be allocated using {@code allocatedirect()}
- * or obtained from {@link NioByteBuffer}.
+ * or obtained from {@link DirectByteBuffer}.
* </p>
*
* @param buffer
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Writer.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Writer.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Writer.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/io/Writer.java
Mon Jul 18 04:51:11 2011
@@ -20,6 +20,7 @@ package org.apache.commons.runtime.io;
import java.io.IOException;
import java.nio.ByteBuffer;
+import org.apache.commons.runtime.DirectByteBuffer;
import org.apache.commons.runtime.Pointer;
/**
@@ -166,7 +167,7 @@ public interface Writer extends Syncable
* to this stream, starting at the current stream position.
* <p>
* {@code ByteBuffer} must be allocated using {@code allocatedirect()}
- * or obtained from {@link NioByteBuffer}.
+ * or obtained from {@link DirectByteBuffer}.
* </p>
*
* @param buffer
@@ -193,7 +194,7 @@ public interface Writer extends Syncable
* to get bytes.
* <p>
* {@code ByteBuffer} must be allocated using {@code allocatedirect()}
- * or obtained from {@link NioByteBuffer}.
+ * or obtained from {@link DirectByteBuffer}.
* </p>
*
* @param buffer
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/LocalServerEndpoint.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/LocalServerEndpoint.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/LocalServerEndpoint.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/LocalServerEndpoint.java
Mon Jul 18 04:51:11 2011
@@ -61,12 +61,12 @@ public class LocalServerEndpoint extends
* Creates a new local server endpoint from the
* given socket descriptor.
*/
- public LocalServerEndpoint(SocketDescriptor sd)
+ public LocalServerEndpoint(Descriptor sd)
{
super(EndpointType.LOCAL);
if (sd == null)
throw new NullPointerException();
- this.sd = sd;
+ this.sd = (SocketDescriptor)sd;
}
@Override
@@ -157,7 +157,7 @@ public class LocalServerEndpoint extends
}
private static native int bind0(long fd, byte[] sa, int backlog);
- private static native long accept0(long fd, byte[] sa, boolean
block)
+ private static native long accept0(long fd, boolean block)
throws SocketException;
@Override
@@ -196,7 +196,7 @@ public class LocalServerEndpoint extends
{
if (sd.closed())
throw new ClosedDescriptorException();
- long fd = accept0(sd.fd(), sa.sockaddr(), blocking);
+ long fd = accept0(sd.fd(), blocking);
SocketDescriptor ad = new SocketDescriptor(fd);
return new LocalEndpoint(ad, sa);
}
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketAddress.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketAddress.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketAddress.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketAddress.java
Mon Jul 18 04:51:11 2011
@@ -26,8 +26,9 @@ import org.apache.commons.runtime.Status
import org.apache.commons.runtime.InvalidArgumentException;
/**
- * This class represents a socket endpoint described by a IP address and a port
- * number. It is a concrete implementation of {@code java.net.SocketAddress}
for IP.
+ * This class represents a socket endpoint described by a IP address
+ * and a port number.
+ * It is a concrete implementation of {@code EndpointAddress} for IP.
*/
public abstract class SocketAddress extends EndpointAddress
{
@@ -55,14 +56,16 @@ public abstract class SocketAddress exte
}
/**
- * Creates an new object
+ * Creates a new object.
+ *
*/
protected SocketAddress()
{
}
/**
- * Creates an new object
+ * Creates a new object.
+ *
* @param family indicates a protocl family accepted.
*/
protected SocketAddress(AddressFamily family)
@@ -82,7 +85,8 @@ public abstract class SocketAddress exte
}
/**
- * Creates an new object
+ * Creates a new object.
+ *
* @param family indicates a protocl family accepted.
* @param sockaddr binary socket address.
*/
@@ -93,7 +97,7 @@ public abstract class SocketAddress exte
}
/**
- * Gets the hostname of this socket.
+ * Gets the hostname of this address.
* <p>
* The methods does not resolve the host name. If the address was
* not resloved already the method returns IP notion of the address.
@@ -120,9 +124,9 @@ public abstract class SocketAddress exte
}
/**
- * Gets the service name of this socket.
+ * Gets the service name for this address.
*
- * @return the socket endpoint service name.
+ * @return the service name.
*/
public final String getServiceName()
{
@@ -130,15 +134,20 @@ public abstract class SocketAddress exte
}
/**
- * Gets the port number of this socket.
+ * Gets the port number of this address.
*
- * @return the socket endpoint port number.
+ * @return the port number.
*/
public final int getPort()
{
return port0(super.sa);
}
+ /**
+ * Gets the string IP representation of this address.
+ *
+ * @return IP address string.
+ */
public final String getIpAddress()
{
return ipaddr0(super.sa);
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketDescriptor.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketDescriptor.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketDescriptor.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketDescriptor.java
Mon Jul 18 04:51:11 2011
@@ -68,8 +68,10 @@ final class SocketDescriptor extends Des
public void create(AddressFamily af, SocketType type, boolean blocking)
throws IOException
{
- this.fd = socket0(af.valueOf(), type.valueOf(), blocking);
- closed = false;
+ if (valid())
+ close0(fd);
+ fd = socket0(af.valueOf(), type.valueOf(), blocking);
+ closed = false;
}
public void create(SocketType type)
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketEndpoint.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketEndpoint.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketEndpoint.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketEndpoint.java
Mon Jul 18 04:51:11 2011
@@ -25,9 +25,13 @@ import java.io.SyncFailedException;
import java.net.SocketException;
import org.apache.commons.runtime.io.ClosedDescriptorException;
import org.apache.commons.runtime.io.Descriptor;
+import org.apache.commons.runtime.io.OperationInProgressException;
+import org.apache.commons.runtime.io.OperationWouldBlockException;
import org.apache.commons.runtime.io.Stream;
-import org.apache.commons.runtime.OverflowException;
+import org.apache.commons.runtime.Errno;
import org.apache.commons.runtime.Status;
+import org.apache.commons.runtime.OverflowException;
+import org.apache.commons.runtime.TimeoutException;
/**
* This class represents a socket endpoint.
@@ -40,6 +44,7 @@ public class SocketEndpoint extends Conn
private SocketAddress sa = null; // Local address
private SocketStream stream = null;
private boolean connected = false;
+ private static native int connect0(long fd, byte[] sa, int
timeout);
/**
* Creates a new unconnected socket endpoint.
@@ -52,15 +57,16 @@ public class SocketEndpoint extends Conn
/**
* Creates a new connected socket endpoint from a socket descriptor
- * and remote address.
+ * and remote and local address.
*/
- public SocketEndpoint(Descriptor sd, SocketAddress ea)
+ public SocketEndpoint(Descriptor sd, SocketAddress remote, SocketAddress
local)
{
super(EndpointType.SOCKET);
- if (sd == null || ea == null)
+ if (sd == null || remote == null)
throw new NullPointerException();
this.sd = (SocketDescriptor)sd;
- this.ea = ea;
+ this.ea = remote;
+ this.sa = local;
connected = true;
}
@@ -77,6 +83,18 @@ public class SocketEndpoint extends Conn
sd.create(sockaddr.getFamily(), SocketType.STREAM);
if (Address.resolved(sockaddr))
ea = sockaddr;
+ int rc = connect0(sd.fd(), sockaddr.sockaddr(), timeout);
+ if (rc != 0) {
+ // XXX: Should exception throwing go inside native?
+ if (Status.IS_TIMEUP(rc))
+ throw new TimeoutException();
+ else if (Status.IS_EAGAIN(rc))
+ throw new OperationWouldBlockException();
+ else if (Status.IS_EINPROGRESS(rc))
+ throw new OperationInProgressException();
+ else
+ throw new IOException(Status.describe(rc));
+ }
connected = true;
}
@@ -205,34 +223,75 @@ public class SocketEndpoint extends Conn
return connected;
}
+ /**
+ * Gets the local address this socket is bound to.
+ *
+ * @return the local address of this socket or {@code null} if
+ * the socket is unbound.
+ * @throws IOException if there was an error when resolving
+ * the local address.
+ */
public synchronized SocketAddress getLocalAddress()
throws IOException
{
- if (closed())
- throw new ClosedDescriptorException();
- if (ea != null)
- return ea;
- ea = new AbstractSocketAddress(sd.getLocalAddress());
+ if (connected) {
+ if (ea != null)
+ return ea;
+ ea = new AbstractSocketAddress(sd.getLocalAddress());
+ }
return ea;
}
+ /**
+ * Gets the remote address of this socket or {@code null} if
+ * the socket is not connected.
+ *
+ * @return the remote address of this socket or {@code null} if
+ * the socket is not connected.
+ * @throws IOException if there was an error when resolving
+ * the remote address.
+ */
public synchronized SocketAddress getRemoteAddress()
throws IOException
{
- if (closed())
- throw new ClosedDescriptorException();
+ if (!connected)
+ return null;
if (sa != null)
return sa;
sa = new AbstractSocketAddress(sd.getRemoteAddress());
return sa;
}
+ /**
+ * Returns the {@code SocketOption} state of the socket.
+ *
+ * @param key the {@code SocketOption} to check.
+ * @return {@code true} if the socket has the requested
+ * option set.
+ * @throws OperationNotImplededException
+ * if the socket does not support this option.
+ * @throws IOException if there was an error when querying
+ * the socket option.
+ * @see SocketOption
+ */
public boolean hasOption(SocketOption key)
throws IOException
{
return sd.hasOption(key);
}
+ /**
+ * Returns the {@code TcpOption} state of the socket.
+ *
+ * @param key the {@code TcpOption} to check.
+ * @return {@code true} if the socket has the requested
+ * option set.
+ * @throws OperationNotImplededException
+ * if the socket does not support this option.
+ * @throws IOException if there was an error when querying
+ * the socket option.
+ * @see TcpOption
+ */
public boolean hasOption(TcpOption key)
throws IOException
{
@@ -241,22 +300,64 @@ public class SocketEndpoint extends Conn
return sd.hasOption(key);
}
+ /**
+ * Sets the value of the {@code SocketOption} for this socket.
+ *
+ * @param key the {@code SocketOption} to set.
+ * @param val the value to set.
+ *
+ * @throws InvalidArgumentException
+ * if the socket does not support setting numerical values.
+ * @throws OperationNotImplededException
+ * if the socket does not support this option.
+ * @throws IOException if there was an error when setting
+ * the socket option.
+ * @see SocketOption
+ */
public void setOption(SocketOption key, int val)
throws IOException
{
sd.setOption(key, val);
}
- public void setOption(SocketOption key, boolean val)
+ /**
+ * Sets the state of the {@code SocketOption} for this socket.
+ *
+ * @param key the {@code SocketOption} to set.
+ * @param on the state whether this option is enabled or not.
+ *
+ * @throws InvalidArgumentException
+ * if the socket does not support setting boolean values.
+ * @throws OperationNotImplededException
+ * if the socket does not support this option.
+ * @throws IOException if there was an error when setting
+ * the socket option.
+ * @see SocketOption
+ */
+ public void setOption(SocketOption key, boolean on)
throws IOException
{
- sd.setOption(key, val);
+ sd.setOption(key, on);
}
- public void setOption(TcpOption key, boolean val)
+ /**
+ * Sets the state of the {@code TcpOption} for this socket.
+ *
+ * @param key the {@code TcpOption} to set.
+ * @param on the state whether this option is enabled or not.
+ *
+ * @throws InvalidArgumentException
+ * if the socket does not support setting boolean values.
+ * @throws OperationNotImplededException
+ * if the socket does not support this option.
+ * @throws IOException if there was an error when setting
+ * the socket option.
+ * @see TcpOption
+ */
+ public void setOption(TcpOption key, boolean on)
throws IOException
{
- sd.setOption(key, val);
+ sd.setOption(key, on);
}
}
Modified:
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketServerEndpoint.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketServerEndpoint.java?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketServerEndpoint.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/java/org/apache/commons/runtime/net/SocketServerEndpoint.java
Mon Jul 18 04:51:11 2011
@@ -27,6 +27,7 @@ import org.apache.commons.runtime.io.Clo
import org.apache.commons.runtime.io.Descriptor;
import org.apache.commons.runtime.io.OperationInProgressException;
import org.apache.commons.runtime.io.OperationWouldBlockException;
+import org.apache.commons.runtime.OperationNotImplementedException;
import org.apache.commons.runtime.OverflowException;
import org.apache.commons.runtime.Status;
@@ -58,12 +59,12 @@ public class SocketServerEndpoint extend
* Creates a new local server endpoint from the
* given socket descriptor.
*/
- public SocketServerEndpoint(SocketDescriptor sd)
+ public SocketServerEndpoint(Descriptor sd)
{
super(EndpointType.SOCKET);
if (sd == null)
throw new NullPointerException();
- this.sd = sd;
+ this.sd = (SocketDescriptor)sd;
}
@Override
@@ -152,7 +153,7 @@ public class SocketServerEndpoint extend
}
private static native int bind0(long fd, byte[] sa, int backlog);
- private static native long accept0(long fd, byte[] sa, boolean
block)
+ private static native long accept0(long fd, byte[] aa, boolean
block)
throws SocketException;
@Override
@@ -161,11 +162,31 @@ public class SocketServerEndpoint extend
{
if (bound)
throw new IOException(Local.sm.get("endpoint.EBOUND"));
- if (sd.closed())
+ if (sd.closed()) {
sd.create(endpoint.getFamily(), SocketType.STREAM, blocking);
- sa = new AbstractSocketAddress(endpoint);
+ // Set default options
+ // NOTICE: Options are taken from the Apache Web Server.
+ // and should be set up for any server.
+ sd.setOption(SocketOption.KEEPALIVE, true);
+ sd.setOption(TcpOption.NODELAY, true);
+ if (endpoint.getFamily() == AddressFamily.INET6) {
+ try {
+ sd.setOption(SocketOption.IPV6_V6ONLY, true);
+ } catch (OperationNotImplementedException u) {
+ // IPV6_V6ONLY might not be supported
+ // on all platforms
+ }
+ }
+
+ // NOTICE: REUSEADDR is set inside native bind0 call
+ // depending on the operating system.
+ }
if (backlog == 0)
backlog = LISTEN_BACKLOG;
+ sa = new AbstractSocketAddress(endpoint);
+ // Bind and listen on the socket.
+ // sa will be updated in case it had ephemeral port or
+ // its address was INADDR_ANY.
int rc = bind0(sd.fd(), sa.sockaddr(), backlog);
if (rc != 0) {
try {
@@ -191,9 +212,11 @@ public class SocketServerEndpoint extend
{
if (sd.closed())
throw new ClosedDescriptorException();
- long fd = accept0(sd.fd(), sa.sockaddr(), blocking);
+ byte[] aa = Address.alloc();
+ long fd = accept0(sd.fd(), aa, blocking);
SocketDescriptor ad = new SocketDescriptor(fd);
- return new SocketEndpoint(ad, sa);
+ SocketAddress ra = new AbstractSocketAddress(aa);
+ return new SocketEndpoint(ad, ra, sa);
}
@Override
@@ -210,4 +233,33 @@ public class SocketServerEndpoint extend
return sd.closed();
}
+ /**
+ * Gets the local port of this server endpoint is listening on.
+ *
+ * @return the local port number this server is listening on
+ * or {@code -1} if the socket is unbound.
+ */
+ public final int getLocalPort()
+ {
+ if (bound)
+ return sa.getPort();
+ else
+ return -1;
+ }
+
+ /**
+ * Gets the local socket address of this server endpoint or {@code null}
+ * if the socket is unbound. This is useful on multihomed hosts.
+ *
+ * @return the local socket address this socket is bound to.
+ */
+ public final SocketAddress getLocalAddress()
+ {
+ if (bound)
+ return sa;
+ else
+ return null;
+ }
+
+
}
Modified: commons/sandbox/runtime/trunk/src/main/native/os/unix/inetsock.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/os/unix/inetsock.c?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/os/unix/inetsock.c (original)
+++ commons/sandbox/runtime/trunk/src/main/native/os/unix/inetsock.c Mon Jul 18
04:51:11 2011
@@ -16,6 +16,7 @@
#include "acr/jnitypes.h"
#include "acr/error.h"
+#include "acr/debug.h"
#include "acr/memory.h"
#include "acr/netapi.h"
#include "acr/unsafe.h"
@@ -294,7 +295,7 @@ ACR_NET_EXPORT(jint, SocketServerEndpoin
*/
if (aa->port == 0 || memcmp(&aa->sa.sin.sin_addr, any, aa->iplen) == 0) {
/* Get the real address for the bound socket.
- */
+ */
if ((rc = AcrGetLocalAddr(fd, aa)) != 0)
goto finally;
}
@@ -306,7 +307,7 @@ finally:
}
ACR_NET_EXPORT(jlong, SocketServerEndpoint, accept0)(JNI_STDARGS, jlong fp,
- jbyteArray ba,
+ jbyteArray ra,
jboolean block)
{
int sd;
@@ -323,9 +324,10 @@ ACR_NET_EXPORT(jlong, SocketServerEndpoi
flags |= SOCK_NONBLOCK;
# endif
#endif
- ad = AcrSdRetain(fd);
memset(&aa, 0, sizeof(aa));
- aalen = ISIZEOF(struct sockaddr_un);
+ aalen = SSIZEOF(acr_sockaddr_t);
+
+ ad = AcrSdRetain(fd);
do {
#if HAVE_ACCEPT4
sd = accept4(ad, (struct sockaddr *)&aa.sa, &aalen, flags);
@@ -412,5 +414,77 @@ ACR_NET_EXPORT(jlong, SocketServerEndpoi
if (sp->timeout == 0)
sp->timeout = -1;
}
+ switch (aa.sa.sin.sin_family) {
+ case AF_INET:
+ aa.addrlen = 16;
+ aa.iplen = ISIZEOF(struct in_addr);
+ break;
+ case AF_INET6:
+ aa.addrlen = 46;
+ aa.iplen = ISIZEOF(struct in6_addr);
+ default:
+#if defined(DEBUG) || defined(_DEBUG)
+ ACR_DEBUG_TRACE("Unexpected address family %d\n", aa.family);
+#endif
+ break;
+ }
+#if defined(DEBUG) || defined(_DEBUG)
+ if (aa.addrlen != (int)aalen) {
+ ACR_DEBUG_TRACE("Address length missmatch. Expected %d, found %d\n",
aa.addrlen, (int)aalen);
+ }
+#endif
+ aa.salen = aa.iplen;
+ aa.family = aa.sa.sin.sin_family;
+ aa.port = ntohs(aa.sa.sin.sin_port);
+ (*env)->SetByteArrayRegion(env, ra, 0, ISIZEOF(acr_sockaddr_t), (jbyte
*)&aa);
return P2J(sp);
}
+
+ACR_NET_EXPORT(jint, SocketEndpoint, connect0)(JNI_STDARGS, jlong fp,
+ jbyteArray cb, jint timeout)
+{
+ int rc;
+ int sd;
+ acr_sockaddr_t *ca = SOCKADDR_CAST(cb);
+ acr_sd_t *fd = J2P(fp, acr_sd_t *);
+
+ sd = AcrSdRetain(fd);
+ if (timeout == 0)
+ timeout = fd->timeout;
+ if (timeout > 0 && !ACR_HASFLAG(fd, ACR_SO_NONBLOCK)) {
+ /* Turn the socket to non-blocking mode
+ */
+ if ((rc = AcrNonblock(sd, 1)) != 0)
+ goto finally;
+ }
+ do {
+ /* Restartable connect */
+ rc = connect(sd, (const struct sockaddr *)&ca->sa.sin, ca->salen);
+ } while (rc == -1 && errno == EINTR);
+
+ if (rc == -1)
+ rc = errno;
+ SOCKADDR_RELEASE(cb, ca);
+ if (rc != 0) {
+ if (timeout > 0) {
+ if (rc == EINPROGRESS || rc == EALREADY) {
+ rc = AcrWaitIO(fd->s, timeout, POLLOUT);
+#if defined(SO_ERROR)
+ if (rc == 0) {
+ int err;
+ socklen_t len = sizeof(err);
+ if (getsockopt(sd, SOL_SOCKET, SO_ERROR, (char *)&err,
&len) == -1);
+ rc = errno;
+ if (err != 0)
+ rc = err;
+ }
+#endif
+ }
+ if (!ACR_HASFLAG(fd, ACR_SO_NONBLOCK))
+ AcrNonblock(fd->s, 0);
+ }
+ }
+finally:
+ AcrSdRelease(fd);
+ return rc;
+}
Modified: commons/sandbox/runtime/trunk/src/main/native/os/unix/localsock.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/os/unix/localsock.c?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/os/unix/localsock.c (original)
+++ commons/sandbox/runtime/trunk/src/main/native/os/unix/localsock.c Mon Jul
18 04:51:11 2011
@@ -119,7 +119,6 @@ ACR_NET_EXPORT(void, LocalServerEndpoint
}
ACR_NET_EXPORT(jlong, LocalServerEndpoint, accept0)(JNI_STDARGS, jlong fp,
- jbyteArray ba,
jboolean block)
{
int sd;
@@ -164,17 +163,6 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
}
}
#endif
-#if !TCP_NODELAY_INHERITED
- if (ACR_HASFLAG(fd, ACR_TCP_NODELAY)) {
- int on = 1;
- if (setsockopt(sd, IPPROTO_TCP, TCP_NODELAY, (void *)&on,
SSIZEOF(int)) != 0) {
- int rc = ACR_GET_NETOS_ERROR();
- r_close(sd);
- ACR_THROW_NET_ERROR(rc);
- return 0;
- }
- }
-#endif
if (block == JNI_FALSE) {
int rc = 0;
#if O_NONBLOCK_INHERITED
@@ -210,10 +198,6 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
return 0;
}
sp->type = ACR_DT_LOCALSOCK;
- /* Inherit nodelay flag
- * It's always valid cause we either inherit or explicitly set that flag.
- */
- sp->flags = fd->flags & ACR_TCP_NODELAY;
#if O_NONBLOCK_INHERITED
/* Inherit timeout */
sp->timeout = fd->timeout;
@@ -226,7 +210,6 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
sp->timeout = 0;
}
else {
- ACR_CLRFLAG(sp, ACR_SO_NONBLOCK);
if (sp->timeout == 0)
sp->timeout = -1;
}
Modified: commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c
(original)
+++ commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c Mon Jul
18 04:51:11 2011
@@ -194,12 +194,10 @@ ACR_NET_EXPORT(void, LocalServerEndpoint
}
ACR_NET_EXPORT(jlong, LocalServerEndpoint, accept0)(JNI_STDARGS, jlong fp,
- jbyteArray ba,
jboolean block)
{
SOCKET sd;
SOCKET ad;
- acr_sockaddr_t *aa;
struct sockaddr_in sa;
int sas = ISIZEOF(sa);
acr_sd_t *sp;
@@ -242,14 +240,6 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
closesocket(sd);
return 0;
}
- aa = SOCKADDR_CAST(ba);
- if (aa != 0) {
- /* The only thing that is unique is port
- */
- aa->port = ntohs(sa.sin_port);
- }
- SOCKADDR_RELEASE(ba, aa);
-
sp->type = ACR_DT_LOCALSOCK;
sp->flags = fd->flags & ACR_SO_NONBLOCK;
sp->timeout = fd->timeout;
Modified: commons/sandbox/runtime/trunk/src/main/native/shared/netaddr.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/shared/netaddr.c?rev=1147743&r1=1147742&r2=1147743&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/shared/netaddr.c (original)
+++ commons/sandbox/runtime/trunk/src/main/native/shared/netaddr.c Mon Jul 18
04:51:11 2011
@@ -447,18 +447,15 @@ sockaddr_vars_set(acr_sockaddr_t *addr,
if (family == AF_INET) {
addr->salen = ISIZEOF(struct sockaddr_in);
addr->addrlen = 16;
- addr->iplen = sizeof(struct in_addr);
}
else if (family == AF_INET6) {
addr->salen = ISIZEOF(struct sockaddr_in6);
addr->addrlen = 46;
- addr->iplen = sizeof(struct in6_addr);
}
#if HAVE_SYS_UN_H
else if (family == AF_LOCAL) {
addr->salen = ISIZEOF(struct sockaddr_un);
addr->addrlen = ISIZEOF(addr->sa.unx.sun_path);
- addr->iplen = addr->addrlen;
}
#endif
else {
@@ -466,6 +463,7 @@ sockaddr_vars_set(acr_sockaddr_t *addr,
addr->addrlen = 0;
addr->iplen = 0;
}
+ addr->iplen = addr->salen;
}
static int
Copied:
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestLocalEndpoint.java
(from r1147618,
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java)
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestLocalEndpoint.java?p2=commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestLocalEndpoint.java&p1=commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java&r1=1147618&r2=1147743&rev=1147743&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestLocalEndpoint.java
Mon Jul 18 04:51:11 2011
@@ -24,7 +24,7 @@ import org.testng.Assert;
import org.apache.commons.runtime.io.Descriptor;
-public class TestPosixEndpoint extends Assert
+public class TestLocalEndpoint extends Assert
{
@BeforeSuite
Added:
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestServerEndpoint.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestServerEndpoint.java?rev=1147743&view=auto
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestServerEndpoint.java
(added)
+++
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestServerEndpoint.java
Mon Jul 18 04:51:11 2011
@@ -0,0 +1,157 @@
+/* Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.commons.runtime.net;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.testng.annotations.*;
+import org.testng.Assert;
+
+import org.apache.commons.runtime.io.Descriptor;
+
+public class TestServerEndpoint extends Assert
+{
+
+ private Object sync = new Object();
+
+ class Acceptor extends Thread
+ {
+ Selector ps;
+ SocketServerEndpoint ss;
+ public volatile boolean running = true;
+
+ public Acceptor(SocketServerEndpoint ss, Selector ps)
+ {
+ this.ss = ss;
+ this.ps = ps;
+ }
+
+ public void run()
+ {
+ SelectionKey sk;
+ try {
+ sk = ss.register(ps, SelectionKey.OP_ACCEPT);
+ assertEquals(ps.size(), 1);
+ } catch (Exception x) {
+ fail("Acceptor setup failed " + x.toString());
+ running = false;
+ return;
+ }
+ while (true) {
+ try {
+ synchronized (sync) {
+ // Notify that we are ready to
+ // accept the connections
+ sync.notifyAll();
+ }
+ System.out.print("Accepting connection ... ");
+ List<SelectionKey> set = ps.select();
+ if (set.size() == 0) {
+ System.out.println("interrupted.");
+ break;
+ }
+ System.out.println("done (" + set.size() + " connection)");
+ assertEquals(set.size(), 1);
+ sk = set.get(0);
+ SocketEndpoint e =
((SocketServerEndpoint)sk.endpoint()).accept();
+ assertNotNull(e.descriptor());
+ // Another select with zero timeout
+ set = ps.select(0);
+ assertEquals(set.size(), 0);
+ } catch (Exception x) {
+ fail("Accept failed " + x.toString());
+ break;
+ }
+ }
+ try {
+ ss.close();
+ } catch (Exception x) {
+ fail("Acceptor shutdown failed " + x.toString());
+ }
+ running = false;
+ System.out.println("Accepting thread finished");
+ synchronized (sync) {
+ // Notify that we are ready to
+ // accept the connections
+ sync.notifyAll();
+ }
+ }
+ }
+
+ @Test(groups = { "core" })
+ public void connectServerEndpoint()
+ throws Exception
+ {
+ System.out.println("Testing ServerSocketEndpoint ...");
+
+ Selector ps = Selector.open(EndpointType.LOCAL);
+ SocketServerEndpoint ss = new SocketServerEndpoint();
+ InetSocketAddress sa = new InetSocketAddress("127.0.0.1", 0);
+ ss.configureBlocking(false);
+ assertEquals(ss.getLocalPort(), -1);
+ ss.bind(sa);
+ assertFalse(ss.getLocalPort() == 0);
+ // keyFor() should return null since the endpoint
+ // is not yet registered.
+ assertNull(ss.keyFor(ps));
+ // ps.setAutoCancel(true);
+ Acceptor aw = new Acceptor(ss, ps);
+ aw.setName("TestSocketAcceptorThread");
+ aw.start();
+ try {
+ synchronized (sync) {
+ // Wait until Acceptor is ready to accept connections
+ //
+ sync.wait();
+ }
+ Thread.sleep(100);
+ } catch (InterruptedException x) {
+ // Ignore
+ }
+ // By this time the key is set up.
+ assertNotNull(ss.keyFor(ps));
+ // Connect to the Acceptor
+ //
+ SocketEndpoint cs = new SocketEndpoint();
+ cs.connect(ss.getLocalAddress());
+ assertTrue(cs.isBlocking());
+ cs.close();
+ ps.interrupt();
+ try {
+ synchronized (sync) {
+ // Wait until Acceptor process the accepted connection
+ //
+ if (aw.running)
+ sync.wait();
+ }
+ Thread.sleep(200);
+ } catch (InterruptedException x) {
+ // Ignore
+ }
+ System.out.print("Interrupting selector ... ");
+ ps.interrupt();
+ System.out.println("OK");
+ // Wait for the Acceptor thread to finish
+ aw.join();
+ List<SelectionKey> set = ps.clear();
+ // Shoud equal zero if autoCancel is set to true
+ assertEquals(set.size(), 1);
+ ps.close();
+ }
+
+}
Propchange:
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestServerEndpoint.java
------------------------------------------------------------------------------
svn:eol-style = native
