andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Convert DOS line ending to UNIX line ending

Browse Source
This commit is contained in:
Trustin Lee 2012-05-15 17:14:02 +09:00
parent dd2e36e5d9
commit 894ececbb7
25 changed files with 3629 additions and 3629 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

268
codec/src/test/java/io/netty/handler/codec/redis/RedisCodecTest.java
View File

@ -1,134 +1,134 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.codec.redis; package io.netty.handler.codec.redis;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers; import io.netty.buffer.ChannelBuffers;
import io.netty.handler.codec.embedder.DecoderEmbedder; import io.netty.handler.codec.embedder.DecoderEmbedder;
import io.netty.util.CharsetUtil; import io.netty.util.CharsetUtil;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import java.io.IOException; import java.io.IOException;
import static io.netty.buffer.ChannelBuffers.wrappedBuffer; import static io.netty.buffer.ChannelBuffers.wrappedBuffer;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull; import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
public class RedisCodecTest { public class RedisCodecTest {
private DecoderEmbedder<ChannelBuffer> embedder; private DecoderEmbedder<ChannelBuffer> embedder;
@Before @Before
public void setUp() { public void setUp() {
embedder = new DecoderEmbedder<ChannelBuffer>(new RedisDecoder()); embedder = new DecoderEmbedder<ChannelBuffer>(new RedisDecoder());
} }
@Test @Test
public void decodeReplies() throws IOException { public void decodeReplies() throws IOException {
{ {
Object receive = decode("+OK\r\n".getBytes()); Object receive = decode("+OK\r\n".getBytes());
assertTrue(receive instanceof StatusReply); assertTrue(receive instanceof StatusReply);
assertEquals("OK", ((StatusReply) receive).status.toString(CharsetUtil.UTF_8)); assertEquals("OK", ((StatusReply) receive).status.toString(CharsetUtil.UTF_8));
} }
{ {
Object receive = decode("-ERROR\r\n".getBytes()); Object receive = decode("-ERROR\r\n".getBytes());
assertTrue(receive instanceof ErrorReply); assertTrue(receive instanceof ErrorReply);
assertEquals("ERROR", ((ErrorReply) receive).error.toString(CharsetUtil.UTF_8)); assertEquals("ERROR", ((ErrorReply) receive).error.toString(CharsetUtil.UTF_8));
} }
{ {
Object receive = decode(":123\r\n".getBytes()); Object receive = decode(":123\r\n".getBytes());
assertTrue(receive instanceof IntegerReply); assertTrue(receive instanceof IntegerReply);
assertEquals(123, ((IntegerReply) receive).integer); assertEquals(123, ((IntegerReply) receive).integer);
} }
{ {
Object receive = decode("$5\r\nnetty\r\n".getBytes()); Object receive = decode("$5\r\nnetty\r\n".getBytes());
assertTrue(receive instanceof BulkReply); assertTrue(receive instanceof BulkReply);
assertEquals("netty", new String(((BulkReply) receive).bytes)); assertEquals("netty", new String(((BulkReply) receive).bytes));
} }
{ {
Object receive = decode("*2\r\n$5\r\nnetty\r\n$5\r\nrules\r\n".getBytes()); Object receive = decode("*2\r\n$5\r\nnetty\r\n$5\r\nrules\r\n".getBytes());
assertTrue(receive instanceof MultiBulkReply); assertTrue(receive instanceof MultiBulkReply);
assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0])); assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0]));
assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1])); assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1]));
} }
} }
private Object decode(byte[] bytes) { private Object decode(byte[] bytes) {
embedder.offer(wrappedBuffer(bytes)); embedder.offer(wrappedBuffer(bytes));
return embedder.poll(); return embedder.poll();
} }
@Test @Test
public void encodeCommands() throws IOException { public void encodeCommands() throws IOException {
String setCommand = "*3\r\n" + String setCommand = "*3\r\n" +
"$3\r\n" + "$3\r\n" +
"SET\r\n" + "SET\r\n" +
"$5\r\n" + "$5\r\n" +
"mykey\r\n" + "mykey\r\n" +
"$7\r\n" + "$7\r\n" +
"myvalue\r\n"; "myvalue\r\n";
Command command = new Command("SET", "mykey", "myvalue"); Command command = new Command("SET", "mykey", "myvalue");
ChannelBuffer cb = ChannelBuffers.dynamicBuffer(); ChannelBuffer cb = ChannelBuffers.dynamicBuffer();
command.write(cb); command.write(cb);
assertEquals(setCommand, cb.toString(CharsetUtil.US_ASCII)); assertEquals(setCommand, cb.toString(CharsetUtil.US_ASCII));
} }
@Test @Test
public void testReplayDecoding() { public void testReplayDecoding() {
{ {
embedder.offer(wrappedBuffer("*2\r\n$5\r\nnetty\r\n".getBytes())); embedder.offer(wrappedBuffer("*2\r\n$5\r\nnetty\r\n".getBytes()));
Object receive = embedder.poll(); Object receive = embedder.poll();
assertNull(receive); assertNull(receive);
embedder.offer(wrappedBuffer("$5\r\nrules\r\n".getBytes())); embedder.offer(wrappedBuffer("$5\r\nrules\r\n".getBytes()));
receive = embedder.poll(); receive = embedder.poll();
assertTrue(receive instanceof MultiBulkReply); assertTrue(receive instanceof MultiBulkReply);
assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0])); assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0]));
assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1])); assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1]));
} }
{ {
embedder.offer(wrappedBuffer("*2\r\n$5\r\nnetty\r\n$5\r\nr".getBytes())); embedder.offer(wrappedBuffer("*2\r\n$5\r\nnetty\r\n$5\r\nr".getBytes()));
Object receive = embedder.poll(); Object receive = embedder.poll();
assertNull(receive); assertNull(receive);
embedder.offer(wrappedBuffer("ules\r\n".getBytes())); embedder.offer(wrappedBuffer("ules\r\n".getBytes()));
receive = embedder.poll(); receive = embedder.poll();
assertTrue(receive instanceof MultiBulkReply); assertTrue(receive instanceof MultiBulkReply);
assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0])); assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0]));
assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1])); assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1]));
} }
{ {
embedder.offer(wrappedBuffer("*2".getBytes())); embedder.offer(wrappedBuffer("*2".getBytes()));
Object receive = embedder.poll(); Object receive = embedder.poll();
assertNull(receive); assertNull(receive);
embedder.offer(wrappedBuffer("\r\n$5\r\nnetty\r\n$5\r\nrules\r\n".getBytes())); embedder.offer(wrappedBuffer("\r\n$5\r\nnetty\r\n$5\r\nrules\r\n".getBytes()));
receive = embedder.poll(); receive = embedder.poll();
assertTrue(receive instanceof MultiBulkReply); assertTrue(receive instanceof MultiBulkReply);
assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0])); assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0]));
assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1])); assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1]));
} }
{ {
embedder.offer(wrappedBuffer("*2\r\n$5\r\nnetty\r\n$5\r\nrules\r".getBytes())); embedder.offer(wrappedBuffer("*2\r\n$5\r\nnetty\r\n$5\r\nrules\r".getBytes()));
Object receive = embedder.poll(); Object receive = embedder.poll();
assertNull(receive); assertNull(receive);
embedder.offer(wrappedBuffer("\n".getBytes())); embedder.offer(wrappedBuffer("\n".getBytes()));
receive = embedder.poll(); receive = embedder.poll();
assertTrue(receive instanceof MultiBulkReply); assertTrue(receive instanceof MultiBulkReply);
assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0])); assertEquals("netty", new String((byte[]) ((MultiBulkReply) receive).byteArrays[0]));
assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1])); assertEquals("rules", new String((byte[]) ((MultiBulkReply) receive).byteArrays[1]));
} }
} }
} }

154
common/src/main/java/io/netty/util/internal/DetectionUtil.java
View File

@ -1,77 +1,77 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.util.internal; package io.netty.util.internal;
import java.security.AccessController; import java.security.AccessController;
import java.security.PrivilegedActionException; import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction; import java.security.PrivilegedExceptionAction;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.Deflater; import java.util.zip.Deflater;
/** /**
* Utility that detects various properties specific to the current runtime * Utility that detects various properties specific to the current runtime
* environment, such as Java version and the availability of the * environment, such as Java version and the availability of the
* {@code sun.misc.Unsafe} object. * {@code sun.misc.Unsafe} object.
*/ */
public final class DetectionUtil { public final class DetectionUtil {
private static final int JAVA_VERSION = javaVersion0(); private static final int JAVA_VERSION = javaVersion0();
private static final boolean HAS_UNSAFE = hasUnsafe(AtomicInteger.class.getClassLoader()); private static final boolean HAS_UNSAFE = hasUnsafe(AtomicInteger.class.getClassLoader());
public static boolean hasUnsafe() { public static boolean hasUnsafe() {
return HAS_UNSAFE; return HAS_UNSAFE;
} }
public static int javaVersion() { public static int javaVersion() {
return JAVA_VERSION; return JAVA_VERSION;
} }
private static boolean hasUnsafe(ClassLoader loader) { private static boolean hasUnsafe(ClassLoader loader) {
try { try {
Class<?> unsafeClazz = Class.forName("sun.misc.Unsafe", true, loader); Class<?> unsafeClazz = Class.forName("sun.misc.Unsafe", true, loader);
return hasUnsafeField(unsafeClazz); return hasUnsafeField(unsafeClazz);
} catch (Exception e) { } catch (Exception e) {
// Ignore // Ignore
} }
return false; return false;
} }
private static boolean hasUnsafeField(final Class<?> unsafeClass) throws PrivilegedActionException { private static boolean hasUnsafeField(final Class<?> unsafeClass) throws PrivilegedActionException {
return AccessController.doPrivileged(new PrivilegedExceptionAction<Boolean>() { return AccessController.doPrivileged(new PrivilegedExceptionAction<Boolean>() {
@Override @Override
public Boolean run() throws Exception { public Boolean run() throws Exception {
unsafeClass.getDeclaredField("theUnsafe"); unsafeClass.getDeclaredField("theUnsafe");
return true; return true;
} }
}); });
} }
private static int javaVersion0() { private static int javaVersion0() {
try { try {
Deflater.class.getDeclaredField("SYNC_FLUSH"); Deflater.class.getDeclaredField("SYNC_FLUSH");
return 7; return 7;
} catch (Exception e) { } catch (Exception e) {
// Ignore // Ignore
} }
return 6; return 6;
} }
private DetectionUtil() { private DetectionUtil() {
// only static method supported // only static method supported
} }
} }

2752
common/src/main/java/io/netty/util/internal/LinkedTransferQueue.java
View File

File diff suppressed because it is too large Load Diff

