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

Resolved issue: NETTY-262 Static ChannelPipeline implementation

Browse Source 
* Added StaticChannelPipeline
* Changed all Bootstrap implementations to work with StaticChannelPipeline
* Changed ServerBootstrap and AbstractCodecEmbedder to leverage StaticChannelPipeline
* StackTraceSimplifier should recognize StaticChannelPipeline
This commit is contained in:
Trustin Lee 2009-12-18 06:03:37 +00:00
parent 9951e19573
commit 553f8cf1e9
6 changed files with 595 additions and 229 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
src/main/java/org/jboss/netty/bootstrap/ClientBootstrap.java
View File

@ -15,28 +15,18 @@
*/
package org.jboss.netty.bootstrap;
import static org.jboss.netty.channel.Channels.*;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.channels.NotYetConnectedException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelConfig;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelPipelineException;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
/**
* A helper class which creates a new client-side {@link Channel} and makes a
@ -211,9 +201,6 @@ public class ClientBootstrap extends Bootstrap {
throw new NullPointerException("remoteAddress");
}
final BlockingQueue<ChannelFuture> futureQueue =
new LinkedBlockingQueue<ChannelFuture>();
ChannelPipeline pipeline;
try {
pipeline = getPipelineFactory().getPipeline();
@ -221,97 +208,16 @@ public class ClientBootstrap extends Bootstrap {
throw new ChannelPipelineException("Failed to initialize a pipeline.", e);
}
pipeline.addFirst(
"connector", new Connector(
this, remoteAddress, localAddress, futureQueue));
// Set the options.
Channel ch = getFactory().newChannel(pipeline);
ch.getConfig().setOptions(getOptions());
getFactory().newChannel(pipeline);
// Wait until the future is available.
ChannelFuture future = null;
boolean interrupted = false;
do {
try {
future = futureQueue.poll(Integer.MAX_VALUE, TimeUnit.SECONDS);
} catch (InterruptedException e) {
interrupted = true;
}
} while (future == null);
pipeline.remove("connector");
if (interrupted) {
Thread.currentThread().interrupt();
}
return future;
}
@ChannelPipelineCoverage("one")
static final class Connector extends SimpleChannelUpstreamHandler {
private final Bootstrap bootstrap;
private final SocketAddress localAddress;
private final BlockingQueue<ChannelFuture> futureQueue;
private final SocketAddress remoteAddress;
private volatile boolean finished = false;
Connector(
Bootstrap bootstrap,
SocketAddress remoteAddress,
SocketAddress localAddress,
BlockingQueue<ChannelFuture> futureQueue) {
this.bootstrap = bootstrap;
this.localAddress = localAddress;
this.futureQueue = futureQueue;
this.remoteAddress = remoteAddress;
}
@Override
public void channelOpen(
ChannelHandlerContext context,
ChannelStateEvent event) {
try {
// Apply options.
event.getChannel().getConfig().setOptions(bootstrap.getOptions());
} finally {
context.sendUpstream(event);
}
// Bind or connect.
// Bind.
if (localAddress != null) {
event.getChannel().bind(localAddress);
} else {
finished = futureQueue.offer(event.getChannel().connect(remoteAddress));
assert finished;
}
ch.bind(localAddress);
}
@Override
public void channelBound(
ChannelHandlerContext context,
ChannelStateEvent event) {
context.sendUpstream(event);
// Connect if not connected yet.
if (localAddress != null) {
finished = futureQueue.offer(event.getChannel().connect(remoteAddress));
assert finished;
}
}
@Override
public void exceptionCaught(
ChannelHandlerContext ctx, ExceptionEvent e)
throws Exception {
ctx.sendUpstream(e);
Throwable cause = e.getCause();
if (!(cause instanceof NotYetConnectedException) && !finished) {
e.getChannel().close();
finished = futureQueue.offer(failedFuture(e.getChannel(), cause));
assert finished;
}
}
// Connect.
return ch.connect(remoteAddress);
}
}

109
src/main/java/org/jboss/netty/bootstrap/ConnectionlessBootstrap.java
View File

@ -15,13 +15,8 @@
*/
package org.jboss.netty.bootstrap;
import static org.jboss.netty.channel.Channels.*;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelConfig;
@ -29,14 +24,10 @@ import org.jboss.netty.channel.ChannelException;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelPipelineException;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
/**
* A helper class which creates a new server-side {@link Channel} for a
@ -183,9 +174,6 @@ public class ConnectionlessBootstrap extends Bootstrap {
throw new NullPointerException("localAddress");
}
final BlockingQueue<ChannelFuture> futureQueue =
new LinkedBlockingQueue<ChannelFuture>();
ChannelPipeline pipeline;
try {
pipeline = getPipelineFactory().getPipeline();
@ -193,26 +181,14 @@ public class ConnectionlessBootstrap extends Bootstrap {
throw new ChannelPipelineException("Failed to initialize a pipeline.", e);
}
pipeline.addFirst("binder", new ConnectionlessBinder(localAddress, futureQueue));
Channel ch = getFactory().newChannel(pipeline);
Channel channel = getFactory().newChannel(pipeline);
// Apply options.
ch.getConfig().setPipelineFactory(getPipelineFactory());
ch.getConfig().setOptions(getOptions());
// Wait until the future is available.
ChannelFuture future = null;
boolean interrupted = false;
do {
try {
future = futureQueue.poll(Integer.MAX_VALUE, TimeUnit.SECONDS);
} catch (InterruptedException e) {
interrupted = true;
}
} while (future == null);
pipeline.remove("binder");
if (interrupted) {
Thread.currentThread().interrupt();
}
// Bind
ChannelFuture future = ch.bind(localAddress);
// Wait for the future.
future.awaitUninterruptibly();
@ -221,7 +197,7 @@ public class ConnectionlessBootstrap extends Bootstrap {
throw new ChannelException("Failed to bind to: " + localAddress, future.getCause());
}
return channel;
return ch;
}
/**
@ -304,9 +280,6 @@ public class ConnectionlessBootstrap extends Bootstrap {
throw new NullPointerException("remoteAddress");
}
final BlockingQueue<ChannelFuture> futureQueue =
new LinkedBlockingQueue<ChannelFuture>();
ChannelPipeline pipeline;
try {
pipeline = getPipelineFactory().getPipeline();
@ -314,68 +287,16 @@ public class ConnectionlessBootstrap extends Bootstrap {
throw new ChannelPipelineException("Failed to initialize a pipeline.", e);
}
pipeline.addFirst(
"connector", new ClientBootstrap.Connector(
this, remoteAddress, localAddress, futureQueue));
// Set the options.
Channel ch = getFactory().newChannel(pipeline);
ch.getConfig().setOptions(getOptions());
getFactory().newChannel(pipeline);
// Wait until the future is available.
ChannelFuture future = null;
boolean interrupted = false;
do {
try {
future = futureQueue.poll(Integer.MAX_VALUE, TimeUnit.SECONDS);
} catch (InterruptedException e) {
interrupted = true;
}
} while (future == null);
pipeline.remove("connector");
if (interrupted) {
Thread.currentThread().interrupt();
// Bind.
if (localAddress != null) {
ch.bind(localAddress);
}
return future;
}
@ChannelPipelineCoverage("one")
private final class ConnectionlessBinder extends SimpleChannelUpstreamHandler {
private final SocketAddress localAddress;
private final BlockingQueue<ChannelFuture> futureQueue;
ConnectionlessBinder(SocketAddress localAddress, BlockingQueue<ChannelFuture> futureQueue) {
this.localAddress = localAddress;
this.futureQueue = futureQueue;
}
@Override
public void channelOpen(
ChannelHandlerContext ctx,
ChannelStateEvent evt) {
try {
evt.getChannel().getConfig().setPipelineFactory(getPipelineFactory());
// Apply options.
evt.getChannel().getConfig().setOptions(getOptions());
} finally {
ctx.sendUpstream(evt);
}
boolean finished = futureQueue.offer(evt.getChannel().bind(localAddress));
assert finished;
}
@Override
public void exceptionCaught(
ChannelHandlerContext ctx, ExceptionEvent e)
throws Exception {
boolean finished = futureQueue.offer(failedFuture(e.getChannel(), e.getCause()));
assert finished;
ctx.sendUpstream(e);
}
// Connect.
return ch.connect(remoteAddress);
}
}

10
src/main/java/org/jboss/netty/bootstrap/ServerBootstrap.java
View File

@ -41,6 +41,7 @@ import org.jboss.netty.channel.ChildChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.ServerChannelFactory;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.StaticChannelPipeline;
/**
* A helper class which creates a new server-side {@link Channel} and accepts
@ -264,12 +265,13 @@ public class ServerBootstrap extends Bootstrap {
final BlockingQueue<ChannelFuture> futureQueue =
new LinkedBlockingQueue<ChannelFuture>();
ChannelPipeline bossPipeline = pipeline();
bossPipeline.addLast("binder", new Binder(localAddress, futureQueue));
ChannelPipeline bossPipeline;
ChannelHandler binder = new Binder(localAddress, futureQueue);
ChannelHandler parentHandler = getParentHandler();
if (parentHandler != null) {
bossPipeline.addLast("userHandler", parentHandler);
bossPipeline = new StaticChannelPipeline(binder, parentHandler);
} else {
bossPipeline = new StaticChannelPipeline(binder);
}
Channel channel = getFactory().newChannel(bossPipeline);

547
src/main/java/org/jboss/netty/channel/StaticChannelPipeline.java Normal file
View File

@ -0,0 +1,547 @@
/*
* Copyright 2009 Red Hat, Inc.
*
* Red Hat licenses this file to you under the Apache License, version 2.0
* (the "License"); you may not use this file except in compliance with the
* License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package org.jboss.netty.channel;
import static org.jboss.netty.channel.ChannelPipelineCoverage.*;
import java.lang.annotation.AnnotationFormatError;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
/**
* A {@link ChannelPipeline} that might perform better at the cost of
* disabled dynamic insertion and removal of {@link ChannelHandler}s.
* An attempt to insert, remove, or replace a handler in this pipeline will
* trigger an {@link UnsupportedOperationException}.
*
* @author The Netty Project (netty-dev@lists.jboss.org)
* @author Trustin Lee (trustin@gmail.com)
*
* @version $Rev$, $Date$
*
*/
public class StaticChannelPipeline implements ChannelPipeline {
// FIXME Code duplication with DefaultChannelPipeline
static final InternalLogger logger = InternalLoggerFactory.getInstance(StaticChannelPipeline.class);
// The names of the first 16 handlers to avoid int->string conversion cost
private static final String[] NAMES = {
"0", "1", "2", "3", "4", "5", "6", "7", "8", "9",
"10","11","12","13","14","15",
};
private volatile Channel channel;
private volatile ChannelSink sink;
private final StaticChannelHandlerContext[] contexts;
private final int lastIndex;
private final Map<String, StaticChannelHandlerContext> name2ctx =
new HashMap<String, StaticChannelHandlerContext>(4);
/**
* Creates a new pipeline from the specified handlers.
* The names of the specified handlers are generated automatically;
* the first handler's name is {@code "0"}, the second handler's name is
* {@code "1"}, the third handler's name is {@code "2"}, and so on.
*/
public StaticChannelPipeline(ChannelHandler... handlers) {
if (handlers == null) {
throw new NullPointerException("handlers");
}
if (handlers.length == 0) {
throw new IllegalArgumentException("no handlers specified");
}
contexts = new StaticChannelHandlerContext[handlers.length];
lastIndex = contexts.length - 1;
for (int i = 0; i < contexts.length; i ++) {
ChannelHandler h = handlers[i];
if (h == null) {
throw new NullPointerException("handlers[" + i + ']');
}
String name;
if (i < NAMES.length) {
name = NAMES[i];
} else {
name = Integer.toString(i);
}
StaticChannelHandlerContext ctx =
new StaticChannelHandlerContext(i, name, h);
contexts[i] = ctx;
name2ctx.put(name, ctx);
}
for (ChannelHandlerContext ctx: contexts) {
callBeforeAdd(ctx);
callAfterAdd(ctx);
}
}
public Channel getChannel() {
return channel;
}
public ChannelSink getSink() {
ChannelSink sink = this.sink;
if (sink == null) {
return DefaultChannelPipeline.discardingSink;
}
return sink;
}
public void attach(Channel channel, ChannelSink sink) {
if (channel == null) {
throw new NullPointerException("channel");
}
if (sink == null) {
throw new NullPointerException("sink");
}
if (this.channel != null || this.sink != null) {
throw new IllegalStateException("attached already");
}
this.channel = channel;
this.sink = sink;
}
public boolean isAttached() {
return sink != null;
}
public void addFirst(String name, ChannelHandler handler) {
throw new UnsupportedOperationException();
}
public void addLast(String name, ChannelHandler handler) {
throw new UnsupportedOperationException();
}
public void addBefore(String baseName, String name, ChannelHandler handler) {
throw new UnsupportedOperationException();
}
public void addAfter(String baseName, String name, ChannelHandler handler) {
throw new UnsupportedOperationException();
}
public void remove(ChannelHandler handler) {
throw new UnsupportedOperationException();
}
public ChannelHandler remove(String name) {
throw new UnsupportedOperationException();
}
public <T extends ChannelHandler> T remove(Class<T> handlerType) {
throw new UnsupportedOperationException();
}
public ChannelHandler removeFirst() {
throw new UnsupportedOperationException();
}
public ChannelHandler removeLast() {
throw new UnsupportedOperationException();
}
public void replace(ChannelHandler oldHandler, String newName, ChannelHandler newHandler) {
throw new UnsupportedOperationException();
}
public ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler) {
throw new UnsupportedOperationException();
}
public <T extends ChannelHandler> T replace(
Class<T> oldHandlerType, String newName, ChannelHandler newHandler) {
throw new UnsupportedOperationException();
}
private void callBeforeAdd(ChannelHandlerContext ctx) {
if (!(ctx.getHandler() instanceof LifeCycleAwareChannelHandler)) {
return;
}
LifeCycleAwareChannelHandler h =
(LifeCycleAwareChannelHandler) ctx.getHandler();
try {
h.beforeAdd(ctx);
} catch (Throwable t) {
throw new ChannelHandlerLifeCycleException(
h.getClass().getName() +
".beforeAdd() has thrown an exception; not adding.", t);
}
}
private void callAfterAdd(ChannelHandlerContext ctx) {
if (!(ctx.getHandler() instanceof LifeCycleAwareChannelHandler)) {
return;
}
LifeCycleAwareChannelHandler h =
(LifeCycleAwareChannelHandler) ctx.getHandler();
try {
h.afterAdd(ctx);
} catch (Throwable t) {
boolean removed = false;
try {
callBeforeRemove(ctx);
callAfterRemove(ctx);
removed = true;
} catch (Throwable t2) {
logger.warn("Failed to remove a handler: " + ctx.getName(), t2);
}
if (removed) {
throw new ChannelHandlerLifeCycleException(
h.getClass().getName() +
".afterAdd() has thrown an exception; removed.", t);
} else {
throw new ChannelHandlerLifeCycleException(
h.getClass().getName() +
".afterAdd() has thrown an exception; also failed to remove.", t);
}
}
}
private void callBeforeRemove(ChannelHandlerContext ctx) {
if (!(ctx.getHandler() instanceof LifeCycleAwareChannelHandler)) {
return;
}
LifeCycleAwareChannelHandler h =
(LifeCycleAwareChannelHandler) ctx.getHandler();
try {
h.beforeRemove(ctx);
} catch (Throwable t) {
throw new ChannelHandlerLifeCycleException(
h.getClass().getName() +
".beforeRemove() has thrown an exception; not removing.", t);
}
}
private void callAfterRemove(ChannelHandlerContext ctx) {
if (!(ctx.getHandler() instanceof LifeCycleAwareChannelHandler)) {
return;
}
LifeCycleAwareChannelHandler h =
(LifeCycleAwareChannelHandler) ctx.getHandler();
try {
h.afterRemove(ctx);
} catch (Throwable t) {
throw new ChannelHandlerLifeCycleException(
h.getClass().getName() +
".afterRemove() has thrown an exception.", t);
}
}
public ChannelHandler getFirst() {
return contexts[0].getHandler();
}
public ChannelHandler getLast() {
return contexts[contexts.length - 1].getHandler();
}
public ChannelHandler get(String name) {
StaticChannelHandlerContext ctx = name2ctx.get(name);
if (ctx == null) {
return null;
} else {
return ctx.getHandler();
}
}
@SuppressWarnings("unchecked")
public <T extends ChannelHandler> T get(Class<T> handlerType) {
ChannelHandlerContext ctx = getContext(handlerType);
if (ctx == null) {
return null;
} else {
return (T) ctx.getHandler();
}
}
public ChannelHandlerContext getContext(String name) {
if (name == null) {
throw new NullPointerException("name");
}
return name2ctx.get(name);
}
public ChannelHandlerContext getContext(ChannelHandler handler) {
if (handler == null) {
throw new NullPointerException("handler");
}
for (StaticChannelHandlerContext ctx: contexts) {
if (ctx.getHandler() == handler) {
return ctx;
}
}
return null;
}
public ChannelHandlerContext getContext(Class<? extends ChannelHandler> handlerType) {
if (handlerType == null) {
throw new NullPointerException("handlerType");
}
for (StaticChannelHandlerContext ctx: contexts) {
if (handlerType.isAssignableFrom(ctx.getHandler().getClass())) {
return ctx;
}
}
return null;
}
public Map<String, ChannelHandler> toMap() {
Map<String, ChannelHandler> map = new LinkedHashMap<String, ChannelHandler>();
for (StaticChannelHandlerContext ctx: contexts) {
map.put(ctx.getName(), ctx.getHandler());
}
return map;
}
/**
* Returns the {@link String} representation of this pipeline.
*/
@Override
public String toString() {
StringBuilder buf = new StringBuilder();
buf.append(getClass().getSimpleName());
buf.append('{');
for (StaticChannelHandlerContext ctx: contexts) {
buf.append('(');
buf.append(ctx.getName());
buf.append(" = ");
buf.append(ctx.getHandler().getClass().getName());
buf.append(')');
buf.append(", ");
}
buf.replace(buf.length() - 2, buf.length(), "}");
return buf.toString();
}
public void sendUpstream(ChannelEvent e) {
StaticChannelHandlerContext head = getActualUpstreamContext(0);
if (head == null) {
logger.warn(
"The pipeline contains no upstream handlers; discarding: " + e);
return;
}
sendUpstream(head, e);
}
void sendUpstream(StaticChannelHandlerContext ctx, ChannelEvent e) {
try {
((ChannelUpstreamHandler) ctx.getHandler()).handleUpstream(ctx, e);
} catch (Throwable t) {
notifyHandlerException(e, t);
}
}
public void sendDownstream(ChannelEvent e) {
StaticChannelHandlerContext tail = getActualDownstreamContext(lastIndex);
if (tail == null) {
try {
getSink().eventSunk(this, e);
return;
} catch (Throwable t) {
notifyHandlerException(e, t);
return;
}
}
sendDownstream(tail, e);
}
void sendDownstream(StaticChannelHandlerContext ctx, ChannelEvent e) {
try {
((ChannelDownstreamHandler) ctx.getHandler()).handleDownstream(ctx, e);
} catch (Throwable t) {
notifyHandlerException(e, t);
}
}
StaticChannelHandlerContext getActualUpstreamContext(int index) {
for (int i = index; i < contexts.length; i ++) {
StaticChannelHandlerContext ctx = contexts[i];
if (ctx.canHandleUpstream()) {
return ctx;
}
}
return null;
}
StaticChannelHandlerContext getActualDownstreamContext(int index) {
for (int i = index; i >= 0; i --) {
StaticChannelHandlerContext ctx = contexts[i];
if (ctx.canHandleDownstream()) {
return ctx;
}
}
return null;
}
protected void notifyHandlerException(ChannelEvent e, Throwable t) {
if (e instanceof ExceptionEvent) {
logger.warn(
"An exception was thrown by a user handler " +
"while handling an exception event (" + e + ")", t);
return;
}
ChannelPipelineException pe;
if (t instanceof ChannelPipelineException) {
pe = (ChannelPipelineException) t;
} else {
pe = new ChannelPipelineException(t);
}
try {
sink.exceptionCaught(this, e, pe);
} catch (Exception e1) {
logger.warn("An exception was thrown by an exception handler.", e1);
}
}
private final class StaticChannelHandlerContext implements ChannelHandlerContext {
private final int index;
private final String name;
private final ChannelHandler handler;
private final boolean canHandleUpstream;
private final boolean canHandleDownstream;
private volatile Object attachment;
StaticChannelHandlerContext(
int index, String name, ChannelHandler handler) {
if (name == null) {
throw new NullPointerException("name");
}
if (handler == null) {
throw new NullPointerException("handler");
}
canHandleUpstream = handler instanceof ChannelUpstreamHandler;
canHandleDownstream = handler instanceof ChannelDownstreamHandler;
if (!canHandleUpstream && !canHandleDownstream) {
throw new IllegalArgumentException(
"handler must be either " +
ChannelUpstreamHandler.class.getName() + " or " +
ChannelDownstreamHandler.class.getName() + '.');
}
ChannelPipelineCoverage coverage = handler.getClass().getAnnotation(ChannelPipelineCoverage.class);
if (coverage == null) {
logger.warn(
"Handler '" + handler.getClass().getName() +
"' does not have a '" +
ChannelPipelineCoverage.class.getSimpleName() +
"' annotation with its class declaration. " +
"It is strongly recommended to add the annotation " +
"for a documentation purpose to tell if a single " +
"handler instance can handle more than one pipeline " +
"(\"" + ALL + "\") or not (\"" + ONE + "\")");
} else {
String coverageValue = coverage.value();
if (coverageValue == null) {
throw new AnnotationFormatError(
ChannelPipelineCoverage.class.getSimpleName() +
" annotation value is undefined for type: " +
handler.getClass().getName());
}
if (!coverageValue.equals(ALL) && !coverageValue.equals(ONE)) {
throw new AnnotationFormatError(
ChannelPipelineCoverage.class.getSimpleName() +
" annotation value: " + coverageValue +
" (must be either \"" + ALL + "\" or \"" + ONE + ")");
}
}
this.index = index;
this.name = name;
this.handler = handler;
}
public Channel getChannel() {
return getPipeline().getChannel();
}
public ChannelPipeline getPipeline() {
return StaticChannelPipeline.this;
}
public boolean canHandleDownstream() {
return canHandleDownstream;
}
public boolean canHandleUpstream() {
return canHandleUpstream;
}
public ChannelHandler getHandler() {
return handler;
}
public String getName() {
return name;
}
public Object getAttachment() {
return attachment;
}
public void setAttachment(Object attachment) {
this.attachment = attachment;
}
public void sendDownstream(ChannelEvent e) {
StaticChannelHandlerContext prev = getActualDownstreamContext(index - 1);
if (prev == null) {
try {
getSink().eventSunk(StaticChannelPipeline.this, e);
} catch (Throwable t) {
notifyHandlerException(e, t);
}
} else {
StaticChannelPipeline.this.sendDownstream(prev, e);
}
}
public void sendUpstream(ChannelEvent e) {
StaticChannelHandlerContext next = getActualUpstreamContext(index + 1);
if (next != null) {
StaticChannelPipeline.this.sendUpstream(next, e);
}
}
}
}

