diff --git a/all/pom.xml b/all/pom.xml
index a85be0c8d4..1f553f0372 100644
--- a/all/pom.xml
+++ b/all/pom.xml
@@ -20,7 +20,7 @@
io.netty
netty-parent
- 4.0.0.Alpha5-SNAPSHOT
+ 4.0.0.Alpha6-SNAPSHOT
netty
@@ -117,7 +117,6 @@
maven-clean-plugin
- 2.5
clean-first
@@ -130,7 +129,6 @@
maven-dependency-plugin
- 2.4
unpack-sources
@@ -162,7 +160,6 @@
org.codehaus.mojo
build-helper-maven-plugin
- 1.7
add-source
@@ -203,7 +200,6 @@
maven-resources-plugin
- 2.5
default-resources
@@ -239,7 +235,6 @@
maven-jxr-plugin
- 2.2
generate-xref
@@ -259,6 +254,13 @@
Netty Source Xref (${project.version})
Netty Source Xref (${project.version})
+
+
+ ${project.groupId}
+ netty-build
+ 10
+
+
maven-javadoc-plugin
diff --git a/buffer/pom.xml b/buffer/pom.xml
index 17574e422b..f661965a66 100644
--- a/buffer/pom.xml
+++ b/buffer/pom.xml
@@ -20,7 +20,7 @@
io.netty
netty-parent
- 4.0.0.Alpha5-SNAPSHOT
+ 4.0.0.Alpha6-SNAPSHOT
netty-buffer
diff --git a/buffer/src/main/java/io/netty/buffer/DefaultCompositeByteBuf.java b/buffer/src/main/java/io/netty/buffer/DefaultCompositeByteBuf.java
index 96c8036c55..860925f506 100644
--- a/buffer/src/main/java/io/netty/buffer/DefaultCompositeByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/DefaultCompositeByteBuf.java
@@ -578,14 +578,19 @@ public class DefaultCompositeByteBuf extends AbstractByteBuf implements Composit
@Override
public void getBytes(int index, byte[] dst, int dstIndex, int length) {
- int componentId = toComponentIndex(index);
if (index > capacity() - length || dstIndex > dst.length - length) {
throw new IndexOutOfBoundsException("Too many bytes to read - Needs "
+ (index + length) + ", maximum is " + capacity() + " or "
+ dst.length);
}
+ if (index < 0) {
+ throw new IndexOutOfBoundsException("index must be >= 0");
+ }
+ if (length == 0) {
+ return;
+ }
+ int i = toComponentIndex(index);
- int i = componentId;
while (length > 0) {
Component c = components.get(i);
ByteBuf s = c.buf;
@@ -601,15 +606,20 @@ public class DefaultCompositeByteBuf extends AbstractByteBuf implements Composit
@Override
public void getBytes(int index, ByteBuffer dst) {
- int componentId = toComponentIndex(index);
int limit = dst.limit();
int length = dst.remaining();
+
if (index > capacity() - length) {
throw new IndexOutOfBoundsException("Too many bytes to be read - Needs "
+ (index + length) + ", maximum is " + capacity());
}
-
- int i = componentId;
+ if (index < 0) {
+ throw new IndexOutOfBoundsException("index must be >= 0");
+ }
+ if (length == 0) {
+ return;
+ }
+ int i = toComponentIndex(index);
try {
while (length > 0) {
Component c = components.get(i);
@@ -629,14 +639,18 @@ public class DefaultCompositeByteBuf extends AbstractByteBuf implements Composit
@Override
public void getBytes(int index, ByteBuf dst, int dstIndex, int length) {
- int componentId = toComponentIndex(index);
if (index > capacity() - length || dstIndex > dst.capacity() - length) {
throw new IndexOutOfBoundsException("Too many bytes to be read - Needs "
+ (index + length) + " or " + (dstIndex + length) + ", maximum is "
+ capacity() + " or " + dst.capacity());
}
-
- int i = componentId;
+ if (index < 0) {
+ throw new IndexOutOfBoundsException("index must be >= 0");
+ }
+ if (length == 0) {
+ return;
+ }
+ int i = toComponentIndex(index);
while (length > 0) {
Component c = components.get(i);
ByteBuf s = c.buf;
@@ -670,13 +684,18 @@ public class DefaultCompositeByteBuf extends AbstractByteBuf implements Composit
@Override
public void getBytes(int index, OutputStream out, int length)
throws IOException {
- int componentId = toComponentIndex(index);
if (index > capacity() - length) {
throw new IndexOutOfBoundsException("Too many bytes to be read - needs "
+ (index + length) + ", maximum of " + capacity());
}
+ if (index < 0) {
+ throw new IndexOutOfBoundsException("index must be >= 0");
+ }
+ if (length == 0) {
+ return;
+ }
- int i = componentId;
+ int i = toComponentIndex(index);
while (length > 0) {
Component c = components.get(i);
ByteBuf s = c.buf;
@@ -1031,11 +1050,17 @@ public class DefaultCompositeByteBuf extends AbstractByteBuf implements Composit
@Override
public ByteBuffer[] nioBuffers(int index, int length) {
- int componentId = toComponentIndex(index);
if (index + length > capacity()) {
throw new IndexOutOfBoundsException("Too many bytes to convert - Needs"
+ (index + length) + ", maximum is " + capacity());
}
+ if (index < 0) {
+ throw new IndexOutOfBoundsException("index must be >= 0");
+ }
+ if (length == 0) {
+ return new ByteBuffer[0];
+ }
+ int componentId = toComponentIndex(index);
List buffers = new ArrayList(components.size());
diff --git a/buffer/src/test/java/io/netty/buffer/AbstractCompositeChannelBufferTest.java b/buffer/src/test/java/io/netty/buffer/AbstractCompositeChannelBufferTest.java
index 2900336c25..b73513e358 100644
--- a/buffer/src/test/java/io/netty/buffer/AbstractCompositeChannelBufferTest.java
+++ b/buffer/src/test/java/io/netty/buffer/AbstractCompositeChannelBufferTest.java
@@ -410,4 +410,11 @@ public abstract class AbstractCompositeChannelBufferTest extends
wrappedBuffer(new byte[] { 0, 1, 2, 3, 4, 6, 7, 8, 5, 9, 10, 11 }, 6, 5).order(order));
assertFalse(ByteBufUtil.equals(a, b));
}
+
+ @Test
+ public void testEmptyBuffer() {
+ ByteBuf b = wrappedBuffer(new byte[] {1, 2}, new byte[] {3, 4});
+ b.readBytes(new byte[4]);
+ b.readBytes(new byte[0]);
+ }
}
diff --git a/codec-http/pom.xml b/codec-http/pom.xml
index 7c369d4d0d..8e9d245872 100644
--- a/codec-http/pom.xml
+++ b/codec-http/pom.xml
@@ -20,7 +20,7 @@
io.netty
netty-parent
- 4.0.0.Alpha5-SNAPSHOT
+ 4.0.0.Alpha6-SNAPSHOT
netty-codec-http
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunk.java b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunk.java
index 96705bf4ac..329c203211 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunk.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunk.java
@@ -20,7 +20,7 @@ import io.netty.buffer.ByteBuf;
/**
* The default {@link HttpChunk} implementation.
*/
-public class DefaultHttpChunk implements HttpChunk {
+public class DefaultHttpChunk extends DefaultHttpObject implements HttpChunk {
private ByteBuf content;
private boolean last;
@@ -51,4 +51,24 @@ public class DefaultHttpChunk implements HttpChunk {
public boolean isLast() {
return last;
}
+
+ @Override
+ public String toString() {
+ StringBuilder buf = new StringBuilder();
+ buf.append(getClass().getSimpleName());
+
+ final boolean last = isLast();
+ buf.append("(last: ");
+ buf.append(last);
+ if (!last) {
+ buf.append(", size: ");
+ buf.append(getContent().readableBytes());
+ }
+
+ buf.append(", decodeResult: ");
+ buf.append(getDecoderResult());
+ buf.append(')');
+
+ return buf.toString();
+ }
}
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunkTrailer.java b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunkTrailer.java
index 6cb31aa0bb..29b4cdfced 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunkTrailer.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpChunkTrailer.java
@@ -17,6 +17,7 @@ package io.netty.handler.codec.http;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import io.netty.util.internal.StringUtil;
import java.util.List;
import java.util.Map;
@@ -25,7 +26,7 @@ import java.util.Set;
/**
* The default {@link HttpChunkTrailer} implementation.
*/
-public class DefaultHttpChunkTrailer implements HttpChunkTrailer {
+public class DefaultHttpChunkTrailer extends DefaultHttpObject implements HttpChunkTrailer {
private final HttpHeaders headers = new HttpHeaders() {
@Override
@@ -104,4 +105,37 @@ public class DefaultHttpChunkTrailer implements HttpChunkTrailer {
public void setContent(ByteBuf content) {
throw new IllegalStateException("read-only");
}
+
+ @Override
+ public String toString() {
+ StringBuilder buf = new StringBuilder();
+ buf.append(getClass().getSimpleName());
+
+ final boolean last = isLast();
+ buf.append("(last: ");
+ buf.append(last);
+ if (!last) {
+ buf.append(", size: ");
+ buf.append(getContent().readableBytes());
+ }
+
+ buf.append(", decodeResult: ");
+ buf.append(getDecoderResult());
+ buf.append(')');
+ buf.append(StringUtil.NEWLINE);
+ appendHeaders(buf);
+
+ // Remove the last newline.
+ buf.setLength(buf.length() - StringUtil.NEWLINE.length());
+ return buf.toString();
+ }
+
+ private void appendHeaders(StringBuilder buf) {
+ for (Map.Entry e: getHeaders()) {
+ buf.append(e.getKey());
+ buf.append(": ");
+ buf.append(e.getValue());
+ buf.append(StringUtil.NEWLINE);
+ }
+ }
}
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpMessage.java b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpMessage.java
index a0697aa12f..09d79e0a3a 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpMessage.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpMessage.java
@@ -26,7 +26,7 @@ import java.util.Set;
/**
* The default {@link HttpMessage} implementation.
*/
-public class DefaultHttpMessage implements HttpMessage {
+public class DefaultHttpMessage extends DefaultHttpObject implements HttpMessage {
private final HttpHeaders headers = new HttpHeaders();
private HttpVersion version;
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpObject.java b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpObject.java
new file mode 100644
index 0000000000..232eeecea1
--- /dev/null
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpObject.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 io.netty.handler.codec.http;
+
+import io.netty.handler.codec.DecoderResult;
+
+public class DefaultHttpObject implements HttpObject {
+
+ private DecoderResult decodeResult = DecoderResult.SUCCESS;
+
+ protected DefaultHttpObject() {
+ // Disallow direct instantiation
+ }
+
+ @Override
+ public DecoderResult getDecoderResult() {
+ return decodeResult;
+ }
+
+ @Override
+ public void setDecoderResult(DecoderResult result) {
+ if (result == null) {
+ throw new NullPointerException("result");
+ }
+ decodeResult = result;
+ }
+}
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpRequest.java b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpRequest.java
index afadec21bd..134be89901 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpRequest.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpRequest.java
@@ -70,6 +70,8 @@ public class DefaultHttpRequest extends DefaultHttpMessage implements HttpReques
buf.append(getClass().getSimpleName());
buf.append("(transferEncoding: ");
buf.append(getTransferEncoding());
+ buf.append(", decodeResult: ");
+ buf.append(getDecoderResult());
buf.append(')');
buf.append(StringUtil.NEWLINE);
buf.append(getMethod().toString());
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpResponse.java b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpResponse.java
index 440b34fed0..9872868ae1 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpResponse.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpResponse.java
@@ -54,6 +54,8 @@ public class DefaultHttpResponse extends DefaultHttpMessage implements HttpRespo
buf.append(getClass().getSimpleName());
buf.append("(transferEncoding: ");
buf.append(getTransferEncoding());
+ buf.append(", decodeResult: ");
+ buf.append(getDecoderResult());
buf.append(')');
buf.append(StringUtil.NEWLINE);
buf.append(getProtocolVersion().getText());
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunk.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunk.java
index 4157fc91ec..dbf5e3917b 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunk.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunk.java
@@ -18,6 +18,7 @@ package io.netty.handler.codec.http;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.DecoderResult;
import java.util.Collections;
import java.util.List;
@@ -33,7 +34,7 @@ import java.util.Set;
* {@link ChannelPipeline}.
* @apiviz.landmark
*/
-public interface HttpChunk {
+public interface HttpChunk extends HttpObject {
/**
* The 'end of content' marker in chunked encoding.
@@ -103,6 +104,16 @@ public interface HttpChunk {
public void setHeader(String name, Iterable> values) {
throw new IllegalStateException("read-only");
}
+
+ @Override
+ public DecoderResult getDecoderResult() {
+ return DecoderResult.SUCCESS;
+ }
+
+ @Override
+ public void setDecoderResult(DecoderResult result) {
+ throw new IllegalStateException("read-only");
+ }
};
/**
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java
index e175ffcdb9..f32963f308 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java
@@ -22,6 +22,7 @@ import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.TooLongFrameException;
import io.netty.util.CharsetUtil;
@@ -46,7 +47,7 @@ import java.util.Map.Entry;
* @apiviz.landmark
* @apiviz.has io.netty.handler.codec.http.HttpChunk oneway - - filters out
*/
-public class HttpChunkAggregator extends MessageToMessageDecoder
@@ -378,10 +378,40 @@
-
+
+ maven-clean-plugin
+ 2.5
+
+
+ maven-resources-plugin
+ 2.5
+
+
+ maven-jar-plugin
+ 2.4
+
+
+ maven-dependency-plugin
+ 2.4
+
+
+ maven-assembly-plugin
+ 2.3
+
+
+ maven-jxr-plugin
+ 2.2
+
+
+ org.codehaus.mojo
+ build-helper-maven-plugin
+ 1.7
+
+
+
org.eclipse.m2e
lifecycle-mapping
diff --git a/tarball/pom.xml b/tarball/pom.xml
index 880371f9aa..a6bf2c7ee3 100644
--- a/tarball/pom.xml
+++ b/tarball/pom.xml
@@ -20,7 +20,7 @@
io.netty
netty-parent
- 4.0.0.Alpha5-SNAPSHOT
+ 4.0.0.Alpha6-SNAPSHOT
netty-tarball
@@ -51,7 +51,6 @@
maven-clean-plugin
- 2.5
clean-first
@@ -87,7 +86,6 @@
maven-deploy-plugin
- 2.7
true
@@ -102,7 +100,6 @@
maven-dependency-plugin
- 2.4
copy-jars
@@ -158,7 +155,6 @@
maven-assembly-plugin
- 2.3
build-tarball
@@ -177,11 +173,9 @@
-
-
diff --git a/testsuite/pom.xml b/testsuite/pom.xml
index bd7b5e5c23..76a92ed472 100644
--- a/testsuite/pom.xml
+++ b/testsuite/pom.xml
@@ -20,7 +20,7 @@
io.netty
netty-parent
- 4.0.0.Alpha5-SNAPSHOT
+ 4.0.0.Alpha6-SNAPSHOT
netty-testsuite
@@ -45,7 +45,6 @@
maven-deploy-plugin
- 2.7
true
diff --git a/testsuite/src/test/java/io/netty/testsuite/transport/socket/ServerSocketSuspendTest.java b/testsuite/src/test/java/io/netty/testsuite/transport/socket/ServerSocketSuspendTest.java
index cfcd8db9d6..41124a5e1c 100644
--- a/testsuite/src/test/java/io/netty/testsuite/transport/socket/ServerSocketSuspendTest.java
+++ b/testsuite/src/test/java/io/netty/testsuite/transport/socket/ServerSocketSuspendTest.java
@@ -72,6 +72,8 @@ public class ServerSocketSuspendTest extends AbstractServerSocketTest {
}
}
+ Thread.sleep(TIMEOUT / 1000000);
+
try {
long startTime = System.nanoTime();
for (int i = 0; i < NUM_CHANNELS; i ++) {
diff --git a/testsuite/src/test/java/io/netty/testsuite/util/TestUtils.java b/testsuite/src/test/java/io/netty/testsuite/util/TestUtils.java
index 33c8242c01..38e2f8ac78 100644
--- a/testsuite/src/test/java/io/netty/testsuite/util/TestUtils.java
+++ b/testsuite/src/test/java/io/netty/testsuite/util/TestUtils.java
@@ -15,13 +15,38 @@
*/
package io.netty.testsuite.util;
+import io.netty.util.NetworkConstants;
+
import java.io.IOException;
+import java.net.InetSocketAddress;
import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
public class TestUtils {
- private static int START_PORT = 20000;
- private static int END_PORT = 30000;
+ private static final int START_PORT = 32768;
+ private static final int END_PORT = 65536;
+ private static final int NUM_CANDIDATES = END_PORT - START_PORT;
+
+ private static final List PORTS = new ArrayList();
+ private static Iterator PORTS_ITERATOR;
+
+ static {
+ for (int i = START_PORT; i < END_PORT; i ++) {
+ PORTS.add(i);
+ }
+ Collections.shuffle(PORTS);
+ }
+
+ private static int nextCandidatePort() {
+ if (PORTS_ITERATOR == null || !PORTS_ITERATOR.hasNext()) {
+ PORTS_ITERATOR = PORTS.iterator();
+ }
+ return PORTS_ITERATOR.next();
+ }
/**
* Return a free port which can be used to bind to
@@ -29,19 +54,28 @@ public class TestUtils {
* @return port
*/
public static int getFreePort() {
- for(int start = START_PORT; start <= END_PORT; start++) {
+ for (int i = 0; i < NUM_CANDIDATES; i ++) {
+ int port = nextCandidatePort();
try {
- ServerSocket socket = new ServerSocket(start);
- socket.setReuseAddress(true);
- socket.close();
- START_PORT = start + 1;
- return start;
+ // Ensure it is possible to bind on both wildcard and loopback.
+ ServerSocket ss;
+ ss = new ServerSocket();
+ ss.setReuseAddress(false);
+ ss.bind(new InetSocketAddress(port));
+ ss.close();
+
+ ss = new ServerSocket();
+ ss.setReuseAddress(false);
+ ss.bind(new InetSocketAddress(NetworkConstants.LOCALHOST, port));
+ ss.close();
+
+ return port;
} catch (IOException e) {
// ignore
}
-
}
- throw new RuntimeException("Unable to find a free port....");
+
+ throw new RuntimeException("unable to find a free port");
}
private TestUtils() { }
diff --git a/transport/pom.xml b/transport/pom.xml
index e04bb88f74..e5733faf00 100644
--- a/transport/pom.xml
+++ b/transport/pom.xml
@@ -20,7 +20,7 @@
io.netty
netty-parent
- 4.0.0.Alpha5-SNAPSHOT
+ 4.0.0.Alpha6-SNAPSHOT
netty-transport
diff --git a/transport/src/main/java/com/sun/nio/sctp/SctpServerChannel.java b/transport/src/main/java/com/sun/nio/sctp/SctpServerChannel.java
index eaf617e5f7..2b47d1daf3 100644
--- a/transport/src/main/java/com/sun/nio/sctp/SctpServerChannel.java
+++ b/transport/src/main/java/com/sun/nio/sctp/SctpServerChannel.java
@@ -30,13 +30,13 @@ public abstract class SctpServerChannel extends AbstractSelectableChannel {
public static SctpServerChannel open() throws IOException {
return null;
}
-
+
protected SctpServerChannel(SelectorProvider provider) {
super(provider);
}
public abstract T getOption(SctpSocketOption name) throws IOException;
- public abstract SctpChannel setOption(SctpSocketOption name, T value) throws IOException;
+ public abstract SctpServerChannel setOption(SctpSocketOption name, T value) throws IOException;
public abstract Set getAllLocalAddresses() throws IOException;
diff --git a/transport/src/main/java/io/netty/bootstrap/ServerBootstrap.java b/transport/src/main/java/io/netty/bootstrap/ServerBootstrap.java
index c57b4b2afc..c7fce485bc 100644
--- a/transport/src/main/java/io/netty/bootstrap/ServerBootstrap.java
+++ b/transport/src/main/java/io/netty/bootstrap/ServerBootstrap.java
@@ -17,17 +17,16 @@ package io.netty.bootstrap;
import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
-
import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelInboundMessageHandler;
import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInboundMessageHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ServerChannel;
import io.netty.channel.socket.SocketChannel;
@@ -232,6 +231,7 @@ public class ServerBootstrap extends AbstractBootstrap {
try {
childGroup.register(child);
} catch (Throwable t) {
+ child.unsafe().closeForcibly();
logger.warn("Failed to register an accepted channel: " + child, t);
}
}
diff --git a/transport/src/main/java/io/netty/channel/AbstractChannel.java b/transport/src/main/java/io/netty/channel/AbstractChannel.java
index 46096f882d..49d854f82c 100644
--- a/transport/src/main/java/io/netty/channel/AbstractChannel.java
+++ b/transport/src/main/java/io/netty/channel/AbstractChannel.java
@@ -547,6 +547,15 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
}
}
+ @Override
+ public final void closeForcibly() {
+ try {
+ doClose();
+ } catch (Exception e) {
+ logger.warn("Failed to close a channel.", e);
+ }
+ }
+
@Override
public final void deregister(final ChannelFuture future) {
if (eventLoop().inEventLoop()) {
diff --git a/transport/src/main/java/io/netty/channel/Channel.java b/transport/src/main/java/io/netty/channel/Channel.java
index 788daedce8..ea0d3e6969 100755
--- a/transport/src/main/java/io/netty/channel/Channel.java
+++ b/transport/src/main/java/io/netty/channel/Channel.java
@@ -248,6 +248,12 @@ public interface Channel extends AttributeMap, ChannelOutboundInvoker, ChannelFu
*/
void close(ChannelFuture future);
+ /**
+ * Closes the {@link Channel} immediately without firing any events. Probably only useful
+ * when registration attempt failed.
+ */
+ void closeForcibly();
+
/**
* Deregister the {@link Channel} of the {@link ChannelFuture} from {@link EventLoop} and notify the
* {@link ChannelFuture} once the operation was complete.
diff --git a/transport/src/main/java/io/netty/channel/ChannelInitializer.java b/transport/src/main/java/io/netty/channel/ChannelInitializer.java
index 0ec4da0a17..a347e75eda 100644
--- a/transport/src/main/java/io/netty/channel/ChannelInitializer.java
+++ b/transport/src/main/java/io/netty/channel/ChannelInitializer.java
@@ -26,6 +26,7 @@ public abstract class ChannelInitializer extends ChannelState
public abstract void initChannel(C ch) throws Exception;
+ @SuppressWarnings("unchecked")
@Override
public final void channelRegistered(ChannelHandlerContext ctx)
throws Exception {
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java b/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
index 9bb9b6f690..b7fd02ed60 100644
--- a/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
@@ -48,6 +48,7 @@ public class DefaultChannelConfig implements ChannelConfig {
return result;
}
+ @SuppressWarnings("unchecked")
@Override
public boolean setOptions(Map, ?> options) {
if (options == null) {
@@ -64,6 +65,7 @@ public class DefaultChannelConfig implements ChannelConfig {
return setAllOptions;
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == null) {
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
index bf6d4aea66..6231ac7796 100755
--- a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
@@ -388,6 +388,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
return remove(getContextOrDie(name)).handler();
}
+ @SuppressWarnings("unchecked")
@Override
public T remove(Class handlerType) {
return (T) remove(getContextOrDie(handlerType)).handler();
@@ -784,6 +785,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
}
}
+ @SuppressWarnings("unchecked")
@Override
public T get(Class handlerType) {
ChannelHandlerContext ctx = context(handlerType);
diff --git a/transport/src/main/java/io/netty/channel/local/LocalChannel.java b/transport/src/main/java/io/netty/channel/local/LocalChannel.java
index c36db6b588..0b02e1290f 100755
--- a/transport/src/main/java/io/netty/channel/local/LocalChannel.java
+++ b/transport/src/main/java/io/netty/channel/local/LocalChannel.java
@@ -23,6 +23,7 @@ import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelMetadata;
+import io.netty.channel.ChannelPipeline;
import io.netty.channel.DefaultChannelConfig;
import io.netty.channel.EventLoop;
import io.netty.channel.SingleThreadEventExecutor;
@@ -33,6 +34,7 @@ import java.nio.channels.AlreadyConnectedException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ConnectionPendingException;
import java.nio.channels.NotYetConnectedException;
+import java.util.Collections;
/**
* A {@link Channel} for the local transport.
@@ -215,16 +217,24 @@ public class LocalChannel extends AbstractChannel {
}
final LocalChannel peer = this.peer;
- assert peer != null;
+ final ChannelPipeline peerPipeline = peer.pipeline();
+ final EventLoop peerLoop = peer.eventLoop();
- buf.drainTo(peer.pipeline().inboundMessageBuffer());
-
- peer.eventLoop().execute(new Runnable() {
- @Override
- public void run() {
- peer.pipeline().fireInboundBufferUpdated();
- }
- });
+ if (peerLoop == eventLoop()) {
+ buf.drainTo(peerPipeline.inboundMessageBuffer());
+ peerPipeline.fireInboundBufferUpdated();
+ } else {
+ final Object[] msgs = buf.toArray();
+ buf.clear();
+ peerLoop.execute(new Runnable() {
+ @Override
+ public void run() {
+ MessageBuf buf = peerPipeline.inboundMessageBuffer();
+ Collections.addAll(buf, msgs);
+ peerPipeline.fireInboundBufferUpdated();
+ }
+ });
+ }
}
@Override
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
index 665196a1b4..f1f9a1c2b5 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
@@ -61,6 +61,7 @@ public class DefaultDatagramChannelConfig extends DefaultChannelConfig implement
IP_MULTICAST_ADDR, IP_MULTICAST_IF, IP_MULTICAST_TTL, IP_TOS, UDP_RECEIVE_PACKET_SIZE);
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == SO_BROADCAST) {
@@ -82,12 +83,10 @@ public class DefaultDatagramChannelConfig extends DefaultChannelConfig implement
return (T) Boolean.valueOf(isLoopbackModeDisabled());
}
if (option == IP_MULTICAST_ADDR) {
- @SuppressWarnings("unchecked")
T i = (T) getInterface();
return i;
}
if (option == IP_MULTICAST_IF) {
- @SuppressWarnings("unchecked")
T i = (T) getNetworkInterface();
return i;
}
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultSctpChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultSctpChannelConfig.java
index 03ab6d9ad0..fa0dbae987 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultSctpChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultSctpChannelConfig.java
@@ -48,6 +48,7 @@ public class DefaultSctpChannelConfig extends DefaultChannelConfig implements Sc
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == SO_RCVBUF) {
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultSctpServerChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultSctpServerChannelConfig.java
index c5d2c44bb5..fb7a87568a 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultSctpServerChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultSctpServerChannelConfig.java
@@ -28,7 +28,6 @@ import java.util.Map;
import static com.sun.nio.sctp.SctpStandardSocketOptions.SCTP_INIT_MAXSTREAMS;
import static com.sun.nio.sctp.SctpStandardSocketOptions.SO_RCVBUF;
import static com.sun.nio.sctp.SctpStandardSocketOptions.SO_SNDBUF;
-import static io.netty.channel.ChannelOption.SCTP_NODELAY;
/**
* The default {@link SctpServerChannelConfig} implementation for SCTP.
@@ -56,6 +55,7 @@ public class DefaultSctpServerChannelConfig extends DefaultChannelConfig impleme
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == ChannelOption.SO_RCVBUF) {
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java
index 71ada68669..2d75c4fcf0 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java
@@ -49,6 +49,7 @@ public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
return getOptions(super.getOptions(), SO_RCVBUF, SO_REUSEADDR, SO_BACKLOG);
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == SO_RCVBUF) {
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java
index e900b68fee..9c1521e4a2 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java
@@ -51,6 +51,7 @@ public class DefaultSocketChannelConfig extends DefaultChannelConfig
ALLOW_HALF_CLOSURE);
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == SO_RCVBUF) {
diff --git a/transport/src/main/java/io/netty/channel/socket/SctpNotificationEvent.java b/transport/src/main/java/io/netty/channel/socket/SctpNotificationEvent.java
index 3fefa73162..b9a97416f3 100644
--- a/transport/src/main/java/io/netty/channel/socket/SctpNotificationEvent.java
+++ b/transport/src/main/java/io/netty/channel/socket/SctpNotificationEvent.java
@@ -27,10 +27,17 @@ public final class SctpNotificationEvent {
this.attachment = attachment;
}
+ /**
+ * Return the {@link Notification}
+ */
public Notification notification() {
return notification;
}
+ /**
+ * Return the attachment of this {@link SctpNotificationEvent}, or
+ * null
if no attachment was provided
+ */
public Object attachment() {
return attachment;
}
diff --git a/transport/src/main/java/io/netty/channel/socket/aio/AbstractAioChannel.java b/transport/src/main/java/io/netty/channel/socket/aio/AbstractAioChannel.java
index ac31001d50..7b7dc8e136 100755
--- a/transport/src/main/java/io/netty/channel/socket/aio/AbstractAioChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/aio/AbstractAioChannel.java
@@ -68,7 +68,7 @@ abstract class AbstractAioChannel extends AbstractChannel {
@Override
protected Runnable doRegister() throws Exception {
- if (((AioEventLoop) eventLoop()).parent() != group) {
+ if (eventLoop().parent() != group) {
throw new ChannelException(
getClass().getSimpleName() + " must be registered to the " +
AioEventLoopGroup.class.getSimpleName() + " which was specified in the constructor.");
diff --git a/transport/src/main/java/io/netty/channel/socket/aio/AioServerSocketChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/aio/AioServerSocketChannelConfig.java
index 990e8172b5..d23dae7241 100644
--- a/transport/src/main/java/io/netty/channel/socket/aio/AioServerSocketChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/aio/AioServerSocketChannelConfig.java
@@ -45,6 +45,7 @@ final class AioServerSocketChannelConfig extends DefaultChannelConfig
return getOptions(super.getOptions(), SO_RCVBUF, SO_REUSEADDR, SO_BACKLOG);
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == SO_RCVBUF) {
diff --git a/transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannelConfig.java
index 3be37ea7a2..a03acb25d4 100644
--- a/transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannelConfig.java
@@ -57,6 +57,7 @@ final class AioSocketChannelConfig extends DefaultChannelConfig
AIO_READ_TIMEOUT, AIO_WRITE_TIMEOUT, ALLOW_HALF_CLOSURE);
}
+ @SuppressWarnings("unchecked")
@Override
public T getOption(ChannelOption option) {
if (option == SO_RCVBUF) {
diff --git a/transport/src/main/java/io/netty/channel/socket/nio/SelectorUtil.java b/transport/src/main/java/io/netty/channel/socket/nio/SelectorUtil.java
index 459fa06a8b..e47802ac62 100644
--- a/transport/src/main/java/io/netty/channel/socket/nio/SelectorUtil.java
+++ b/transport/src/main/java/io/netty/channel/socket/nio/SelectorUtil.java
@@ -27,7 +27,7 @@ import java.util.concurrent.TimeUnit;
final class SelectorUtil {
private static final InternalLogger logger =
InternalLoggerFactory.getInstance(SelectorUtil.class);
- static final long DEFAULT_SELECT_TIMEOUT = 10;
+ static final long DEFAULT_SELECT_TIMEOUT = 500;
static final long SELECT_TIMEOUT =
SystemPropertyUtil.getLong("io.netty.selectTimeout", DEFAULT_SELECT_TIMEOUT);
static final long SELECT_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(SELECT_TIMEOUT);
diff --git a/transport/src/main/java/io/netty/channel/socket/oio/OioEventLoopGroup.java b/transport/src/main/java/io/netty/channel/socket/oio/OioEventLoopGroup.java
index d2232cadfe..75e65d8056 100644
--- a/transport/src/main/java/io/netty/channel/socket/oio/OioEventLoopGroup.java
+++ b/transport/src/main/java/io/netty/channel/socket/oio/OioEventLoopGroup.java
@@ -19,9 +19,9 @@ package io.netty.channel.socket.oio;
import io.netty.channel.Channel;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelTaskScheduler;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
-import io.netty.channel.ChannelTaskScheduler;
import java.util.Collections;
import java.util.Queue;
diff --git a/transport/src/main/java/io/netty/channel/socket/oio/OioSctpChannel.java b/transport/src/main/java/io/netty/channel/socket/oio/OioSctpChannel.java
index 586e885cc2..feb1e904f6 100755
--- a/transport/src/main/java/io/netty/channel/socket/oio/OioSctpChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/oio/OioSctpChannel.java
@@ -52,7 +52,7 @@ public class OioSctpChannel extends AbstractOioMessageChannel
private final SctpChannel ch;
private final SctpChannelConfig config;
- private final NotificationHandler notificationHandler;
+ private final NotificationHandler> notificationHandler;
private static SctpChannel openChannel() {
try {
diff --git a/transport/src/main/java/io/netty/channel/socket/oio/OioSctpServerChannel.java b/transport/src/main/java/io/netty/channel/socket/oio/OioSctpServerChannel.java
index 5f31c9a507..11d0ebe30d 100755
--- a/transport/src/main/java/io/netty/channel/socket/oio/OioSctpServerChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/oio/OioSctpServerChannel.java
@@ -21,9 +21,7 @@ import io.netty.buffer.ChannelBufType;
import io.netty.buffer.MessageBuf;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelMetadata;
-import io.netty.channel.socket.DefaultSctpChannelConfig;
import io.netty.channel.socket.DefaultSctpServerChannelConfig;
-import io.netty.channel.socket.SctpNotificationHandler;
import io.netty.channel.socket.SctpServerChannelConfig;
import io.netty.logging.InternalLogger;
import io.netty.logging.InternalLoggerFactory;
diff --git a/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest2.java b/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest2.java
new file mode 100644
index 0000000000..87fdfe41ac
--- /dev/null
+++ b/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest2.java
@@ -0,0 +1,153 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 io.netty.channel.local;
+
+import static org.junit.Assert.*;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.MessageBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundMessageHandlerAdapter;
+
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+public class LocalTransportThreadModelTest2 {
+
+ private static final String LOCAL_CHANNEL = LocalTransportThreadModelTest2.class.getName();
+
+ static final int messageCountPerRun = 4;
+
+ @Test(timeout = 15000)
+ public void testSocketReuse() throws InterruptedException {
+ ServerBootstrap serverBootstrap = new ServerBootstrap();
+ LocalHander serverHandler = new LocalHander("SERVER");
+ serverBootstrap
+ .group(new LocalEventLoopGroup(), new LocalEventLoopGroup())
+ .channel(LocalServerChannel.class)
+ .localAddress(new LocalAddress(LOCAL_CHANNEL))
+ .childHandler(serverHandler);
+
+ Bootstrap clientBootstrap = new Bootstrap();
+ LocalHander clientHandler = new LocalHander("CLIENT");
+ clientBootstrap
+ .group(new LocalEventLoopGroup())
+ .channel(LocalChannel.class)
+ .remoteAddress(new LocalAddress(LOCAL_CHANNEL)).handler(clientHandler);
+
+ serverBootstrap.bind().sync();
+
+ int count = 100;
+ for (int i = 1; i < count + 1; i ++) {
+ Channel ch = clientBootstrap.connect().sync().channel();
+
+ // SPIN until we get what we are looking for.
+ int target = i * messageCountPerRun;
+ while (serverHandler.count.get() != target || clientHandler.count.get() != target) {
+ Thread.sleep(50);
+ }
+ close(ch, clientHandler);
+ }
+
+ assertEquals(count * 2 * messageCountPerRun, serverHandler.count.get() +
+ clientHandler.count.get());
+ }
+
+ public void close(final Channel localChannel, final LocalHander localRegistrationHandler) {
+ // we want to make sure we actually shutdown IN the event loop
+ if (localChannel.eventLoop().inEventLoop()) {
+ MessageBuf outboundMessageBuffer =
+ localChannel.pipeline().outboundMessageBuffer();
+ if (!outboundMessageBuffer.isEmpty()) {
+ System.err.println("NOT EMPTY TO SEND!");
+ }
+
+ // Wait until all messages are flushed before closing the channel.
+ if (localRegistrationHandler.lastWriteFuture != null) {
+ localRegistrationHandler.lastWriteFuture.awaitUninterruptibly();
+ }
+
+ MessageBuf inboundMessageBuffer =
+ localChannel.pipeline().inboundMessageBuffer();
+ if (!inboundMessageBuffer.isEmpty()) {
+ // sometimes we close the pipeline before everything on it has been notified/received.
+ // we want these messages, since they are in our queue.
+ Iterator iterator = inboundMessageBuffer.iterator();
+ while (iterator.hasNext()) {
+ Object next = iterator.next();
+ System.err.println("DEFERRED on close: " + next);
+ iterator.remove();
+ }
+ }
+
+ localChannel.close();
+ return;
+ } else {
+ localChannel.eventLoop().execute(new Runnable() {
+ @Override
+ public void run() {
+ close(localChannel, localRegistrationHandler);
+ }
+ });
+ }
+
+ // Wait until the connection is closed or the connection attempt fails.
+ localChannel.closeFuture().awaitUninterruptibly();
+
+ MessageBuf inboundMessageBuffer = localChannel.pipeline().inboundMessageBuffer();
+ if (!inboundMessageBuffer.isEmpty()) {
+ // sometimes we close the pipeline before everything on it has been notified/received.
+ // we want these messages, since they are in our queue.
+ Iterator iterator = inboundMessageBuffer.iterator();
+ while (iterator.hasNext()) {
+ Object next = iterator.next();
+ System.err.println("DEFERRED on close: " + next);
+ iterator.remove();
+ }
+ }
+ }
+
+ @Sharable
+ class LocalHander extends ChannelInboundMessageHandlerAdapter {
+ private final String name;
+
+ public volatile ChannelFuture lastWriteFuture = null;
+
+ public AtomicInteger count = new AtomicInteger(0);
+
+ public LocalHander(String name) {
+ this.name = name;
+
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ for (int i = 0; i < messageCountPerRun; i ++) {
+ lastWriteFuture = ctx.channel().write(name + " " + i);
+ }
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
+ count.incrementAndGet();
+ }
+ }
+}
\ No newline at end of file