124
common/src/main/java/io/netty/util/internal/QueueFactory.java
View File

@ -1,62 +1,62 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.util.internal; package io.netty.util.internal;
import java.util.Collection; import java.util.Collection;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
/** /**
* This factory should be used to create the "optimal" {@link BlockingQueue} * This factory should be used to create the "optimal" {@link BlockingQueue}
* instance for the running JVM. * instance for the running JVM.
*/ */
public final class QueueFactory { public final class QueueFactory {
private static final boolean useUnsafe = DetectionUtil.hasUnsafe(); private static final boolean useUnsafe = DetectionUtil.hasUnsafe();
private QueueFactory() { private QueueFactory() {
// only use static methods! // only use static methods!
} }
/** /**
* Create a new unbound {@link BlockingQueue} * Create a new unbound {@link BlockingQueue}
* *
* @param itemClass the {@link Class} type which will be used as {@link BlockingQueue} items * @param itemClass the {@link Class} type which will be used as {@link BlockingQueue} items
* @return queue the {@link BlockingQueue} implementation * @return queue the {@link BlockingQueue} implementation
*/ */
public static <T> BlockingQueue<T> createQueue(Class<T> itemClass) { public static <T> BlockingQueue<T> createQueue(Class<T> itemClass) {
if (useUnsafe) { if (useUnsafe) {
return new LinkedTransferQueue<T>(); return new LinkedTransferQueue<T>();
} else { } else {
return new LegacyLinkedTransferQueue<T>(); return new LegacyLinkedTransferQueue<T>();
} }
} }
/** /**
* Create a new unbound {@link BlockingQueue} * Create a new unbound {@link BlockingQueue}
* *
* @param collection the collection which should get copied to the newly created {@link BlockingQueue} * @param collection the collection which should get copied to the newly created {@link BlockingQueue}
* @param itemClass the {@link Class} type which will be used as {@link BlockingQueue} items * @param itemClass the {@link Class} type which will be used as {@link BlockingQueue} items
* @return queue the {@link BlockingQueue} implementation * @return queue the {@link BlockingQueue} implementation
*/ */
public static <T> BlockingQueue<T> createQueue(Collection<? extends T> collection, Class<T> itemClass) { public static <T> BlockingQueue<T> createQueue(Collection<? extends T> collection, Class<T> itemClass) {
if (useUnsafe) { if (useUnsafe) {
return new LinkedTransferQueue<T>(collection); return new LinkedTransferQueue<T>(collection);
} else { } else {
return new LegacyLinkedTransferQueue<T>(collection); return new LegacyLinkedTransferQueue<T>(collection);
} }
} }
} }

170
handler/src/main/java/io/netty/handler/execution/ChainedExecutor.java
View File

@ -1,85 +1,85 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.execution; package io.netty.handler.execution;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
import io.netty.util.ExternalResourceReleasable; import io.netty.util.ExternalResourceReleasable;
import io.netty.util.internal.ExecutorUtil; import io.netty.util.internal.ExecutorUtil;
/** /**
* A special {@link Executor} which allows to chain a series of * A special {@link Executor} which allows to chain a series of
* {@link Executor}s and {@link ChannelEventRunnableFilter}. * {@link Executor}s and {@link ChannelEventRunnableFilter}.
*/ */
public class ChainedExecutor implements Executor, ExternalResourceReleasable { public class ChainedExecutor implements Executor, ExternalResourceReleasable {
private final Executor cur; private final Executor cur;
private final Executor next; private final Executor next;
private final ChannelEventRunnableFilter filter; private final ChannelEventRunnableFilter filter;
/** /**
* Create a new {@link ChainedExecutor} which will used the given {@link ChannelEventRunnableFilter} to see if the {@link #cur} {@link Executor} should get used. * Create a new {@link ChainedExecutor} which will used the given {@link ChannelEventRunnableFilter} to see if the {@link #cur} {@link Executor} should get used.
* Otherwise it will pass the work to the {@link #next} {@link Executor} * Otherwise it will pass the work to the {@link #next} {@link Executor}
* *
* @param filter the {@link ChannelEventRunnableFilter} which will be used to check if the {@link ChannelEventRunnable} should be passed to the cur or next {@link Executor} * @param filter the {@link ChannelEventRunnableFilter} which will be used to check if the {@link ChannelEventRunnable} should be passed to the cur or next {@link Executor}
* @param cur the {@link Executor} to use if the {@link ChannelEventRunnableFilter} match * @param cur the {@link Executor} to use if the {@link ChannelEventRunnableFilter} match
* @param next the {@link Executor} to use if the {@link ChannelEventRunnableFilter} does not match * @param next the {@link Executor} to use if the {@link ChannelEventRunnableFilter} does not match
*/ */
public ChainedExecutor(ChannelEventRunnableFilter filter, Executor cur, Executor next) { public ChainedExecutor(ChannelEventRunnableFilter filter, Executor cur, Executor next) {
if (filter == null) { if (filter == null) {
throw new NullPointerException("filter"); throw new NullPointerException("filter");
} }
if (cur == null) { if (cur == null) {
throw new NullPointerException("cur"); throw new NullPointerException("cur");
} }
if (next == null) { if (next == null) {
throw new NullPointerException("next"); throw new NullPointerException("next");
} }
this.filter = filter; this.filter = filter;
this.cur = cur; this.cur = cur;
this.next = next; this.next = next;
} }
/** /**
* Execute the passed {@link ChannelEventRunnable} with the current {@link Executor} if the {@link ChannelEventRunnableFilter} match. * Execute the passed {@link ChannelEventRunnable} with the current {@link Executor} if the {@link ChannelEventRunnableFilter} match.
* Otherwise pass it to the next {@link Executor} in the chain. * Otherwise pass it to the next {@link Executor} in the chain.
*/ */
@Override @Override
public void execute(Runnable command) { public void execute(Runnable command) {
assert command instanceof ChannelEventRunnable; assert command instanceof ChannelEventRunnable;
if (filter.filter((ChannelEventRunnable) command)) { if (filter.filter((ChannelEventRunnable) command)) {
cur.execute(command); cur.execute(command);
} else { } else {
next.execute(command); next.execute(command);
} }
} }
@Override @Override
public void releaseExternalResources() { public void releaseExternalResources() {
ExecutorUtil.terminate(cur, next); ExecutorUtil.terminate(cur, next);
releaseExternal(cur); releaseExternal(cur);
releaseExternal(next); releaseExternal(next);
} }
private static void releaseExternal(Executor executor) { private static void releaseExternal(Executor executor) {
if (executor instanceof ExternalResourceReleasable) { if (executor instanceof ExternalResourceReleasable) {
((ExternalResourceReleasable) executor).releaseExternalResources(); ((ExternalResourceReleasable) executor).releaseExternalResources();
} }
} }
} }

56
handler/src/main/java/io/netty/handler/execution/ChannelDownstreamEventRunnableFilter.java
View File

@ -1,28 +1,28 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.execution; package io.netty.handler.execution;
/** /**
* {@link ChannelEventRunnableFilter} implementation which matches {@link ChannelDownstreamEventRunnable} * {@link ChannelEventRunnableFilter} implementation which matches {@link ChannelDownstreamEventRunnable}
* *
*/ */
public class ChannelDownstreamEventRunnableFilter implements ChannelEventRunnableFilter { public class ChannelDownstreamEventRunnableFilter implements ChannelEventRunnableFilter {
@Override @Override
public boolean filter(ChannelEventRunnable event) { public boolean filter(ChannelEventRunnable event) {
return event instanceof ChannelDownstreamEventRunnable; return event instanceof ChannelDownstreamEventRunnable;
} }
} }

112
handler/src/main/java/io/netty/handler/execution/ChannelEventRunnable.java
View File

@ -1,56 +1,56 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.execution; package io.netty.handler.execution;
import io.netty.channel.ChannelEvent; import io.netty.channel.ChannelEvent;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.util.EstimatableObjectWrapper; import io.netty.util.EstimatableObjectWrapper;
public abstract class ChannelEventRunnable implements Runnable, EstimatableObjectWrapper { public abstract class ChannelEventRunnable implements Runnable, EstimatableObjectWrapper {
protected final ChannelHandlerContext ctx; protected final ChannelHandlerContext ctx;
protected final ChannelEvent e; protected final ChannelEvent e;
int estimatedSize; int estimatedSize;
/** /**
* Creates a {@link Runnable} which sends the specified {@link ChannelEvent} * Creates a {@link Runnable} which sends the specified {@link ChannelEvent}
* upstream via the specified {@link ChannelHandlerContext}. * upstream via the specified {@link ChannelHandlerContext}.
*/ */
public ChannelEventRunnable(ChannelHandlerContext ctx, ChannelEvent e) { public ChannelEventRunnable(ChannelHandlerContext ctx, ChannelEvent e) {
this.ctx = ctx; this.ctx = ctx;
this.e = e; this.e = e;
} }
/** /**
* Returns the {@link ChannelHandlerContext} which will be used to * Returns the {@link ChannelHandlerContext} which will be used to
* send the {@link ChannelEvent} upstream. * send the {@link ChannelEvent} upstream.
*/ */
public ChannelHandlerContext getContext() { public ChannelHandlerContext getContext() {
return ctx; return ctx;
} }
/** /**
* Returns the {@link ChannelEvent} which will be sent upstream. * Returns the {@link ChannelEvent} which will be sent upstream.
*/ */
public ChannelEvent getEvent() { public ChannelEvent getEvent() {
return e; return e;
} }
@Override @Override
public Object unwrap() { public Object unwrap() {
return e; return e;
} }
} }

54
handler/src/main/java/io/netty/handler/execution/ChannelEventRunnableFilter.java
View File

@ -1,27 +1,27 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.execution; package io.netty.handler.execution;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
public interface ChannelEventRunnableFilter { public interface ChannelEventRunnableFilter {
/** /**
* Return <code>true</code> if the {@link ChannelEventRunnable} should get handled by the {@link Executor} * Return <code>true</code> if the {@link ChannelEventRunnable} should get handled by the {@link Executor}
* *
*/ */
boolean filter(ChannelEventRunnable event); boolean filter(ChannelEventRunnable event);
} }

52
handler/src/main/java/io/netty/handler/execution/ChannelUpstreamEventRunnableFilter.java
View File

@ -1,26 +1,26 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.execution; package io.netty.handler.execution;
/** /**
* {@link ChannelEventRunnableFilter} which matches {@link ChannelDownstreamEventRunnable} * {@link ChannelEventRunnableFilter} which matches {@link ChannelDownstreamEventRunnable}
*/ */
public class ChannelUpstreamEventRunnableFilter implements ChannelEventRunnableFilter { public class ChannelUpstreamEventRunnableFilter implements ChannelEventRunnableFilter {
@Override @Override
public boolean filter(ChannelEventRunnable event) { public boolean filter(ChannelEventRunnable event) {
return event instanceof ChannelDownstreamEventRunnable; return event instanceof ChannelDownstreamEventRunnable;
} }
} }