42
src/main/java/org/jboss/netty/handler/codec/embedder/AbstractCodecEmbedder.java
View File

@ -32,9 +32,9 @@ import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelPipelineException;
import org.jboss.netty.channel.ChannelSink;
import org.jboss.netty.channel.ChannelUpstreamHandler;
import org.jboss.netty.channel.DefaultChannelPipeline;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.StaticChannelPipeline;
/**
* A skeletal {@link CodecEmbedder} implementation.
@ -56,8 +56,17 @@ abstract class AbstractCodecEmbedder<E> implements CodecEmbedder<E> {
* handlers.
*/
protected AbstractCodecEmbedder(ChannelHandler... handlers) {
pipeline = new EmbeddedChannelPipeline();
configurePipeline(handlers);
if (handlers == null) {
throw new NullPointerException("handlers");
}
// Initialize the pipeline.
ChannelHandler[] tmp = new ChannelHandler[handlers.length + 1];
System.arraycopy(handlers, 0, tmp, 0, handlers.length);
tmp[handlers.length] = sink;
pipeline = new EmbeddedChannelPipeline(tmp);
// Initialize the channel.
channel = new EmbeddedChannel(pipeline, sink);
fireInitialEvents();
}
@ -81,27 +90,6 @@ abstract class AbstractCodecEmbedder<E> implements CodecEmbedder<E> {
fireChannelConnected(channel, channel.getRemoteAddress());
}
private void configurePipeline(ChannelHandler... handlers) {
if (handlers == null) {
throw new NullPointerException("handlers");
}
if (handlers.length == 0) {
throw new IllegalArgumentException(
"handlers should contain at least one " +
ChannelHandler.class.getSimpleName() + '.');
}
for (int i = 0; i < handlers.length; i ++) {
ChannelHandler h = handlers[i];
if (h == null) {
throw new NullPointerException("handlers[" + i + "]");
}
pipeline.addLast(String.valueOf(i), handlers[i]);
}
pipeline.addLast("SINK", sink);
}
public boolean finish() {
fireChannelDisconnected(channel);
fireChannelUnbound(channel);
@ -218,10 +206,10 @@ abstract class AbstractCodecEmbedder<E> implements CodecEmbedder<E> {
}
}
private static final class EmbeddedChannelPipeline extends DefaultChannelPipeline {
private static final class EmbeddedChannelPipeline extends StaticChannelPipeline {
EmbeddedChannelPipeline() {
super();
EmbeddedChannelPipeline(ChannelHandler... handlers) {
super(handlers);
}
@Override

6
src/main/java/org/jboss/netty/util/internal/StackTraceSimplifier.java
View File

@ -21,6 +21,7 @@ import java.util.regex.Pattern;
import org.jboss.netty.channel.DefaultChannelPipeline;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.StaticChannelPipeline;
import org.jboss.netty.util.DebugUtil;
import org.jboss.netty.util.ThreadRenamingRunnable;
@ -42,12 +43,13 @@ public class StackTraceSimplifier {
Pattern.compile(
"^org\\.jboss\\.netty\\." +
"(util\\.(ThreadRenamingRunnable)" +
"|channel\\.(SimpleChannel(Upstream|Downstream)?Handler|DefaultChannelPipeline.*))$");
"|channel\\.(SimpleChannel(Upstream|Downstream)?Handler|(Default|Static)ChannelPipeline.*))$");
/**
* Removes unnecessary {@link StackTraceElement}s from the specified
* exception. {@link ThreadRenamingRunnable}, {@link SimpleChannelHandler},
* and {@link DefaultChannelPipeline} will be dropped from the trace.
* {@link DefaultChannelPipeline}, and {@link StaticChannelPipeline}
* will be dropped from the trace.
*/
public static void simplify(Throwable e) {
if (!SIMPLIFY_STACK_TRACE) {