134
handler/src/main/java/io/netty/handler/ipfilter/IpFilterListener.java
View File

@ -1,67 +1,67 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.ipfilter; package io.netty.handler.ipfilter;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import io.netty.channel.ChannelEvent; import io.netty.channel.ChannelEvent;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
/** /**
* The listener interface for receiving ipFilter events. * The listener interface for receiving ipFilter events.
* *
* @see IpFilteringHandler * @see IpFilteringHandler
*/ */
public interface IpFilterListener { public interface IpFilterListener {
/** /**
* Called when the channel has the CONNECTED status and the channel was allowed by a previous call to accept(). * Called when the channel has the CONNECTED status and the channel was allowed by a previous call to accept().
* This method enables your implementation to send a message back to the client before closing * This method enables your implementation to send a message back to the client before closing
* or whatever you need. This method returns a ChannelFuture on which the implementation * or whatever you need. This method returns a ChannelFuture on which the implementation
* can wait uninterruptibly before continuing.<br> * can wait uninterruptibly before continuing.<br>
* For instance, If a message is sent back, the corresponding ChannelFuture has to be returned. * For instance, If a message is sent back, the corresponding ChannelFuture has to be returned.
* *
* @param inetSocketAddress the remote {@link InetSocketAddress} from client * @param inetSocketAddress the remote {@link InetSocketAddress} from client
* @return the associated ChannelFuture to be waited for before closing the channel. Null is allowed. * @return the associated ChannelFuture to be waited for before closing the channel. Null is allowed.
*/ */
ChannelFuture allowed(ChannelHandlerContext ctx, ChannelEvent e, InetSocketAddress inetSocketAddress); ChannelFuture allowed(ChannelHandlerContext ctx, ChannelEvent e, InetSocketAddress inetSocketAddress);
/** /**
* Called when the channel has the CONNECTED status and the channel was refused by a previous call to accept(). * Called when the channel has the CONNECTED status and the channel was refused by a previous call to accept().
* This method enables your implementation to send a message back to the client before closing * This method enables your implementation to send a message back to the client before closing
* or whatever you need. This method returns a ChannelFuture on which the implementation * or whatever you need. This method returns a ChannelFuture on which the implementation
* will wait uninterruptibly before closing the channel.<br> * will wait uninterruptibly before closing the channel.<br>
* For instance, If a message is sent back, the corresponding ChannelFuture has to be returned. * For instance, If a message is sent back, the corresponding ChannelFuture has to be returned.
* *
* @param inetSocketAddress the remote {@link InetSocketAddress} from client * @param inetSocketAddress the remote {@link InetSocketAddress} from client
* @return the associated ChannelFuture to be waited for before closing the channel. Null is allowed. * @return the associated ChannelFuture to be waited for before closing the channel. Null is allowed.
*/ */
ChannelFuture refused(ChannelHandlerContext ctx, ChannelEvent e, InetSocketAddress inetSocketAddress); ChannelFuture refused(ChannelHandlerContext ctx, ChannelEvent e, InetSocketAddress inetSocketAddress);
/** /**
* Called in handleUpstream, if this channel was previously blocked, * Called in handleUpstream, if this channel was previously blocked,
* to check if whatever the event, it should be passed to the next entry in the pipeline.<br> * to check if whatever the event, it should be passed to the next entry in the pipeline.<br>
* If one wants to not block events, just overridden this method by returning always true.<br><br> * If one wants to not block events, just overridden this method by returning always true.<br><br>
* <b>Note that OPENED and BOUND events are still passed to the next entry in the pipeline since * <b>Note that OPENED and BOUND events are still passed to the next entry in the pipeline since
* those events come out before the CONNECTED event and so the possibility to filter the connection.</b> * those events come out before the CONNECTED event and so the possibility to filter the connection.</b>
* *
* @return True if the event should continue, False if the event should not continue * @return True if the event should continue, False if the event should not continue
* since this channel was blocked by this filter * since this channel was blocked by this filter
*/ */
boolean continues(ChannelHandlerContext ctx, ChannelEvent e); boolean continues(ChannelHandlerContext ctx, ChannelEvent e);
} }

190
handler/src/main/java/io/netty/handler/ipfilter/IpFilterRuleList.java
View File

@ -1,95 +1,95 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.ipfilter; package io.netty.handler.ipfilter;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.util.ArrayList; import java.util.ArrayList;
import io.netty.logging.InternalLogger; import io.netty.logging.InternalLogger;
import io.netty.logging.InternalLoggerFactory; import io.netty.logging.InternalLoggerFactory;
/** /**
* The Class IpFilterRuleList is a helper class to generate a List of Rules from a string. * The Class IpFilterRuleList is a helper class to generate a List of Rules from a string.
* In case of parse errors no exceptions are thrown. The error is logged. * In case of parse errors no exceptions are thrown. The error is logged.
* <br> * <br>
* Rule List Syntax: * Rule List Syntax:
* <br> * <br>
* <pre> * <pre>
* RuleList ::= Rule[,Rule]* * RuleList ::= Rule[,Rule]*
* Rule ::= AllowRule | BlockRule * Rule ::= AllowRule | BlockRule
* AllowRule ::= +Filter * AllowRule ::= +Filter
* BlockRule ::= -Filter * BlockRule ::= -Filter
* Filter ::= PatternFilter | CIDRFilter * Filter ::= PatternFilter | CIDRFilter
* PatternFilter ::= @see PatternRule * PatternFilter ::= @see PatternRule
* CIDRFilter ::= c:CIDRFilter * CIDRFilter ::= c:CIDRFilter
* CIDRFilter ::= @see CIDR.newCIDR(String) * CIDRFilter ::= @see CIDR.newCIDR(String)
* </pre> * </pre>
* <br> * <br>
* Example: allow only localhost: * Example: allow only localhost:
* <br> * <br>
* new IPFilterRuleHandler().addAll(new IpFilterRuleList("+n:localhost, -n:*")); * new IPFilterRuleHandler().addAll(new IpFilterRuleList("+n:localhost, -n:*"));
* <br> * <br>
*/ */
public class IpFilterRuleList extends ArrayList<IpFilterRule> { public class IpFilterRuleList extends ArrayList<IpFilterRule> {
private static final long serialVersionUID = -6164162941749588780L; private static final long serialVersionUID = -6164162941749588780L;
private static final InternalLogger logger = InternalLoggerFactory.getInstance(IpFilterRuleList.class); private static final InternalLogger logger = InternalLoggerFactory.getInstance(IpFilterRuleList.class);
/** /**
* Instantiates a new ip filter rule list. * Instantiates a new ip filter rule list.
* *
* @param rules the rules * @param rules the rules
*/ */
public IpFilterRuleList(String rules) { public IpFilterRuleList(String rules) {
parseRules(rules); parseRules(rules);
} }
private void parseRules(String rules) { private void parseRules(String rules) {
String[] ruless = rules.split(","); String[] ruless = rules.split(",");
for (String rule : ruless) { for (String rule : ruless) {
parseRule(rule.trim()); parseRule(rule.trim());
} }
} }
private void parseRule(String rule) { private void parseRule(String rule) {
if (rule == null || rule.length() == 0) { if (rule == null || rule.length() == 0) {
return; return;
} }
if (!(rule.startsWith("+") || rule.startsWith("-"))) { if (!(rule.startsWith("+") || rule.startsWith("-"))) {
if (logger.isErrorEnabled()) { if (logger.isErrorEnabled()) {
logger.error("syntax error in ip filter rule:" + rule); logger.error("syntax error in ip filter rule:" + rule);
} }
return; return;
} }
boolean allow = rule.startsWith("+"); boolean allow = rule.startsWith("+");
if (rule.charAt(1) == 'n' || rule.charAt(1) == 'i') { if (rule.charAt(1) == 'n' || rule.charAt(1) == 'i') {
this.add(new PatternRule(allow, rule.substring(1))); this.add(new PatternRule(allow, rule.substring(1)));
} else if (rule.charAt(1) == 'c') { } else if (rule.charAt(1) == 'c') {
try { try {
this.add(new IpSubnetFilterRule(allow, rule.substring(3))); this.add(new IpSubnetFilterRule(allow, rule.substring(3)));
} catch (UnknownHostException e) { } catch (UnknownHostException e) {
if (logger.isErrorEnabled()) { if (logger.isErrorEnabled()) {
logger.error("error parsing ip filter " + rule, e); logger.error("error parsing ip filter " + rule, e);
} }
} }
} else { } else {
if (logger.isErrorEnabled()) { if (logger.isErrorEnabled()) {
logger.error("syntax error in ip filter rule:" + rule); logger.error("syntax error in ip filter rule:" + rule);
} }
} }
} }
} }

328
handler/src/main/java/io/netty/handler/ipfilter/IpFilteringHandlerImpl.java
View File

@ -1,164 +1,164 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.ipfilter; package io.netty.handler.ipfilter;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import io.netty.channel.ChannelEvent; import io.netty.channel.ChannelEvent;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelStateEvent; import io.netty.channel.ChannelStateEvent;
import io.netty.channel.ChannelUpstreamHandler; import io.netty.channel.ChannelUpstreamHandler;
import io.netty.channel.Channels; import io.netty.channel.Channels;
// TODO: Auto-generated Javadoc // TODO: Auto-generated Javadoc
/** General class that handle Ip Filtering. */ /** General class that handle Ip Filtering. */
public abstract class IpFilteringHandlerImpl implements ChannelUpstreamHandler, IpFilteringHandler { public abstract class IpFilteringHandlerImpl implements ChannelUpstreamHandler, IpFilteringHandler {
private IpFilterListener listener; private IpFilterListener listener;
/** /**
* Called when the channel is connected. It returns True if the corresponding connection * Called when the channel is connected. It returns True if the corresponding connection
* is to be allowed. Else it returns False. * is to be allowed. Else it returns False.
* *
* @param inetSocketAddress the remote {@link InetSocketAddress} from client * @param inetSocketAddress the remote {@link InetSocketAddress} from client
* @return True if the corresponding connection is allowed, else False. * @return True if the corresponding connection is allowed, else False.
*/ */
protected abstract boolean accept(ChannelHandlerContext ctx, ChannelEvent e, InetSocketAddress inetSocketAddress) protected abstract boolean accept(ChannelHandlerContext ctx, ChannelEvent e, InetSocketAddress inetSocketAddress)
throws Exception; throws Exception;
/** /**
* Called when the channel has the CONNECTED status and the channel was refused by a previous call to accept(). * Called when the channel has the CONNECTED status and the channel was refused by a previous call to accept().
* This method enables your implementation to send a message back to the client before closing * This method enables your implementation to send a message back to the client before closing
* or whatever you need. This method returns a ChannelFuture on which the implementation * or whatever you need. This method returns a ChannelFuture on which the implementation
* will wait uninterruptibly before closing the channel.<br> * will wait uninterruptibly before closing the channel.<br>
* For instance, If a message is sent back, the corresponding ChannelFuture has to be returned. * For instance, If a message is sent back, the corresponding ChannelFuture has to be returned.
* *
* @param inetSocketAddress the remote {@link InetSocketAddress} from client * @param inetSocketAddress the remote {@link InetSocketAddress} from client
* @return the associated ChannelFuture to be waited for before closing the channel. Null is allowed. * @return the associated ChannelFuture to be waited for before closing the channel. Null is allowed.
*/ */
protected ChannelFuture handleRefusedChannel(ChannelHandlerContext ctx, ChannelEvent e, protected ChannelFuture handleRefusedChannel(ChannelHandlerContext ctx, ChannelEvent e,
InetSocketAddress inetSocketAddress) throws Exception { InetSocketAddress inetSocketAddress) throws Exception {
if (listener == null) { if (listener == null) {
return null; return null;
} }
return listener.refused(ctx, e, inetSocketAddress); return listener.refused(ctx, e, inetSocketAddress);
} }
protected ChannelFuture handleAllowedChannel(ChannelHandlerContext ctx, ChannelEvent e, protected ChannelFuture handleAllowedChannel(ChannelHandlerContext ctx, ChannelEvent e,
InetSocketAddress inetSocketAddress) throws Exception { InetSocketAddress inetSocketAddress) throws Exception {
if (listener == null) { if (listener == null) {
return null; return null;
} }
return listener.allowed(ctx, e, inetSocketAddress); return listener.allowed(ctx, e, inetSocketAddress);
} }
/** /**
* Internal method to test if the current channel is blocked. Should not be overridden. * Internal method to test if the current channel is blocked. Should not be overridden.
* *
* @return True if the current channel is blocked, else False * @return True if the current channel is blocked, else False
*/ */
protected boolean isBlocked(ChannelHandlerContext ctx) { protected boolean isBlocked(ChannelHandlerContext ctx) {
return ctx.getAttachment() != null; return ctx.getAttachment() != null;
} }
/** /**
* Called in handleUpstream, if this channel was previously blocked, * Called in handleUpstream, if this channel was previously blocked,
* to check if whatever the event, it should be passed to the next entry in the pipeline.<br> * to check if whatever the event, it should be passed to the next entry in the pipeline.<br>
* If one wants to not block events, just overridden this method by returning always true.<br><br> * If one wants to not block events, just overridden this method by returning always true.<br><br>
* <b>Note that OPENED and BOUND events are still passed to the next entry in the pipeline since * <b>Note that OPENED and BOUND events are still passed to the next entry in the pipeline since
* those events come out before the CONNECTED event and so the possibility to filter the connection.</b> * those events come out before the CONNECTED event and so the possibility to filter the connection.</b>
* *
* @return True if the event should continue, False if the event should not continue * @return True if the event should continue, False if the event should not continue
* since this channel was blocked by this filter * since this channel was blocked by this filter
*/ */
protected boolean continues(ChannelHandlerContext ctx, ChannelEvent e) throws Exception { protected boolean continues(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
if (listener != null) { if (listener != null) {
return listener.continues(ctx, e); return listener.continues(ctx, e);
} else { } else {
return false; return false;
} }
} }
@Override @Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception { public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
if (e instanceof ChannelStateEvent) { if (e instanceof ChannelStateEvent) {
ChannelStateEvent evt = (ChannelStateEvent) e; ChannelStateEvent evt = (ChannelStateEvent) e;
switch (evt.getState()) { switch (evt.getState()) {
case OPEN: case OPEN:
case BOUND: case BOUND:
// Special case: OPEND and BOUND events are before CONNECTED, // Special case: OPEND and BOUND events are before CONNECTED,
// but CLOSED and UNBOUND events are after DISCONNECTED: should those events be blocked too? // but CLOSED and UNBOUND events are after DISCONNECTED: should those events be blocked too?
if (isBlocked(ctx) && !continues(ctx, evt)) { if (isBlocked(ctx) && !continues(ctx, evt)) {
// don't pass to next level since channel was blocked early // don't pass to next level since channel was blocked early
return; return;
} else { } else {
ctx.sendUpstream(e); ctx.sendUpstream(e);
return; return;
} }
case CONNECTED: case CONNECTED:
if (evt.getValue() != null) { if (evt.getValue() != null) {
// CONNECTED // CONNECTED
InetSocketAddress inetSocketAddress = (InetSocketAddress) e.getChannel().getRemoteAddress(); InetSocketAddress inetSocketAddress = (InetSocketAddress) e.getChannel().getRemoteAddress();
if (!accept(ctx, e, inetSocketAddress)) { if (!accept(ctx, e, inetSocketAddress)) {
ctx.setAttachment(Boolean.TRUE); ctx.setAttachment(Boolean.TRUE);
ChannelFuture future = handleRefusedChannel(ctx, e, inetSocketAddress); ChannelFuture future = handleRefusedChannel(ctx, e, inetSocketAddress);
if (future != null) { if (future != null) {
future.addListener(ChannelFutureListener.CLOSE); future.addListener(ChannelFutureListener.CLOSE);
} else { } else {
Channels.close(e.getChannel()); Channels.close(e.getChannel());
} }
if (isBlocked(ctx) && !continues(ctx, evt)) { if (isBlocked(ctx) && !continues(ctx, evt)) {
// don't pass to next level since channel was blocked early // don't pass to next level since channel was blocked early
return; return;
} }
} else { } else {
handleAllowedChannel(ctx, e, inetSocketAddress); handleAllowedChannel(ctx, e, inetSocketAddress);
} }
// This channel is not blocked // This channel is not blocked
ctx.setAttachment(null); ctx.setAttachment(null);
} else { } else {
// DISCONNECTED // DISCONNECTED
if (isBlocked(ctx) && !continues(ctx, evt)) { if (isBlocked(ctx) && !continues(ctx, evt)) {
// don't pass to next level since channel was blocked early // don't pass to next level since channel was blocked early
return; return;
} }
} }
break; break;
} }
} }
if (isBlocked(ctx) && !continues(ctx, e)) { if (isBlocked(ctx) && !continues(ctx, e)) {
// don't pass to next level since channel was blocked early // don't pass to next level since channel was blocked early
return; return;
} }
// Whatever it is, if not blocked, goes to the next level // Whatever it is, if not blocked, goes to the next level
ctx.sendUpstream(e); ctx.sendUpstream(e);
} }
@Override @Override
public void setIpFilterListener(IpFilterListener listener) { public void setIpFilterListener(IpFilterListener listener) {
this.listener = listener; this.listener = listener;
} }
@Override @Override
public void removeIpFilterListener() { public void removeIpFilterListener() {
this.listener = null; this.listener = null;
} }
} }

404
handler/src/main/java/io/netty/handler/ipfilter/PatternRule.java
View File

@ -1,202 +1,202 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.ipfilter; package io.netty.handler.ipfilter;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import io.netty.logging.InternalLogger; import io.netty.logging.InternalLogger;
import io.netty.logging.InternalLoggerFactory; import io.netty.logging.InternalLoggerFactory;
/** /**
* The Class PatternRule represents an IP filter rule using string patterns. * The Class PatternRule represents an IP filter rule using string patterns.
* <br> * <br>
* Rule Syntax: * Rule Syntax:
* <br> * <br>
* <pre> * <pre>
* Rule ::= [n|i]:address n stands for computer name, i for ip address * Rule ::= [n|i]:address n stands for computer name, i for ip address
* address ::= &lt;regex&gt; | localhost * address ::= &lt;regex&gt; | localhost
* regex is a regular expression with '*' as multi character and '?' as single character wild card * regex is a regular expression with '*' as multi character and '?' as single character wild card
* </pre> * </pre>
* <br> * <br>
* Example: allow localhost: * Example: allow localhost:
* <br> * <br>
* new PatternRule(true, "n:localhost") * new PatternRule(true, "n:localhost")
* <br> * <br>
* Example: allow local lan: * Example: allow local lan:
* <br> * <br>
* new PatternRule(true, "i:192.168.0.*") * new PatternRule(true, "i:192.168.0.*")
* <br> * <br>
* Example: block all * Example: block all
* <br> * <br>
* new PatternRule(false, "n:*") * new PatternRule(false, "n:*")
* <br> * <br>
*/ */
public class PatternRule implements IpFilterRule, Comparable<Object> { public class PatternRule implements IpFilterRule, Comparable<Object> {
private static final InternalLogger logger = InternalLoggerFactory.getInstance(PatternRule.class); private static final InternalLogger logger = InternalLoggerFactory.getInstance(PatternRule.class);
private Pattern ipPattern; private Pattern ipPattern;
private Pattern namePattern; private Pattern namePattern;
private boolean isAllowRule = true; private boolean isAllowRule = true;
private boolean localhost; private boolean localhost;
private String pattern; private String pattern;
/** /**
* Instantiates a new pattern rule. * Instantiates a new pattern rule.
* *
* @param allow indicates if this is an allow or block rule * @param allow indicates if this is an allow or block rule
* @param pattern the filter pattern * @param pattern the filter pattern
*/ */
public PatternRule(boolean allow, String pattern) { public PatternRule(boolean allow, String pattern) {
this.isAllowRule = allow; this.isAllowRule = allow;
this.pattern = pattern; this.pattern = pattern;
parse(pattern); parse(pattern);
} }
/** /**
* returns the pattern. * returns the pattern.
* *
* @return the pattern * @return the pattern
*/ */
public String getPattern() { public String getPattern() {
return this.pattern; return this.pattern;
} }
@Override @Override
public boolean isAllowRule() { public boolean isAllowRule() {
return isAllowRule; return isAllowRule;
} }
@Override @Override
public boolean isDenyRule() { public boolean isDenyRule() {
return !isAllowRule; return !isAllowRule;
} }
@Override @Override
public boolean contains(InetAddress inetAddress) { public boolean contains(InetAddress inetAddress) {
if (localhost) { if (localhost) {
if (isLocalhost(inetAddress)) { if (isLocalhost(inetAddress)) {
return true; return true;
} }
} }
if (ipPattern != null) { if (ipPattern != null) {
if (ipPattern.matcher(inetAddress.getHostAddress()).matches()) { if (ipPattern.matcher(inetAddress.getHostAddress()).matches()) {
return true; return true;
} }
} }
if (namePattern != null) { if (namePattern != null) {
if (namePattern.matcher(inetAddress.getHostName()).matches()) { if (namePattern.matcher(inetAddress.getHostName()).matches()) {
return true; return true;
} }
} }
return false; return false;
} }
private void parse(String pattern) { private void parse(String pattern) {
if (pattern == null) { if (pattern == null) {
return; return;
} }
String[] acls = pattern.split(","); String[] acls = pattern.split(",");
String ip = ""; String ip = "";
String name = ""; String name = "";
for (String c : acls) { for (String c : acls) {
c = c.trim(); c = c.trim();
if (c.equals("n:localhost")) { if (c.equals("n:localhost")) {
this.localhost = true; this.localhost = true;
} else if (c.startsWith("n:")) { } else if (c.startsWith("n:")) {
name = addRule(name, c.substring(2)); name = addRule(name, c.substring(2));
} else if (c.startsWith("i:")) { } else if (c.startsWith("i:")) {
ip = addRule(ip, c.substring(2)); ip = addRule(ip, c.substring(2));
} }
} }
if (ip.length() != 0) { if (ip.length() != 0) {
ipPattern = Pattern.compile(ip); ipPattern = Pattern.compile(ip);
} }
if (name.length() != 0) { if (name.length() != 0) {
namePattern = Pattern.compile(name); namePattern = Pattern.compile(name);
} }
} }
private String addRule(String pattern, String rule) { private String addRule(String pattern, String rule) {
if (rule == null || rule.length() == 0) { if (rule == null || rule.length() == 0) {
return pattern; return pattern;
} }
if (pattern.length() != 0) { if (pattern.length() != 0) {
pattern += "|"; pattern += "|";
} }
rule = rule.replaceAll("\\.", "\\\\."); rule = rule.replaceAll("\\.", "\\\\.");
rule = rule.replaceAll("\\*", ".*"); rule = rule.replaceAll("\\*", ".*");
rule = rule.replaceAll("\\?", "."); rule = rule.replaceAll("\\?", ".");
pattern += "(" + rule + ")"; pattern += "(" + rule + ")";
return pattern; return pattern;
} }
private boolean isLocalhost(InetAddress address) { private boolean isLocalhost(InetAddress address) {
try { try {
if (address.equals(InetAddress.getLocalHost())) { if (address.equals(InetAddress.getLocalHost())) {
return true; return true;
} }
} catch (UnknownHostException e) { } catch (UnknownHostException e) {
if (logger.isInfoEnabled()) { if (logger.isInfoEnabled()) {
logger.info("error getting ip of localhost", e); logger.info("error getting ip of localhost", e);
} }
} }
try { try {
InetAddress[] addrs = InetAddress.getAllByName("127.0.0.1"); InetAddress[] addrs = InetAddress.getAllByName("127.0.0.1");
for (InetAddress addr : addrs) { for (InetAddress addr : addrs) {
if (addr.equals(address)) { if (addr.equals(address)) {
return true; return true;
} }
} }
} catch (UnknownHostException e) { } catch (UnknownHostException e) {
if (logger.isInfoEnabled()) { if (logger.isInfoEnabled()) {
logger.info("error getting ip of localhost", e); logger.info("error getting ip of localhost", e);
} }
} }
return false; return false;
} }
@Override @Override
public int compareTo(Object o) { public int compareTo(Object o) {
if (o == null) { if (o == null) {
return -1; return -1;
} }
if (!(o instanceof PatternRule)) { if (!(o instanceof PatternRule)) {
return -1; return -1;
} }
PatternRule p = (PatternRule) o; PatternRule p = (PatternRule) o;
if (p.isAllowRule() && !this.isAllowRule) { if (p.isAllowRule() && !this.isAllowRule) {
return -1; return -1;
} }
if (this.pattern == null && p.pattern == null) { if (this.pattern == null && p.pattern == null) {
return 0; return 0;
} }
if (this.pattern != null) { if (this.pattern != null) {
return this.pattern.compareTo(p.getPattern()); return this.pattern.compareTo(p.getPattern());
} }
return -1; return -1;
} }
} }

158
handler/src/main/java/io/netty/handler/region/ChannelWritableByteChannel.java
View File

@ -1,79 +1,79 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.region; package io.netty.handler.region;
import java.io.IOException; import java.io.IOException;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel; import java.nio.channels.WritableByteChannel;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers; import io.netty.buffer.ChannelBuffers;
import io.netty.channel.ChannelDownstreamHandler; import io.netty.channel.ChannelDownstreamHandler;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureAggregator; import io.netty.channel.ChannelFutureAggregator;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPipeline;
import io.netty.channel.Channels; import io.netty.channel.Channels;
import io.netty.channel.MessageEvent; import io.netty.channel.MessageEvent;
/** /**
* {@link WritableByteChannel} implementation which will take care to wrap the {@link ByteBuffer} to a {@link ChannelBuffer} and forward it to the next {@link ChannelDownstreamHandler} in the {@link ChannelPipeline} on every {@link #write(ByteBuffer)} * {@link WritableByteChannel} implementation which will take care to wrap the {@link ByteBuffer} to a {@link ChannelBuffer} and forward it to the next {@link ChannelDownstreamHandler} in the {@link ChannelPipeline} on every {@link #write(ByteBuffer)}
* operation. * operation.
*/ */
public class ChannelWritableByteChannel implements WritableByteChannel { public class ChannelWritableByteChannel implements WritableByteChannel {
private boolean closed; private boolean closed;
private final ChannelHandlerContext context; private final ChannelHandlerContext context;
private final ChannelFutureAggregator aggregator; private final ChannelFutureAggregator aggregator;
private final SocketAddress remote; private final SocketAddress remote;
public ChannelWritableByteChannel(ChannelHandlerContext context, MessageEvent event) { public ChannelWritableByteChannel(ChannelHandlerContext context, MessageEvent event) {
this(context, new ChannelFutureAggregator(event.getFuture()), event.getRemoteAddress()); this(context, new ChannelFutureAggregator(event.getFuture()), event.getRemoteAddress());
} }
public ChannelWritableByteChannel(ChannelHandlerContext context, ChannelFutureAggregator aggregator, SocketAddress remote) { public ChannelWritableByteChannel(ChannelHandlerContext context, ChannelFutureAggregator aggregator, SocketAddress remote) {
this.context = context; this.context = context;
this.aggregator = aggregator; this.aggregator = aggregator;
this.remote = remote; this.remote = remote;
} }
@Override @Override
public boolean isOpen() { public boolean isOpen() {
return !closed && context.channel().isOpen(); return !closed && context.channel().isOpen();
} }
@Override @Override
public void close() throws IOException { public void close() throws IOException {
closed = true; closed = true;
} }
@Override @Override
public int write(ByteBuffer src) throws IOException { public int write(ByteBuffer src) throws IOException {
int written = src.remaining(); int written = src.remaining();
// create a new ChannelFuture and add it to the aggregator // create a new ChannelFuture and add it to the aggregator
ChannelFuture future = Channels.future(context.channel(), true); ChannelFuture future = Channels.future(context.channel(), true);
aggregator.addFuture(future); aggregator.addFuture(future);
Channels.write(context, future, ChannelBuffers.wrappedBuffer(src), remote); Channels.write(context, future, ChannelBuffers.wrappedBuffer(src), remote);
return written; return written;
} }
} }

150
handler/src/main/java/io/netty/handler/region/FileRegionEncoder.java
View File

@ -1,75 +1,75 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.region; package io.netty.handler.region;
import java.nio.channels.WritableByteChannel; import java.nio.channels.WritableByteChannel;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
import io.netty.channel.ChannelDownstreamHandler; import io.netty.channel.ChannelDownstreamHandler;
import io.netty.channel.ChannelEvent; import io.netty.channel.ChannelEvent;
import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPipeline;
import io.netty.channel.FileRegion; import io.netty.channel.FileRegion;
import io.netty.channel.MessageEvent; import io.netty.channel.MessageEvent;
/** /**
* {@link ChannelDownstreamHandler} implementation which encodes a {@link FileRegion} to {@link ChannelBuffer}'s if one of the given {@link ChannelHandler} was found in the {@link ChannelPipeline}. * {@link ChannelDownstreamHandler} implementation which encodes a {@link FileRegion} to {@link ChannelBuffer}'s if one of the given {@link ChannelHandler} was found in the {@link ChannelPipeline}.
* *
* This {@link ChannelDownstreamHandler} should be used if you plan to write {@link FileRegion} objects and also have some {@link ChannelDownstreamHandler} in the {@link ChannelPipeline} which needs to transform * This {@link ChannelDownstreamHandler} should be used if you plan to write {@link FileRegion} objects and also have some {@link ChannelDownstreamHandler} in the {@link ChannelPipeline} which needs to transform
* the to be written {@link ChannelBuffer} in any case. This could be for example {@link ChannelDownstreamHandler}'s which needs to encrypt or compress messages. * the to be written {@link ChannelBuffer} in any case. This could be for example {@link ChannelDownstreamHandler}'s which needs to encrypt or compress messages.
* *
* Users of this {@link FileRegionEncoder} should add / remove this {@link ChannelDownstreamHandler} on the fly to get the best performance out of their system. * Users of this {@link FileRegionEncoder} should add / remove this {@link ChannelDownstreamHandler} on the fly to get the best performance out of their system.
* *
* *
*/ */
@ChannelHandler.Sharable @ChannelHandler.Sharable
public class FileRegionEncoder implements ChannelDownstreamHandler { public class FileRegionEncoder implements ChannelDownstreamHandler {
@Override @Override
public void handleDownstream( public void handleDownstream(
ChannelHandlerContext ctx, ChannelEvent evt) throws Exception { ChannelHandlerContext ctx, ChannelEvent evt) throws Exception {
if (!(evt instanceof MessageEvent)) { if (!(evt instanceof MessageEvent)) {
ctx.sendDownstream(evt); ctx.sendDownstream(evt);
return; return;
} }
MessageEvent e = (MessageEvent) evt; MessageEvent e = (MessageEvent) evt;
Object originalMessage = e.getMessage(); Object originalMessage = e.getMessage();
if (originalMessage instanceof FileRegion) { if (originalMessage instanceof FileRegion) {
FileRegion fr = (FileRegion) originalMessage; FileRegion fr = (FileRegion) originalMessage;
WritableByteChannel bchannel = new ChannelWritableByteChannel(ctx, e); WritableByteChannel bchannel = new ChannelWritableByteChannel(ctx, e);
int length = 0; int length = 0;
long i = 0; long i = 0;
while ((i = fr.transferTo(bchannel, length)) > 0) { while ((i = fr.transferTo(bchannel, length)) > 0) {
length += i; length += i;
if (length >= fr.getCount()) { if (length >= fr.getCount()) {
break; break;
} }
} }
} else { } else {
// no converting is needed so just sent the event downstream // no converting is needed so just sent the event downstream
ctx.sendDownstream(evt); ctx.sendDownstream(evt);
} }
} }
} }

694
handler/src/test/java/io/netty/handler/ipfilter/IpFilterRuleTest.java
View File

@ -1,347 +1,347 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.handler.ipfilter; package io.netty.handler.ipfilter;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.SocketAddress; import java.net.SocketAddress;
import junit.framework.TestCase; import junit.framework.TestCase;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelConfig; import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelEvent; import io.netty.channel.ChannelEvent;
import io.netty.channel.ChannelFactory; import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPipeline;
import io.netty.channel.UpstreamMessageEvent; import io.netty.channel.UpstreamMessageEvent;
import org.junit.Test; import org.junit.Test;
public class IpFilterRuleTest extends TestCase public class IpFilterRuleTest extends TestCase
{ {
public static boolean accept(IpFilterRuleHandler h, InetSocketAddress addr) throws Exception public static boolean accept(IpFilterRuleHandler h, InetSocketAddress addr) throws Exception
{ {
return h.accept(new ChannelHandlerContext() return h.accept(new ChannelHandlerContext()
{ {
@Override @Override
public boolean canHandleDownstream() public boolean canHandleDownstream()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public boolean canHandleUpstream() public boolean canHandleUpstream()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public Object getAttachment() public Object getAttachment()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public Channel getChannel() public Channel getChannel()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelHandler getHandler() public ChannelHandler getHandler()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public String getName() public String getName()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelPipeline getPipeline() public ChannelPipeline getPipeline()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public void sendDownstream(ChannelEvent e) public void sendDownstream(ChannelEvent e)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
} }
@Override @Override
public void sendUpstream(ChannelEvent e) public void sendUpstream(ChannelEvent e)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
} }
@Override @Override
public void setAttachment(Object attachment) public void setAttachment(Object attachment)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
} }
}, },
new UpstreamMessageEvent(new Channel() new UpstreamMessageEvent(new Channel()
{ {
@Override @Override
public ChannelFuture bind(SocketAddress localAddress) public ChannelFuture bind(SocketAddress localAddress)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture close() public ChannelFuture close()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture connect(SocketAddress remoteAddress) public ChannelFuture connect(SocketAddress remoteAddress)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture disconnect() public ChannelFuture disconnect()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture getCloseFuture() public ChannelFuture getCloseFuture()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelConfig getConfig() public ChannelConfig getConfig()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFactory getFactory() public ChannelFactory getFactory()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public Integer getId() public Integer getId()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public int getInterestOps() public int getInterestOps()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return 0; return 0;
} }
@Override @Override
public SocketAddress getLocalAddress() public SocketAddress getLocalAddress()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public Channel getParent() public Channel getParent()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelPipeline getPipeline() public ChannelPipeline getPipeline()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public SocketAddress getRemoteAddress() public SocketAddress getRemoteAddress()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public boolean isBound() public boolean isBound()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public boolean isConnected() public boolean isConnected()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public boolean isOpen() public boolean isOpen()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public boolean isReadable() public boolean isReadable()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public boolean isWritable() public boolean isWritable()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public ChannelFuture setInterestOps(int interestOps) public ChannelFuture setInterestOps(int interestOps)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture setReadable(boolean readable) public ChannelFuture setReadable(boolean readable)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture unbind() public ChannelFuture unbind()
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture write(Object message) public ChannelFuture write(Object message)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public ChannelFuture write(Object message, SocketAddress remoteAddress) public ChannelFuture write(Object message, SocketAddress remoteAddress)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public int compareTo(Channel o) public int compareTo(Channel o)
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return 0; return 0;
} }
@Override @Override
public Object getAttachment() { public Object getAttachment() {
return null; return null;
} }
@Override @Override
public void setAttachment(Object attachment) { public void setAttachment(Object attachment) {
} }
}, h, addr), }, h, addr),
addr); addr);
} }
@Test @Test
public void testIpFilterRule() throws Exception public void testIpFilterRule() throws Exception
{ {
IpFilterRuleHandler h = new IpFilterRuleHandler(); IpFilterRuleHandler h = new IpFilterRuleHandler();
h.addAll(new IpFilterRuleList("+n:localhost, -n:*")); h.addAll(new IpFilterRuleList("+n:localhost, -n:*"));
InetSocketAddress addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080); InetSocketAddress addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080); addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080);
assertFalse(accept(h, addr)); assertFalse(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080); addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
h.clear(); h.clear();
h.addAll(new IpFilterRuleList("+n:*"+InetAddress.getLocalHost().getHostName().substring(1)+", -n:*")); h.addAll(new IpFilterRuleList("+n:*"+InetAddress.getLocalHost().getHostName().substring(1)+", -n:*"));
addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080); addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080); addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080);
assertFalse(accept(h, addr)); assertFalse(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080); addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
h.clear(); h.clear();
h.addAll(new IpFilterRuleList("+c:"+InetAddress.getLocalHost().getHostAddress()+"/32, -n:*")); h.addAll(new IpFilterRuleList("+c:"+InetAddress.getLocalHost().getHostAddress()+"/32, -n:*"));
addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080); addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080); addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080);
assertFalse(accept(h, addr)); assertFalse(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080); addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
h.clear(); h.clear();
h.addAll(new IpFilterRuleList("")); h.addAll(new IpFilterRuleList(""));
addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080); addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080); addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080); addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
h.clear(); h.clear();
addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080); addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080); addr = new InetSocketAddress(InetAddress.getByName("127.0.0.2"), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080); addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080);
assertTrue(accept(h, addr)); assertTrue(accept(h, addr));
} }
} }

72
transport-http/src/main/java/io/netty/channel/socket/http/HttpTunnelAcceptedChannelReceiver.java
View File

@ -1,36 +1,36 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.http; package io.netty.channel.socket.http;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
/** /**
* Interface from the server message switch and channel sink to an * Interface from the server message switch and channel sink to an
* accepted channel. Exists primarily for mock testing purposes. * accepted channel. Exists primarily for mock testing purposes.
* *
*/ */
interface HttpTunnelAcceptedChannelReceiver { interface HttpTunnelAcceptedChannelReceiver {
void updateInterestOps(SaturationStateChange transition); void updateInterestOps(SaturationStateChange transition);
void dataReceived(ChannelBuffer data); void dataReceived(ChannelBuffer data);
void clientClosed(); void clientClosed();
} }

138
transport-http/src/main/java/io/netty/channel/socket/http/SaturationManager.java
View File

@ -1,69 +1,69 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.http; package io.netty.channel.socket.http;
import static io.netty.channel.socket.http.SaturationStateChange.DESATURATED; import static io.netty.channel.socket.http.SaturationStateChange.DESATURATED;
import static io.netty.channel.socket.http.SaturationStateChange.NO_CHANGE; import static io.netty.channel.socket.http.SaturationStateChange.NO_CHANGE;
import static io.netty.channel.socket.http.SaturationStateChange.SATURATED; import static io.netty.channel.socket.http.SaturationStateChange.SATURATED;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
/** /**
* This class is used to monitor the amount of data that has yet to be pushed to * This class is used to monitor the amount of data that has yet to be pushed to
* the underlying socket, in order to implement the "high/low water mark" facility * the underlying socket, in order to implement the "high/low water mark" facility
* that controls Channel.isWritable() and the interest ops of http tunnels. * that controls Channel.isWritable() and the interest ops of http tunnels.
* *
*/ */
class SaturationManager { class SaturationManager {
private final AtomicLong desaturationPoint; private final AtomicLong desaturationPoint;
private final AtomicLong saturationPoint; private final AtomicLong saturationPoint;
private final AtomicLong queueSize; private final AtomicLong queueSize;
private final AtomicBoolean saturated; private final AtomicBoolean saturated;
public SaturationManager(long desaturationPoint, long saturationPoint) { public SaturationManager(long desaturationPoint, long saturationPoint) {
this.desaturationPoint = new AtomicLong(desaturationPoint); this.desaturationPoint = new AtomicLong(desaturationPoint);
this.saturationPoint = new AtomicLong(saturationPoint); this.saturationPoint = new AtomicLong(saturationPoint);
queueSize = new AtomicLong(0); queueSize = new AtomicLong(0);
saturated = new AtomicBoolean(false); saturated = new AtomicBoolean(false);
} }
public SaturationStateChange queueSizeChanged(long sizeDelta) { public SaturationStateChange queueSizeChanged(long sizeDelta) {
long newQueueSize = queueSize.addAndGet(sizeDelta); long newQueueSize = queueSize.addAndGet(sizeDelta);
if (newQueueSize <= desaturationPoint.get()) { if (newQueueSize <= desaturationPoint.get()) {
if (saturated.compareAndSet(true, false)) { if (saturated.compareAndSet(true, false)) {
return DESATURATED; return DESATURATED;
} }
} else if (newQueueSize > saturationPoint.get()) { } else if (newQueueSize > saturationPoint.get()) {
if (saturated.compareAndSet(false, true)) { if (saturated.compareAndSet(false, true)) {
return SATURATED; return SATURATED;
} }
} }
return NO_CHANGE; return NO_CHANGE;
} }
public void updateThresholds(long desaturationPoint, long saturationPoint) { public void updateThresholds(long desaturationPoint, long saturationPoint) {
this.desaturationPoint.set(desaturationPoint); this.desaturationPoint.set(desaturationPoint);
this.saturationPoint.set(saturationPoint); this.saturationPoint.set(saturationPoint);
} }
} }

58
transport-http/src/main/java/io/netty/channel/socket/http/SaturationStateChange.java
View File

@ -1,29 +1,29 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.http; package io.netty.channel.socket.http;
/** /**
* Represents the state change of a chanel in response in the amount of pending data to be * Represents the state change of a chanel in response in the amount of pending data to be
* sent - either no change occurs, the channel becomes desaturated (indicating that writing * sent - either no change occurs, the channel becomes desaturated (indicating that writing
* can safely commence) or it becomes saturated (indicating that writing should cease). * can safely commence) or it becomes saturated (indicating that writing should cease).
* *
*/ */
enum SaturationStateChange { enum SaturationStateChange {
NO_CHANGE, DESATURATED, SATURATED NO_CHANGE, DESATURATED, SATURATED
} }

104
transport-http/src/test/java/io/netty/channel/socket/http/SaturationManagerTest.java
View File

@ -1,52 +1,52 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.http; package io.netty.channel.socket.http;
import static org.junit.Assert.*; import static org.junit.Assert.*;
import static io.netty.channel.socket.http.SaturationStateChange.*; import static io.netty.channel.socket.http.SaturationStateChange.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
/** /**
* Tests saturation managers * Tests saturation managers
* *
*/ */
public class SaturationManagerTest { public class SaturationManagerTest {
private SaturationManager manager; private SaturationManager manager;
@Before @Before
public void setUp() { public void setUp() {
manager = new SaturationManager(100L, 200L); manager = new SaturationManager(100L, 200L);
} }
@Test @Test
public void testQueueSizeChanged() { public void testQueueSizeChanged() {
assertEquals(NO_CHANGE, manager.queueSizeChanged(100L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(100L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(99L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(99L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(1L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(1L));
assertEquals(SATURATED, manager.queueSizeChanged(1L)); assertEquals(SATURATED, manager.queueSizeChanged(1L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(10L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(10L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(-10L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(-10L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(-1L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(-1L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(-1L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(-1L));
assertEquals(DESATURATED, manager.queueSizeChanged(-99L)); assertEquals(DESATURATED, manager.queueSizeChanged(-99L));
assertEquals(NO_CHANGE, manager.queueSizeChanged(-100L)); assertEquals(NO_CHANGE, manager.queueSizeChanged(-100L));
} }
} }

218
transport-sctp/src/main/java/io/netty/channel/sctp/SctpJdkChannel.java
View File

@ -1,109 +1,109 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.sctp; package io.netty.channel.sctp;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import com.sun.nio.sctp.SctpChannel; import com.sun.nio.sctp.SctpChannel;
import io.netty.channel.socket.nio.AbstractJdkChannel; import io.netty.channel.socket.nio.AbstractJdkChannel;
public class SctpJdkChannel extends AbstractJdkChannel { public class SctpJdkChannel extends AbstractJdkChannel {
SctpJdkChannel(SctpChannel channel) { SctpJdkChannel(SctpChannel channel) {
super(channel); super(channel);
} }
@Override @Override
protected SctpChannel getChannel() { protected SctpChannel getChannel() {
return (SctpChannel) super.getChannel(); return (SctpChannel) super.getChannel();
} }
@Override @Override
public InetSocketAddress getRemoteSocketAddress() { public InetSocketAddress getRemoteSocketAddress() {
try { try {
for (SocketAddress address : getChannel().getRemoteAddresses()) { for (SocketAddress address : getChannel().getRemoteAddresses()) {
return (InetSocketAddress) address; return (InetSocketAddress) address;
} }
} catch (IOException e) { } catch (IOException e) {
// ignore // ignore
} }
return null; return null;
} }
@Override @Override
public SocketAddress getLocalSocketAddress() { public SocketAddress getLocalSocketAddress() {
try { try {
for (SocketAddress address : getChannel().getAllLocalAddresses()) { for (SocketAddress address : getChannel().getAllLocalAddresses()) {
return (InetSocketAddress) address; return (InetSocketAddress) address;
} }
} catch (IOException e) { } catch (IOException e) {
// ignore // ignore
} }
return null; return null;
} }
@Override @Override
public boolean isConnected() { public boolean isConnected() {
return getChannel().isOpen(); return getChannel().isOpen();
} }
@Override @Override
public boolean isSocketBound() { public boolean isSocketBound() {
try { try {
return !getChannel().getAllLocalAddresses().isEmpty(); return !getChannel().getAllLocalAddresses().isEmpty();
} catch (IOException e) { } catch (IOException e) {
return false; return false;
} }
} }
@Override @Override
public void disconnectSocket() throws IOException { public void disconnectSocket() throws IOException {
closeSocket(); closeSocket();
} }
@Override @Override
public void closeSocket() throws IOException { public void closeSocket() throws IOException {
for (SocketAddress address: getChannel().getAllLocalAddresses()) { for (SocketAddress address: getChannel().getAllLocalAddresses()) {
getChannel().unbindAddress(((InetSocketAddress) address).getAddress()); getChannel().unbindAddress(((InetSocketAddress) address).getAddress());
} }
} }
@Override @Override
public void bind(SocketAddress local) throws IOException { public void bind(SocketAddress local) throws IOException {
getChannel().bind(local); getChannel().bind(local);
} }
@Override @Override
public void connect(SocketAddress remote) throws IOException { public void connect(SocketAddress remote) throws IOException {
getChannel().connect(remote); getChannel().connect(remote);
} }
@Override @Override
public int write(ByteBuffer src) throws IOException { public int write(ByteBuffer src) throws IOException {
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }
@Override @Override
public boolean finishConnect() throws IOException { public boolean finishConnect() throws IOException {
return getChannel().finishConnect(); return getChannel().finishConnect();
} }
} }

66
transport-sctp/src/main/java/io/netty/channel/sctp/SctpWorkerPool.java
View File

@ -1,33 +1,33 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.sctp; package io.netty.channel.sctp;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
import io.netty.channel.socket.nio.AbstractNioWorkerPool; import io.netty.channel.socket.nio.AbstractNioWorkerPool;
public class SctpWorkerPool extends AbstractNioWorkerPool<SctpWorker> { public class SctpWorkerPool extends AbstractNioWorkerPool<SctpWorker> {
public SctpWorkerPool(Executor executor, int workerCount, boolean allowShutdownOnIdle) { public SctpWorkerPool(Executor executor, int workerCount, boolean allowShutdownOnIdle) {
super(executor, workerCount, allowShutdownOnIdle); super(executor, workerCount, allowShutdownOnIdle);
} }
@Override @Override
protected SctpWorker createWorker(Executor executor, boolean allowShutdownOnIdle) { protected SctpWorker createWorker(Executor executor, boolean allowShutdownOnIdle) {
return new SctpWorker(executor, allowShutdownOnIdle); return new SctpWorker(executor, allowShutdownOnIdle);
} }
} }

236
transport/src/main/java/io/netty/channel/socket/oio/AbstractOioChannel.java
View File

@ -1,118 +1,118 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.oio; package io.netty.channel.socket.oio;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.SocketAddress; import java.net.SocketAddress;
import io.netty.channel.AbstractChannel; import io.netty.channel.AbstractChannel;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory; import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelSink; import io.netty.channel.ChannelSink;
import io.netty.channel.socket.Worker; import io.netty.channel.socket.Worker;
abstract class AbstractOioChannel extends AbstractChannel { abstract class AbstractOioChannel extends AbstractChannel {
private volatile InetSocketAddress localAddress; private volatile InetSocketAddress localAddress;
volatile InetSocketAddress remoteAddress; volatile InetSocketAddress remoteAddress;
volatile Thread workerThread; volatile Thread workerThread;
volatile Worker worker; volatile Worker worker;
final Object interestOpsLock = new Object(); final Object interestOpsLock = new Object();
AbstractOioChannel( AbstractOioChannel(
Channel parent, Channel parent,
ChannelFactory factory, ChannelFactory factory,
ChannelPipeline pipeline, ChannelPipeline pipeline,
ChannelSink sink) { ChannelSink sink) {
super(parent, factory, pipeline, sink); super(parent, factory, pipeline, sink);
} }
@Override @Override
protected boolean setClosed() { protected boolean setClosed() {
return super.setClosed(); return super.setClosed();
} }
@Override @Override
protected void setInterestOpsNow(int interestOps) { protected void setInterestOpsNow(int interestOps) {
super.setInterestOpsNow(interestOps); super.setInterestOpsNow(interestOps);
} }
@Override @Override
public ChannelFuture write(Object message, SocketAddress remoteAddress) { public ChannelFuture write(Object message, SocketAddress remoteAddress) {
if (remoteAddress == null || remoteAddress.equals(getRemoteAddress())) { if (remoteAddress == null || remoteAddress.equals(getRemoteAddress())) {
return super.write(message, null); return super.write(message, null);
} else { } else {
return super.write(message, remoteAddress); return super.write(message, remoteAddress);
} }
} }
@Override @Override
public boolean isBound() { public boolean isBound() {
return isOpen() && isSocketBound(); return isOpen() && isSocketBound();
} }
@Override @Override
public boolean isConnected() { public boolean isConnected() {
return isOpen() && isSocketConnected(); return isOpen() && isSocketConnected();
} }
@Override @Override
public InetSocketAddress getLocalAddress() { public InetSocketAddress getLocalAddress() {
InetSocketAddress localAddress = this.localAddress; InetSocketAddress localAddress = this.localAddress;
if (localAddress == null) { if (localAddress == null) {
try { try {
this.localAddress = localAddress = this.localAddress = localAddress =
(InetSocketAddress) getLocalSocketAddress(); (InetSocketAddress) getLocalSocketAddress();
} catch (Throwable t) { } catch (Throwable t) {
// Sometimes fails on a closed socket in Windows. // Sometimes fails on a closed socket in Windows.
return null; return null;
} }
} }
return localAddress; return localAddress;
} }
@Override @Override
public InetSocketAddress getRemoteAddress() { public InetSocketAddress getRemoteAddress() {
InetSocketAddress remoteAddress = this.remoteAddress; InetSocketAddress remoteAddress = this.remoteAddress;
if (remoteAddress == null) { if (remoteAddress == null) {
try { try {
this.remoteAddress = remoteAddress = this.remoteAddress = remoteAddress =
(InetSocketAddress) getRemoteSocketAddress(); (InetSocketAddress) getRemoteSocketAddress();
} catch (Throwable t) { } catch (Throwable t) {
// Sometimes fails on a closed socket in Windows. // Sometimes fails on a closed socket in Windows.
return null; return null;
} }
} }
return remoteAddress; return remoteAddress;
} }
abstract boolean isSocketBound(); abstract boolean isSocketBound();
abstract boolean isSocketConnected(); abstract boolean isSocketConnected();
abstract boolean isSocketClosed(); abstract boolean isSocketClosed();
abstract InetSocketAddress getLocalSocketAddress() throws Exception; abstract InetSocketAddress getLocalSocketAddress() throws Exception;
abstract InetSocketAddress getRemoteSocketAddress() throws Exception; abstract InetSocketAddress getRemoteSocketAddress() throws Exception;
abstract void closeSocket() throws IOException; abstract void closeSocket() throws IOException;
} }

114
transport/src/main/java/io/netty/channel/socket/oio/AbstractOioChannelSink.java
View File

@ -1,57 +1,57 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.oio; package io.netty.channel.socket.oio;
import io.netty.channel.AbstractChannelSink; import io.netty.channel.AbstractChannelSink;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelEvent; import io.netty.channel.ChannelEvent;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.ChannelRunnableWrapper; import io.netty.channel.socket.ChannelRunnableWrapper;
import io.netty.channel.socket.Worker; import io.netty.channel.socket.Worker;
public abstract class AbstractOioChannelSink extends AbstractChannelSink { public abstract class AbstractOioChannelSink extends AbstractChannelSink {
@Override @Override
public ChannelFuture execute(final ChannelPipeline pipeline, final Runnable task) { public ChannelFuture execute(final ChannelPipeline pipeline, final Runnable task) {
Channel ch = pipeline.channel(); Channel ch = pipeline.channel();
if (ch instanceof AbstractOioChannel) { if (ch instanceof AbstractOioChannel) {
AbstractOioChannel channel = (AbstractOioChannel) ch; AbstractOioChannel channel = (AbstractOioChannel) ch;
Worker worker = channel.worker; Worker worker = channel.worker;
if (worker != null) { if (worker != null) {
ChannelRunnableWrapper wrapper = new ChannelRunnableWrapper(pipeline.channel(), task); ChannelRunnableWrapper wrapper = new ChannelRunnableWrapper(pipeline.channel(), task);
channel.worker.executeInIoThread(wrapper); channel.worker.executeInIoThread(wrapper);
return wrapper; return wrapper;
} }
} }
return super.execute(pipeline, task); return super.execute(pipeline, task);
} }
@Override @Override
protected boolean isFireExceptionCaughtLater(ChannelEvent event, Throwable actualCause) { protected boolean isFireExceptionCaughtLater(ChannelEvent event, Throwable actualCause) {
Channel channel = event.getChannel(); Channel channel = event.getChannel();
boolean fireLater = false; boolean fireLater = false;
if (channel instanceof AbstractOioChannel) { if (channel instanceof AbstractOioChannel) {
fireLater = !AbstractOioWorker.isIoThread((AbstractOioChannel) channel); fireLater = !AbstractOioWorker.isIoThread((AbstractOioChannel) channel);
} }
return fireLater; return fireLater;
} }
} }

452
transport/src/main/java/io/netty/channel/socket/oio/AbstractOioWorker.java
View File

@ -1,226 +1,226 @@
/* /*
* Copyright 2011 The Netty Project * Copyright 2011 The Netty Project
* *
* The Netty Project licenses this file to you under the Apache License, * 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 * 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: * with the License. You may obtain a copy of the License at:
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations * License for the specific language governing permissions and limitations
* under the License. * under the License.
*/ */
package io.netty.channel.socket.oio; package io.netty.channel.socket.oio;
import static io.netty.channel.Channels.*; import static io.netty.channel.Channels.*;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.Channels; import io.netty.channel.Channels;
import io.netty.channel.socket.Worker; import io.netty.channel.socket.Worker;
import io.netty.util.internal.QueueFactory; import io.netty.util.internal.QueueFactory;
import java.io.IOException; import java.io.IOException;
import java.util.Queue; import java.util.Queue;
/** /**
* Abstract base class for Oio-Worker implementations * Abstract base class for Oio-Worker implementations
* *
* @param <C> {@link AbstractOioChannel} * @param <C> {@link AbstractOioChannel}
*/ */
abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker { abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker {
private final Queue<Runnable> eventQueue = QueueFactory.createQueue(Runnable.class); private final Queue<Runnable> eventQueue = QueueFactory.createQueue(Runnable.class);
protected final C channel; protected final C channel;
/** /**
* If this worker has been started thread will be a reference to the thread * If this worker has been started thread will be a reference to the thread
* used when starting. i.e. the current thread when the run method is executed. * used when starting. i.e. the current thread when the run method is executed.
*/ */
protected volatile Thread thread; protected volatile Thread thread;
public AbstractOioWorker(C channel) { public AbstractOioWorker(C channel) {
this.channel = channel; this.channel = channel;
channel.worker = this; channel.worker = this;
} }
@Override @Override
public void run() { public void run() {
thread = channel.workerThread = Thread.currentThread(); thread = channel.workerThread = Thread.currentThread();
while (channel.isOpen()) { while (channel.isOpen()) {
synchronized (channel.interestOpsLock) { synchronized (channel.interestOpsLock) {
while (!channel.isReadable()) { while (!channel.isReadable()) {
try { try {
// notify() is not called at all. // notify() is not called at all.
// close() and setInterestOps() calls Thread.interrupt() // close() and setInterestOps() calls Thread.interrupt()
channel.interestOpsLock.wait(); channel.interestOpsLock.wait();
} catch (InterruptedException e) { } catch (InterruptedException e) {
if (!channel.isOpen()) { if (!channel.isOpen()) {
break; break;
} }
} }
} }
} }
try { try {
boolean cont = process(); boolean cont = process();
processEventQueue(); processEventQueue();
if (!cont) { if (!cont) {
break; break;
} }
} catch (Throwable t) { } catch (Throwable t) {
if (!channel.isSocketClosed()) { if (!channel.isSocketClosed()) {
fireExceptionCaught(channel, t); fireExceptionCaught(channel, t);
} }
break; break;
} }
} }
// Setting the workerThread to null will prevent any channel // Setting the workerThread to null will prevent any channel
// operations from interrupting this thread from now on. // operations from interrupting this thread from now on.
channel.workerThread = null; channel.workerThread = null;
// Clean up. // Clean up.
close(channel, succeededFuture(channel), true); close(channel, succeededFuture(channel), true);
} }
static boolean isIoThread(AbstractOioChannel channel) { static boolean isIoThread(AbstractOioChannel channel) {
return Thread.currentThread() == channel.workerThread; return Thread.currentThread() == channel.workerThread;
} }
@Override @Override
public void executeInIoThread(Runnable task) { public void executeInIoThread(Runnable task) {
if (Thread.currentThread() == thread) { if (Thread.currentThread() == thread) {
task.run(); task.run();
} else { } else {
boolean added = eventQueue.offer(task); boolean added = eventQueue.offer(task);
if (added) { if (added) {
// as we set the SO_TIMEOUT to 1 second this task will get picked up in 1 second at latest // as we set the SO_TIMEOUT to 1 second this task will get picked up in 1 second at latest
} }
} }
} }
private void processEventQueue() throws IOException { private void processEventQueue() throws IOException {
for (;;) { for (;;) {
final Runnable task = eventQueue.poll(); final Runnable task = eventQueue.poll();
if (task == null) { if (task == null) {
break; break;
} }
task.run(); task.run();
} }
} }
/** /**
* Process the incoming messages and also is responsible for call {@link Channels#fireMessageReceived(Channel, Object)} once a message * Process the incoming messages and also is responsible for call {@link Channels#fireMessageReceived(Channel, Object)} once a message
* was processed without errors. * was processed without errors.
* *
* @return continue returns <code>true</code> as long as this worker should continue to try processing incoming messages * @return continue returns <code>true</code> as long as this worker should continue to try processing incoming messages
* @throws IOException * @throws IOException
*/ */
abstract boolean process() throws IOException; abstract boolean process() throws IOException;
static void setInterestOps( static void setInterestOps(
AbstractOioChannel channel, ChannelFuture future, int interestOps) { AbstractOioChannel channel, ChannelFuture future, int interestOps) {
boolean iothread = isIoThread(channel); boolean iothread = isIoThread(channel);
// Override OP_WRITE flag - a user cannot change this flag. // Override OP_WRITE flag - a user cannot change this flag.
interestOps &= ~Channel.OP_WRITE; interestOps &= ~Channel.OP_WRITE;
interestOps |= channel.getInterestOps() & Channel.OP_WRITE; interestOps |= channel.getInterestOps() & Channel.OP_WRITE;
boolean changed = false; boolean changed = false;
try { try {
if (channel.getInterestOps() != interestOps) { if (channel.getInterestOps() != interestOps) {
if ((interestOps & Channel.OP_READ) != 0) { if ((interestOps & Channel.OP_READ) != 0) {
channel.setInterestOpsNow(Channel.OP_READ); channel.setInterestOpsNow(Channel.OP_READ);
} else { } else {
channel.setInterestOpsNow(Channel.OP_NONE); channel.setInterestOpsNow(Channel.OP_NONE);
} }
changed = true; changed = true;
} }
future.setSuccess(); future.setSuccess();
if (changed) { if (changed) {
synchronized (channel.interestOpsLock) { synchronized (channel.interestOpsLock) {
channel.setInterestOpsNow(interestOps); channel.setInterestOpsNow(interestOps);
// Notify the worker so it stops or continues reading. // Notify the worker so it stops or continues reading.
Thread currentThread = Thread.currentThread(); Thread currentThread = Thread.currentThread();
Thread workerThread = channel.workerThread; Thread workerThread = channel.workerThread;
if (workerThread != null && currentThread != workerThread) { if (workerThread != null && currentThread != workerThread) {
workerThread.interrupt(); workerThread.interrupt();
} }
} }
if (iothread) { if (iothread) {
fireChannelInterestChanged(channel); fireChannelInterestChanged(channel);
} else { } else {
fireChannelInterestChangedLater(channel); fireChannelInterestChangedLater(channel);
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
if (iothread) { if (iothread) {
fireExceptionCaught(channel, t); fireExceptionCaught(channel, t);
} else { } else {
fireExceptionCaughtLater(channel, t); fireExceptionCaughtLater(channel, t);
} }
} }
} }
static void close(AbstractOioChannel channel, ChannelFuture future) { static void close(AbstractOioChannel channel, ChannelFuture future) {
close(channel, future, isIoThread(channel)); close(channel, future, isIoThread(channel));
} }
private static void close(AbstractOioChannel channel, ChannelFuture future, boolean iothread) { private static void close(AbstractOioChannel channel, ChannelFuture future, boolean iothread) {
boolean connected = channel.isConnected(); boolean connected = channel.isConnected();
boolean bound = channel.isBound(); boolean bound = channel.isBound();
try { try {
channel.closeSocket(); channel.closeSocket();
if (channel.setClosed()) { if (channel.setClosed()) {
future.setSuccess(); future.setSuccess();
if (connected) { if (connected) {
// Notify the worker so it stops reading. // Notify the worker so it stops reading.
Thread currentThread = Thread.currentThread(); Thread currentThread = Thread.currentThread();
Thread workerThread = channel.workerThread; Thread workerThread = channel.workerThread;
if (workerThread != null && currentThread != workerThread) { if (workerThread != null && currentThread != workerThread) {
workerThread.interrupt(); workerThread.interrupt();
} }
if (iothread) { if (iothread) {
fireChannelDisconnected(channel); fireChannelDisconnected(channel);
} else { } else {
fireChannelDisconnectedLater(channel); fireChannelDisconnectedLater(channel);
} }
} }
if (bound) { if (bound) {
if (iothread) { if (iothread) {
fireChannelUnbound(channel); fireChannelUnbound(channel);
} else { } else {
fireChannelUnboundLater(channel); fireChannelUnboundLater(channel);
} }
} }
if (iothread) { if (iothread) {
fireChannelClosed(channel); fireChannelClosed(channel);
} else { } else {
fireChannelClosedLater(channel); fireChannelClosedLater(channel);
} }
} else { } else {
future.setSuccess(); future.setSuccess();
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
if (iothread) { if (iothread) {
fireExceptionCaught(channel, t); fireExceptionCaught(channel, t);
} else { } else {
fireExceptionCaughtLater(channel, t); fireExceptionCaughtLater(channel, t);
} }
} }
} }
} }