Merge remote branch 'upstream/master'

This commit is contained in:
Jestan Nirojan 2011-10-10 03:34:46 +05:30
commit 8f25312be8
85 changed files with 1418 additions and 537 deletions

View File

@ -337,7 +337,7 @@ public class TimeServerHandler extends &SimpleChannelHandler; {
&Channel; ch = e.getChannel();
&ChannelBuffer; time = &ChannelBuffers;.buffer(4);<co id="example.time.co2"/>
time.writeInt(System.currentTimeMillis() / 1000);
time.writeInt((int) (System.currentTimeMillis() / 1000));
&ChannelFuture; f = ch.write(time);<co id="example.time.co3"/>

View File

@ -979,8 +979,8 @@ public class ChannelBuffers {
}
for (int i = byteCount; i > 0; i --) {
byte va = bufferA.getByte(aIndex);
byte vb = bufferB.getByte(bIndex);
short va = bufferA.getUnsignedByte(aIndex);
short vb = bufferB.getUnsignedByte(bIndex);
if (va > vb) {
return 1;
} else if (va < vb) {

View File

@ -32,7 +32,6 @@ import org.jboss.netty.util.internal.ConcurrentHashMap;
public abstract class AbstractChannel implements Channel {
static final ConcurrentMap<Integer, Channel> allChannels = new ConcurrentHashMap<Integer, Channel>();
private static final IdDeallocator ID_DEALLOCATOR = new IdDeallocator();
private static Integer allocateId(Channel channel) {
Integer id = Integer.valueOf(System.identityHashCode(channel));
@ -49,17 +48,6 @@ public abstract class AbstractChannel implements Channel {
}
}
private static final class IdDeallocator implements ChannelFutureListener {
IdDeallocator() {
super();
}
@Override
public void operationComplete(ChannelFuture future) throws Exception {
allChannels.remove(future.getChannel().getId());
}
}
private final Integer id;
private final Channel parent;
private final ChannelFactory factory;
@ -94,7 +82,6 @@ public abstract class AbstractChannel implements Channel {
this.pipeline = pipeline;
id = allocateId(this);
closeFuture.addListener(ID_DEALLOCATOR);
pipeline.attach(this, sink);
}
@ -200,6 +187,10 @@ public abstract class AbstractChannel implements Channel {
* closed yet
*/
protected boolean setClosed() {
// Deallocate the current channel's ID from allChannels so that other
// new channels can use it.
allChannels.remove(id);
return closeFuture.setClosed();
}

View File

@ -51,7 +51,7 @@ package org.jboss.netty.channel;
* public void login(String username, password) {
* {@link Channels}.write(
* <b>this.ctx</b>,
* {@link Channels}.succeededFuture(<b>this.ctx</t>.getChannel()),
* {@link Channels}.succeededFuture(<b>this.ctx</t>.getChannel()</b>),
* new LoginMessage(username, password));
* }
* ...

View File

@ -17,6 +17,7 @@ package org.jboss.netty.channel;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
@ -488,6 +489,11 @@ public interface ChannelPipeline {
*/
boolean isAttached();
/**
* Returns the {@link List} of the handler names.
*/
List<String> getNames();
/**
* Converts this pipeline into an ordered {@link Map} whose keys are
* handler names and whose values are handlers.

View File

@ -15,8 +15,10 @@
*/
package org.jboss.netty.channel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
@ -510,6 +512,24 @@ public class DefaultChannelPipeline implements ChannelPipeline {
return null;
}
@Override
public List<String> getNames() {
List<String> list = new ArrayList<String>();
if (name2ctx.isEmpty()) {
return list;
}
DefaultChannelHandlerContext ctx = head;
for (;;) {
list.add(ctx.getName());
ctx = ctx.next;
if (ctx == null) {
break;
}
}
return list;
}
@Override
public Map<String, ChannelHandler> toMap() {
Map<String, ChannelHandler> map = new LinkedHashMap<String, ChannelHandler>();
@ -590,9 +610,19 @@ public class DefaultChannelPipeline implements ChannelPipeline {
}
void sendDownstream(DefaultChannelHandlerContext ctx, ChannelEvent e) {
if (e instanceof UpstreamMessageEvent) {
throw new IllegalArgumentException("cannot send an upstream event to downstream");
}
try {
((ChannelDownstreamHandler) ctx.getHandler()).handleDownstream(ctx, e);
} catch (Throwable t) {
// Unlike an upstream event, a downstream event usually has an
// incomplete future which is supposed to be updated by ChannelSink.
// However, if an exception is raised before the event reaches at
// ChannelSink, the future is not going to be updated, so we update
// here.
e.getFuture().setFailure(t);
notifyHandlerException(e, t);
}
}

View File

@ -48,12 +48,21 @@ import org.jboss.netty.util.ExternalResourceReleasable;
* transfer, sending a file with {@link FileRegion} might fail or yield worse
* performance. For example, sending a large file doesn't work well in Windows.
*
* <h3>Not all transports support it</h3>
*
* Currently, the NIO transport is the only transport that supports {@link FileRegion}.
* Attempting to write a {@link FileRegion} to non-NIO {@link Channel} will trigger
* a {@link ClassCastException} or a similar exception.
*
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
*/
public interface FileRegion extends ExternalResourceReleasable {
// FIXME Make sure all transports support writing a FileRegion
// Even if zero copy cannot be achieved, all transports should emulate it.
/**
* Returns the offset in the file where the transfer began.
*/

View File

@ -15,8 +15,10 @@
*/
package org.jboss.netty.channel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.jboss.netty.logging.InternalLogger;
@ -340,6 +342,15 @@ public class StaticChannelPipeline implements ChannelPipeline {
return null;
}
@Override
public List<String> getNames() {
List<String> list = new ArrayList<String>();
for (StaticChannelHandlerContext ctx: contexts) {
list.add(ctx.getName());
}
return list;
}
@Override
public Map<String, ChannelHandler> toMap() {
Map<String, ChannelHandler> map = new LinkedHashMap<String, ChannelHandler>();
@ -407,9 +418,19 @@ public class StaticChannelPipeline implements ChannelPipeline {
}
void sendDownstream(StaticChannelHandlerContext ctx, ChannelEvent e) {
if (e instanceof UpstreamMessageEvent) {
throw new IllegalArgumentException("cannot send an upstream event to downstream");
}
try {
((ChannelDownstreamHandler) ctx.getHandler()).handleDownstream(ctx, e);
} catch (Throwable t) {
// Unlike an upstream event, a downstream event usually has an
// incomplete future which is supposed to be updated by ChannelSink.
// However, if an exception is raised before the event reaches at
// ChannelSink, the future is not going to be updated, so we update
// here.
e.getFuture().setFailure(t);
notifyHandlerException(e, t);
}
}

View File

@ -152,17 +152,18 @@ public class DefaultChannelGroupFuture implements ChannelGroupFuture {
@Override
public synchronized boolean isPartialSuccess() {
return !futures.isEmpty() && successCount != 0;
return successCount != 0 && successCount != futures.size();
}
@Override
public synchronized boolean isPartialFailure() {
return !futures.isEmpty() && failureCount != 0;
return failureCount != 0 && failureCount != futures.size();
}
@Override
public synchronized boolean isCompleteFailure() {
return failureCount == futures.size();
int futureCnt = futures.size();
return futureCnt != 0 && failureCount == futureCnt;
}
@Override

View File

@ -41,6 +41,7 @@ class AcceptedServerChannelPipelineFactory implements ChannelPipelineFactory {
this.messageSwitch = messageSwitch;
}
@Override
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();

View File

@ -109,7 +109,7 @@ class AcceptedServerChannelRequestDispatch extends SimpleChannelUpstreamHandler
LOG.debug("send data request received for tunnel " + tunnelId);
}
if (HttpHeaders.getContentLength(request) == 0 ||
if (HttpHeaders.getContentLength(request, 0) == 0 ||
request.getContent() == null ||
request.getContent().readableBytes() == 0) {
respondWithRejection(ctx, request,

View File

@ -47,6 +47,7 @@ class ChannelFutureAggregator implements ChannelFutureListener {
future.addListener(this);
}
@Override
public synchronized void operationComplete(ChannelFuture future)
throws Exception {
if (future.isCancelled()) {

View File

@ -39,6 +39,7 @@ public class DefaultTunnelIdGenerator implements TunnelIdGenerator {
this.generator = generator;
}
@Override
public synchronized String generateId() {
// synchronized to ensure that this code is thread safe. The Sun
// standard implementations seem to be synchronized or lock free

View File

@ -62,36 +62,44 @@ class HttpTunnelAcceptedChannel extends AbstractChannel implements
fireChannelConnected(this, getRemoteAddress());
}
@Override
public SocketChannelConfig getConfig() {
return config;
}
@Override
public InetSocketAddress getLocalAddress() {
return ((HttpTunnelServerChannel) getParent()).getLocalAddress();
}
@Override
public InetSocketAddress getRemoteAddress() {
return remoteAddress;
}
@Override
public boolean isBound() {
return sink.isActive();
}
@Override
public boolean isConnected() {
return sink.isActive();
}
@Override
public void clientClosed() {
this.setClosed();
Channels.fireChannelClosed(this);
}
@Override
public void dataReceived(ChannelBuffer data) {
Channels.fireMessageReceived(this, data);
}
@Override
public void updateInterestOps(SaturationStateChange transition) {
switch (transition) {
case SATURATED:

View File

@ -39,15 +39,15 @@ import org.jboss.netty.channel.MessageEvent;
*/
class HttpTunnelAcceptedChannelSink extends AbstractChannelSink {
private final SaturationManager saturationManager;
final SaturationManager saturationManager;
private final ServerMessageSwitchDownstreamInterface messageSwitch;
private final String tunnelId;
private AtomicBoolean active = new AtomicBoolean(false);
private final AtomicBoolean active = new AtomicBoolean(false);
private HttpTunnelAcceptedChannelConfig config;
private final HttpTunnelAcceptedChannelConfig config;
public HttpTunnelAcceptedChannelSink(
ServerMessageSwitchDownstreamInterface messageSwitch,
@ -55,11 +55,12 @@ class HttpTunnelAcceptedChannelSink extends AbstractChannelSink {
this.messageSwitch = messageSwitch;
this.tunnelId = tunnelId;
this.config = config;
this.saturationManager =
saturationManager =
new SaturationManager(config.getWriteBufferLowWaterMark(),
config.getWriteBufferHighWaterMark());
}
@Override
public void eventSunk(ChannelPipeline pipeline, ChannelEvent e)
throws Exception {
if (e instanceof MessageEvent) {

View File

@ -17,6 +17,7 @@ package org.jboss.netty.channel.socket.http;
import org.jboss.netty.channel.DefaultChannelConfig;
import org.jboss.netty.channel.socket.SocketChannelConfig;
import org.jboss.netty.channel.socket.nio.NioSocketChannelConfig;
/**
* Configuration for HTTP tunnels. Where possible, properties set on this configuration will
@ -80,18 +81,18 @@ public abstract class HttpTunnelChannelConfig extends DefaultChannelConfig
}
/**
* Similarly to {@link org.jboss.netty.channel.socket.nio.NioSocketChannelConfig#setWriteBufferHighWaterMark(int)
* Similarly to {@link org.jboss.netty.channel.socket.nio.NioSocketChannelConfig#setWriteBufferHighWaterMark(int)
* NioSocketChannelConfig.setWriteBufferHighWaterMark()},
* the high water mark refers to the buffer size at which a user of the channel should stop writing. When the
* number of queued bytes exceeds the high water mark, {@link org.jboss.netty.channel.Channel#isWritable() Channel.isWritable()} will
* return false. Once the number of queued bytes falls below the {@link #setWriteBufferLowWaterMark(int) low water mark},
* {@link org.jboss.netty.channel.Channel#isWritable() Channel.isWritable()} will return true again, indicating that the client
* {@link org.jboss.netty.channel.Channel#isWritable() Channel.isWritable()} will return true again, indicating that the client
* can begin to send more data.
*
*
* @param level the number of queued bytes required to flip {@link org.jboss.netty.channel.Channel#isWritable()} to
* false.
*
* @see {@link org.jboss.netty.channel.socket.nio.NioSocketChannelConfig#setWriteBufferHighWaterMark(int) NioSocketChannelConfig.setWriteBufferHighWaterMark()}
*
* @see NioSocketChannelConfig#setWriteBufferHighWaterMark(int)
*/
public void setWriteBufferHighWaterMark(int level) {
if (level <= writeBufferLowWaterMark) {
@ -117,11 +118,11 @@ public abstract class HttpTunnelChannelConfig extends DefaultChannelConfig
/**
* The low water mark refers to the "safe" size of the queued byte buffer at which more data can be enqueued. When
* the {@link #setWriteBufferHighWaterMark(int) high water mark} is exceeded, {@link org.jboss.netty.channel.Channel#isWritable() Channel.isWriteable()}
* the {@link #setWriteBufferHighWaterMark(int) high water mark} is exceeded, {@link org.jboss.netty.channel.Channel#isWritable() Channel.isWriteable()}
* will return false until the buffer drops below this level. By creating a sufficient gap between the high and low
* water marks, rapid oscillation between "write enabled" and "write disabled" can be avoided.
*
* @see {@link org.jboss.netty.channel.socket.nio.NioSocketChannelConfig#setWriteBufferLowWaterMark(int) NioSocketChannelConfig.setWriteBufferLowWaterMark()}
*
* @see org.jboss.netty.channel.socket.nio.NioSocketChannelConfig#setWriteBufferLowWaterMark(int)
*/
public void setWriteBufferLowWaterMark(int level) {
if (level >= writeBufferHighWaterMark) {

View File

@ -48,33 +48,33 @@ import org.jboss.netty.logging.InternalLoggerFactory;
public class HttpTunnelClientChannel extends AbstractChannel implements
SocketChannel {
private static final InternalLogger LOG = InternalLoggerFactory
static final InternalLogger LOG = InternalLoggerFactory
.getInstance(HttpTunnelClientChannel.class);
private final HttpTunnelClientChannelConfig config;
private final SocketChannel sendChannel;
final SocketChannel sendChannel;
private final SocketChannel pollChannel;
final SocketChannel pollChannel;
private volatile String tunnelId;
volatile String tunnelId;
private volatile ChannelFuture connectFuture;
volatile ChannelFuture connectFuture;
private volatile boolean connected;
volatile boolean connected;
private volatile boolean bound;
volatile boolean bound;
volatile InetSocketAddress serverAddress;
private volatile String serverHostName;
volatile String serverHostName;
private final WorkerCallbacks callbackProxy;
private final SaturationManager saturationManager;
/**
* @see {@link HttpTunnelClientChannelFactory#newChannel(ChannelPipeline)}
* @see HttpTunnelClientChannelFactory#newChannel(ChannelPipeline)
*/
protected HttpTunnelClientChannel(ChannelFactory factory,
ChannelPipeline pipeline, HttpTunnelClientChannelSink sink,
@ -82,7 +82,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
ChannelGroup realConnections) {
super(null, factory, pipeline, sink);
this.callbackProxy = new WorkerCallbacks();
callbackProxy = new WorkerCallbacks();
sendChannel = outboundFactory.newChannel(createSendPipeline());
pollChannel = outboundFactory.newChannel(createPollPipeline());
@ -100,26 +100,36 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
Channels.fireChannelOpen(this);
}
@Override
public HttpTunnelClientChannelConfig getConfig() {
return config;
}
@Override
public boolean isBound() {
return bound;
}
@Override
public boolean isConnected() {
return connected;
}
@Override
public InetSocketAddress getLocalAddress() {
return sendChannel.getLocalAddress();
}
@Override
public InetSocketAddress getRemoteAddress() {
return serverAddress;
}
@Override
protected boolean setClosed() {
return super.setClosed();
}
void onConnectRequest(ChannelFuture connectFuture,
InetSocketAddress remoteAddress) {
this.connectFuture = connectFuture;
@ -146,6 +156,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
pollChannel.disconnect().addListener(disconnectListener);
disconnectFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future)
throws Exception {
serverAddress = null;
@ -214,7 +225,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
return pipeline;
}
private void setTunnelIdForPollChannel() {
void setTunnelIdForPollChannel() {
HttpTunnelClientPollHandler pollHandler =
pollChannel.getPipeline()
.get(HttpTunnelClientPollHandler.class);
@ -230,6 +241,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
updateSaturationStatus(messageSize);
Channels.write(sendChannel, e.getMessage()).addListener(
new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future)
throws Exception {
if (future.isSuccess()) {
@ -242,7 +254,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
});
}
private void updateSaturationStatus(int queueSizeDelta) {
void updateSaturationStatus(int queueSizeDelta) {
SaturationStateChange transition =
saturationManager.queueSizeChanged(queueSizeDelta);
switch (transition) {
@ -279,6 +291,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
eventsLeft = new AtomicInteger(numToConsolidate);
}
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
futureFailed(future);
@ -315,7 +328,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
protected void futureFailed(ChannelFuture future) {
LOG.warn("Failed to close one of the child channels of tunnel " +
tunnelId);
HttpTunnelClientChannel.this.setClosed();
setClosed();
}
@Override
@ -323,7 +336,7 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
if (LOG.isDebugEnabled()) {
LOG.debug("Tunnel " + tunnelId + " closed");
}
HttpTunnelClientChannel.this.setClosed();
setClosed();
}
}
@ -332,20 +345,23 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
* Contains the implementing methods of HttpTunnelClientWorkerOwner, so that these are hidden
* from the public API.
*/
private class WorkerCallbacks implements HttpTunnelClientWorkerOwner {
class WorkerCallbacks implements HttpTunnelClientWorkerOwner {
@Override
public void onConnectRequest(ChannelFuture connectFuture,
InetSocketAddress remoteAddress) {
HttpTunnelClientChannel.this.onConnectRequest(connectFuture,
remoteAddress);
}
@Override
public void onTunnelOpened(String tunnelId) {
HttpTunnelClientChannel.this.tunnelId = tunnelId;
setTunnelIdForPollChannel();
Channels.connect(pollChannel, sendChannel.getRemoteAddress());
}
@Override
public void fullyEstablished() {
if (!bound) {
bound = true;
@ -359,10 +375,12 @@ public class HttpTunnelClientChannel extends AbstractChannel implements
getRemoteAddress());
}
@Override
public void onMessageReceived(ChannelBuffer content) {
Channels.fireMessageReceived(HttpTunnelClientChannel.this, content);
}
@Override
public String getServerHostName() {
if (serverHostName == null) {
serverHostName =

View File

@ -31,8 +31,8 @@ import org.jboss.netty.channel.socket.SocketChannelConfig;
* <th>Name</th><th>Associated setter method</th>
* </tr>
* <tr><td>{@code "proxyAddress"}</td><td>{@link #setProxyAddress(SocketAddress)}</td></tr>
* <tr><td>{@code "writeBufferHighWaterMark"}</td><td>{@link #setWriteBufferHighWaterMark(long)}</td></tr>
* <tr><td>{@code "writeBufferLowWaterMark"}</td><td>{@link #setWriteBufferLowWaterMark(long)}</td></tr>
* <tr><td>{@code "writeBufferHighWaterMark"}</td><td>{@link #setWriteBufferHighWaterMark(int)}</td></tr>
* <tr><td>{@code "writeBufferLowWaterMark"}</td><td>{@link #setWriteBufferLowWaterMark(int)}</td></tr>
* </table>
*
* @author The Netty Project (netty-dev@lists.jboss.org)
@ -88,39 +88,48 @@ public class HttpTunnelClientChannelConfig extends HttpTunnelChannelConfig {
/* GENERIC SOCKET CHANNEL CONFIGURATION */
@Override
public int getReceiveBufferSize() {
return pollChannelConfig.getReceiveBufferSize();
}
@Override
public int getSendBufferSize() {
return pollChannelConfig.getSendBufferSize();
}
@Override
public int getSoLinger() {
return pollChannelConfig.getSoLinger();
}
@Override
public int getTrafficClass() {
return pollChannelConfig.getTrafficClass();
}
@Override
public boolean isKeepAlive() {
return pollChannelConfig.isKeepAlive();
}
@Override
public boolean isReuseAddress() {
return pollChannelConfig.isReuseAddress();
}
@Override
public boolean isTcpNoDelay() {
return pollChannelConfig.isTcpNoDelay();
}
@Override
public void setKeepAlive(boolean keepAlive) {
pollChannelConfig.setKeepAlive(keepAlive);
sendChannelConfig.setKeepAlive(keepAlive);
}
@Override
public void setPerformancePreferences(int connectionTime, int latency,
int bandwidth) {
pollChannelConfig.setPerformancePreferences(connectionTime, latency,
@ -129,31 +138,37 @@ public class HttpTunnelClientChannelConfig extends HttpTunnelChannelConfig {
bandwidth);
}
@Override
public void setReceiveBufferSize(int receiveBufferSize) {
pollChannelConfig.setReceiveBufferSize(receiveBufferSize);
sendChannelConfig.setReceiveBufferSize(receiveBufferSize);
}
@Override
public void setReuseAddress(boolean reuseAddress) {
pollChannelConfig.setReuseAddress(reuseAddress);
sendChannelConfig.setReuseAddress(reuseAddress);
}
@Override
public void setSendBufferSize(int sendBufferSize) {
pollChannelConfig.setSendBufferSize(sendBufferSize);
sendChannelConfig.setSendBufferSize(sendBufferSize);
}
@Override
public void setSoLinger(int soLinger) {
pollChannelConfig.setSoLinger(soLinger);
sendChannelConfig.setSoLinger(soLinger);
}
@Override
public void setTcpNoDelay(boolean tcpNoDelay) {
pollChannelConfig.setTcpNoDelay(true);
sendChannelConfig.setTcpNoDelay(true);
}
@Override
public void setTrafficClass(int trafficClass) {
pollChannelConfig.setTrafficClass(1);
sendChannelConfig.setTrafficClass(1);

View File

@ -41,11 +41,13 @@ public class HttpTunnelClientChannelFactory implements
this.factory = factory;
}
@Override
public HttpTunnelClientChannel newChannel(ChannelPipeline pipeline) {
return new HttpTunnelClientChannel(this, pipeline,
new HttpTunnelClientChannelSink(), factory, realConnections);
}
@Override
public void releaseExternalResources() {
realConnections.close().awaitUninterruptibly();
factory.releaseExternalResources();

View File

@ -33,6 +33,7 @@ import org.jboss.netty.channel.MessageEvent;
*/
class HttpTunnelClientChannelSink extends AbstractChannelSink {
@Override
public void eventSunk(ChannelPipeline pipeline, ChannelEvent e)
throws Exception {
if (e instanceof ChannelStateEvent) {

View File

@ -241,7 +241,7 @@ public class HttpTunnelMessageUtils {
public static boolean hasContents(HttpResponse response,
byte[] expectedContents) {
if (response.getContent() != null &&
HttpHeaders.getContentLength(response) == expectedContents.length &&
HttpHeaders.getContentLength(response, 0) == expectedContents.length &&
response.getContent().readableBytes() == expectedContents.length) {
byte[] compareBytes = new byte[expectedContents.length];
response.getContent().readBytes(compareBytes);
@ -300,7 +300,7 @@ public class HttpTunnelMessageUtils {
public static Object extractErrorMessage(HttpResponse response) {
if (response.getContent() == null ||
HttpHeaders.getContentLength(response) == 0) {
HttpHeaders.getContentLength(response, 0) == 0) {
return "";
}

View File

@ -36,12 +36,13 @@ public class HttpTunnelServerChannel extends AbstractServerChannel implements
private final ServerSocketChannel realChannel;
private final HttpTunnelServerChannelConfig config;
final HttpTunnelServerChannelConfig config;
private final ServerMessageSwitch messageSwitch;
final ServerMessageSwitch messageSwitch;
private final ChannelFutureListener CLOSE_FUTURE_PROXY =
new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future)
throws Exception {
HttpTunnelServerChannel.this.setClosed();
@ -62,31 +63,45 @@ public class HttpTunnelServerChannel extends AbstractServerChannel implements
Channels.fireChannelOpen(this);
}
@Override
public ServerSocketChannelConfig getConfig() {
return config;
}
@Override
public InetSocketAddress getLocalAddress() {
return realChannel.getLocalAddress();
}
@Override
public InetSocketAddress getRemoteAddress() {
// server channels never have a remote address
return null;
}
@Override
public boolean isBound() {
return realChannel.isBound();
}
@Override
protected boolean setClosed() {
return super.setClosed();
}
/**
* Used to hide the newChannel method from the public API.
*/
private final class TunnelCreator implements
HttpTunnelAcceptedChannelFactory {
public HttpTunnelAcceptedChannelReceiver newChannel(String newTunnelId,
InetSocketAddress remoteAddress) {
TunnelCreator() {
super();
}
@Override
public HttpTunnelAcceptedChannelReceiver newChannel(
String newTunnelId, InetSocketAddress remoteAddress) {
ChannelPipeline childPipeline = null;
try {
childPipeline = getConfig().getPipelineFactory().getPipeline();
@ -103,6 +118,7 @@ public class HttpTunnelServerChannel extends AbstractServerChannel implements
getFactory(), childPipeline, sink, remoteAddress, config);
}
@Override
public String generateTunnelId() {
return config.getTunnelIdGenerator().generateId();
}

View File

@ -45,56 +45,69 @@ public class HttpTunnelServerChannelConfig implements ServerSocketChannelConfig
return realChannel.getConfig();
}
@Override
public int getBacklog() {
return getWrappedConfig().getBacklog();
}
@Override
public int getReceiveBufferSize() {
return getWrappedConfig().getReceiveBufferSize();
}
@Override
public boolean isReuseAddress() {
return getWrappedConfig().isReuseAddress();
}
@Override
public void setBacklog(int backlog) {
getWrappedConfig().setBacklog(backlog);
}
@Override
public void setPerformancePreferences(int connectionTime, int latency,
int bandwidth) {
getWrappedConfig().setPerformancePreferences(connectionTime, latency,
bandwidth);
}
@Override
public void setReceiveBufferSize(int receiveBufferSize) {
getWrappedConfig().setReceiveBufferSize(receiveBufferSize);
}
@Override
public void setReuseAddress(boolean reuseAddress) {
getWrappedConfig().setReuseAddress(reuseAddress);
}
@Override
public ChannelBufferFactory getBufferFactory() {
return getWrappedConfig().getBufferFactory();
}
@Override
public int getConnectTimeoutMillis() {
return getWrappedConfig().getConnectTimeoutMillis();
}
@Override
public ChannelPipelineFactory getPipelineFactory() {
return pipelineFactory;
}
@Override
public void setBufferFactory(ChannelBufferFactory bufferFactory) {
getWrappedConfig().setBufferFactory(bufferFactory);
}
@Override
public void setConnectTimeoutMillis(int connectTimeoutMillis) {
getWrappedConfig().setConnectTimeoutMillis(connectTimeoutMillis);
}
@Override
public boolean setOption(String name, Object value) {
if (name.equals("pipelineFactory")) {
setPipelineFactory((ChannelPipelineFactory) value);
@ -107,12 +120,14 @@ public class HttpTunnelServerChannelConfig implements ServerSocketChannelConfig
}
}
@Override
public void setOptions(Map<String, Object> options) {
for (Entry<String, Object> e: options.entrySet()) {
setOption(e.getKey(), e.getValue());
}
}
@Override
public void setPipelineFactory(ChannelPipelineFactory pipelineFactory) {
this.pipelineFactory = pipelineFactory;
}

View File

@ -40,6 +40,7 @@ public class HttpTunnelServerChannelFactory implements
realConnections = new DefaultChannelGroup();
}
@Override
public HttpTunnelServerChannel newChannel(ChannelPipeline pipeline) {
return new HttpTunnelServerChannel(this, pipeline);
}
@ -58,6 +59,7 @@ public class HttpTunnelServerChannelFactory implements
return newChannel;
}
@Override
public void releaseExternalResources() {
realConnections.close().awaitUninterruptibly();
realConnectionFactory.releaseExternalResources();

View File

@ -36,6 +36,7 @@ class HttpTunnelServerChannelSink extends AbstractChannelSink {
private ServerSocketChannel realChannel;
@Override
public void eventSunk(ChannelPipeline pipeline, ChannelEvent e)
throws Exception {
@ -67,6 +68,7 @@ class HttpTunnelServerChannelSink extends AbstractChannelSink {
this.upstreamFuture = upstreamFuture;
}
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
upstreamFuture.setSuccess();

View File

@ -38,7 +38,7 @@ import org.jboss.netty.logging.InternalLoggerFactory;
* ends of the http tunnel and the virtual server accepted tunnel. As a tunnel can last for longer than
* the lifetime of the client channels that are used to service it, this layer of abstraction is
* necessary.
*
*
* @author The Netty Project (netty-dev@lists.jboss.org)
* @author Iain McGinniss (iain.mcginniss@onedrum.com)
* @author OneDrum Ltd.
@ -62,6 +62,7 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
tunnelsById = new ConcurrentHashMap<String, TunnelInfo>();
}
@Override
public String createTunnel(InetSocketAddress remoteAddress) {
String newTunnelId =
String.format("%s_%s", tunnelIdPrefix,
@ -74,11 +75,13 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
return newTunnelId;
}
@Override
public boolean isOpenTunnel(String tunnelId) {
TunnelInfo tunnel = tunnelsById.get(tunnelId);
return tunnel != null;
}
@Override
public void pollOutboundData(String tunnelId, Channel channel) {
TunnelInfo tunnel = tunnelsById.get(tunnelId);
if (tunnel == null) {
@ -136,6 +139,7 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
new RelayedChannelFutureListener(originalFuture));
}
@Override
public TunnelStatus routeInboundData(String tunnelId,
ChannelBuffer inboundData) {
TunnelInfo tunnel = tunnelsById.get(tunnelId);
@ -156,14 +160,34 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
return TunnelStatus.ALIVE;
}
@Override
public void clientCloseTunnel(String tunnelId) {
TunnelInfo tunnel = tunnelsById.get(tunnelId);
if (tunnel == null) {
if (LOG.isWarnEnabled()) {
LOG.warn("attempt made to close tunnel id " +
tunnelId + " which is unknown or closed");
}
return;
}
tunnel.localChannel.clientClosed();
tunnelsById.remove(tunnelId);
}
@Override
public void serverCloseTunnel(String tunnelId) {
TunnelInfo tunnel = tunnelsById.get(tunnelId);
if (tunnel == null) {
if (LOG.isWarnEnabled()) {
LOG.warn("attempt made to close tunnel id " +
tunnelId + " which is unknown or closed");
}
return;
}
tunnel.closing.set(true);
Channel responseChannel = tunnel.responseChannel.getAndSet(null);
@ -179,6 +203,7 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
tunnelsById.remove(tunnelId);
}
@Override
public void routeOutboundData(String tunnelId, ChannelBuffer data,
ChannelFuture writeFuture) {
TunnelInfo tunnel = tunnelsById.get(tunnelId);
@ -217,10 +242,11 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
ChannelFutureListener {
private final ChannelFuture originalFuture;
private RelayedChannelFutureListener(ChannelFuture originalFuture) {
RelayedChannelFutureListener(ChannelFuture originalFuture) {
this.originalFuture = originalFuture;
}
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
originalFuture.setSuccess();
@ -231,6 +257,10 @@ class ServerMessageSwitch implements ServerMessageSwitchUpstreamInterface,
}
private static final class TunnelInfo {
TunnelInfo() {
super();
}
public String tunnelId;
public HttpTunnelAcceptedChannelReceiver localChannel;

View File

@ -568,15 +568,21 @@ class NioDatagramWorker implements Runnable {
}
}
channel.inWriteNowLoop = false;
// Initially, the following block was executed after releasing
// the writeLock, but there was a race condition, and it has to be
// executed before releasing the writeLock:
//
// https://issues.jboss.org/browse/NETTY-410
//
if (addOpWrite) {
setOpWrite(channel);
} else if (removeOpWrite) {
clearOpWrite(channel);
}
}
fireWriteComplete(channel, writtenBytes);
if (addOpWrite) {
setOpWrite(channel);
} else if (removeOpWrite) {
clearOpWrite(channel);
}
}
private void setOpWrite(final NioDatagramChannel channel) {

View File

@ -83,15 +83,6 @@ class NioSocketChannel extends AbstractChannel
this.socket = socket;
this.worker = worker;
config = new DefaultNioSocketChannelConfig(socket.socket());
// TODO Move the state variable to AbstractChannel so that we don't need
// to add many listeners.
getCloseFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
state = ST_CLOSED;
}
});
}
@Override
@ -157,6 +148,7 @@ class NioSocketChannel extends AbstractChannel
@Override
protected boolean setClosed() {
state = ST_CLOSED;
return super.setClosed();
}

View File

@ -506,17 +506,23 @@ class NioWorker implements Runnable {
}
}
channel.inWriteNowLoop = false;
// Initially, the following block was executed after releasing
// the writeLock, but there was a race condition, and it has to be
// executed before releasing the writeLock:
//
// https://issues.jboss.org/browse/NETTY-410
//
if (open) {
if (addOpWrite) {
setOpWrite(channel);
} else if (removeOpWrite) {
clearOpWrite(channel);
}
}
}
fireWriteComplete(channel, writtenBytes);
if (open) {
if (addOpWrite) {
setOpWrite(channel);
} else if (removeOpWrite) {
clearOpWrite(channel);
}
}
}
private void setOpWrite(NioSocketChannel channel) {

View File

@ -313,7 +313,7 @@ public class HttpStaticFileServerHandler extends SimpleChannelUpstreamHandler {
* file to extract content type
*/
private void setContentTypeHeader(HttpResponse response, File file) {
MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();
MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();
response.setHeader(HttpHeaders.Names.CONTENT_TYPE, mimeTypesMap.getContentType(file.getPath()));
}

View File

@ -52,7 +52,7 @@ public class HttpResponseHandler extends SimpleChannelUpstreamHandler {
System.out.println();
}
if (response.getStatus().getCode() == 200 && response.isChunked()) {
if (response.isChunked()) {
readingChunks = true;
System.out.println("CHUNKED CONTENT {");
} else {

View File

@ -128,17 +128,17 @@ public class HttpClient {
// Simple Get form: no factory used (not usable)
List<Entry<String,String>> headers =
formget(bootstrap, host, port, get, uriSimple);
formget(bootstrap, host, port, get, uriSimple);
if (headers == null) {
factory.cleanAllHttpDatas();
return;
factory.cleanAllHttpDatas();
return;
}
// Simple Post form: factory used for big attributes
List<InterfaceHttpData> bodylist =
formpost(bootstrap, host, port, uriSimple, file, factory, headers);
formpost(bootstrap, host, port, uriSimple, file, factory, headers);
if (bodylist == null) {
factory.cleanAllHttpDatas();
return;
factory.cleanAllHttpDatas();
return;
}
// Multipart Post form: factory used
formpostmultipart(bootstrap, host, port, uriFile, file, factory, headers, bodylist);
@ -155,7 +155,7 @@ public class HttpClient {
* @return the list of headers that will be used in every example after
**/
private static List<Entry<String,String>> formget(ClientBootstrap bootstrap, String host, int port, String get,
URI uriSimple) {
URI uriSimple) {
// XXX /formget
// No use of HttpPostRequestEncoder since not a POST
// Start the connection attempt.
@ -193,14 +193,14 @@ public class HttpClient {
request.setHeader(HttpHeaders.Names.HOST, host);
request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP+","+
HttpHeaders.Values.DEFLATE);
HttpHeaders.Values.DEFLATE);
request.setHeader(HttpHeaders.Names.ACCEPT_CHARSET, "ISO-8859-1,utf-8;q=0.7,*;q=0.7");
request.setHeader(HttpHeaders.Names.ACCEPT_LANGUAGE, "fr");
request.setHeader(HttpHeaders.Names.REFERER, uriSimple.toString());
request.setHeader(HttpHeaders.Names.USER_AGENT, "Netty Simple Http Client side");
request.setHeader(HttpHeaders.Names.ACCEPT,
"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8");
"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8");
//connection will not close but needed
// request.setHeader("Connection","keep-alive");
// request.setHeader("Keep-Alive","300");
@ -232,12 +232,12 @@ public class HttpClient {
* @return the list of HttpData object (attribute and file) to be reused on next post
*/
private static List<InterfaceHttpData> formpost(ClientBootstrap bootstrap,
String host, int port,
URI uriSimple, File file, HttpDataFactory factory,
List<Entry<String,String>> headers) {
String host, int port,
URI uriSimple, File file, HttpDataFactory factory,
List<Entry<String,String>> headers) {
// XXX /formpost
// Start the connection attempt.
ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
// Wait until the connection attempt succeeds or fails.
Channel channel = future.awaitUninterruptibly().getChannel();
if (!future.isSuccess()) {
@ -266,7 +266,7 @@ public class HttpClient {
// it is legal to add directly header or cookie into the request until finalize
for (Entry<String, String> entry : headers) {
request.setHeader(entry.getKey(), entry.getValue());
}
}
// add Form attribute
try {
@ -327,13 +327,13 @@ public class HttpClient {
* @param bodylist
*/
private static void formpostmultipart(ClientBootstrap bootstrap, String host, int port,
URI uriFile, File file, HttpDataFactory factory,
List<Entry<String,String>> headers, List<InterfaceHttpData> bodylist) {
URI uriFile, File file, HttpDataFactory factory,
List<Entry<String,String>> headers, List<InterfaceHttpData> bodylist) {
// XXX /formpostmultipart
// Start the connection attempt.
ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
// Wait until the connection attempt succeeds or fails.
Channel channel = future.awaitUninterruptibly().getChannel();
Channel channel = future.awaitUninterruptibly().getChannel();
if (!future.isSuccess()) {
future.getCause().printStackTrace();
bootstrap.releaseExternalResources();

View File

@ -36,7 +36,7 @@ import org.jboss.netty.handler.stream.ChunkedWriteHandler;
* @version $Rev$, $Date$
*/
public class HttpClientPipelineFactory implements ChannelPipelineFactory {
private final boolean ssl;
private final boolean ssl;
public HttpClientPipelineFactory(boolean ssl) {
this.ssl = ssl;

View File

@ -36,7 +36,7 @@ public class HttpResponseHandler extends SimpleChannelUpstreamHandler {
private volatile boolean readingChunks;
@Override
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
if (!readingChunks) {
HttpResponse response = (HttpResponse) e.getMessage();

View File

@ -34,8 +34,8 @@ public class HttpServer {
// Configure the server.
ServerBootstrap bootstrap = new ServerBootstrap(
new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
// Set up the event pipeline factory.
bootstrap.setPipelineFactory(new HttpServerPipelineFactory());

View File

@ -51,7 +51,7 @@ public class SecureChatClientPipelineFactory implements
SSLEngine engine =
SecureChatSslContextFactory.getClientContext().createSSLEngine();
engine.setUseClientMode(true);
//engine.setUseClientMode(true);
pipeline.addLast("ssl", new SslHandler(engine));

View File

@ -19,6 +19,7 @@ import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
@ -70,11 +71,16 @@ public class UptimeClient {
// Configure the pipeline factory.
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
private final ChannelHandler timeoutHandler =
new ReadTimeoutHandler(timer, READ_TIMEOUT);
private final ChannelHandler uptimeHandler =
new UptimeClientHandler(bootstrap, timer);
@Override
public ChannelPipeline getPipeline() throws Exception {
return Channels.pipeline(
new ReadTimeoutHandler(timer, READ_TIMEOUT),
new UptimeClientHandler(bootstrap, timer));
timeoutHandler, uptimeHandler);
}
});

View File

@ -24,6 +24,7 @@ import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.handler.timeout.ReadTimeoutException;
import org.jboss.netty.util.Timeout;
import org.jboss.netty.util.Timer;
import org.jboss.netty.util.TimerTask;
@ -85,6 +86,13 @@ public class UptimeClientHandler extends SimpleChannelUpstreamHandler {
startTime = -1;
println("Failed to connect: " + cause.getMessage());
}
if (cause instanceof ReadTimeoutException) {
// The connection was OK but there was no traffic for last period.
println("Disconnecting due to no inbound traffic");
}
else {
cause.printStackTrace();
}
ctx.getChannel().close();
}

View File

@ -187,6 +187,11 @@ abstract class AbstractCodecEmbedder<E> implements CodecEmbedder<E> {
return productQueue.size();
}
@Override
public ChannelPipeline getPipeline() {
return pipeline;
}
private final class EmbeddedChannelSink implements ChannelSink, ChannelUpstreamHandler {
EmbeddedChannelSink() {
super();

View File

@ -17,6 +17,8 @@ package org.jboss.netty.handler.codec.embedder;
import java.util.Collection;
import org.jboss.netty.channel.ChannelPipeline;
/**
* A helper that wraps an encoder or a decoder (codec) so that they can be used
* without doing actual I/O in unit tests or higher level codecs. Please refer
@ -93,4 +95,9 @@ public interface CodecEmbedder<E> {
* Returns the number of encoded or decoded output in the product queue.
*/
int size();
/**
* Returns the {@link ChannelPipeline} that handles the input.
*/
ChannelPipeline getPipeline();
}

View File

@ -290,17 +290,7 @@ public class LengthFieldBasedFrameDecoder extends FrameDecoder {
buffer.skipBytes(localBytesToDiscard);
bytesToDiscard -= localBytesToDiscard;
this.bytesToDiscard = bytesToDiscard;
if (bytesToDiscard == 0) {
// Reset to the initial state and tell the handlers that
// the frame was too large.
// TODO Let user choose when the exception should be raised - early or late?
// If early, fail() should be called when discardingTooLongFrame is set to true.
long tooLongFrameLength = this.tooLongFrameLength;
this.tooLongFrameLength = 0;
fail(ctx, tooLongFrameLength);
} else {
// Keep discarding.
}
failIfNecessary(ctx);
return null;
}
@ -350,6 +340,7 @@ public class LengthFieldBasedFrameDecoder extends FrameDecoder {
tooLongFrameLength = frameLength;
bytesToDiscard = frameLength - buffer.readableBytes();
buffer.skipBytes(buffer.readableBytes());
failIfNecessary(ctx);
return null;
}
@ -375,6 +366,21 @@ public class LengthFieldBasedFrameDecoder extends FrameDecoder {
return frame;
}
private void failIfNecessary(ChannelHandlerContext ctx) {
if (bytesToDiscard == 0) {
// Reset to the initial state and tell the handlers that
// the frame was too large.
// TODO Let user choose when the exception should be raised - early or late?
// If early, fail() should be called when discardingTooLongFrame is set to true.
long tooLongFrameLength = this.tooLongFrameLength;
this.tooLongFrameLength = 0;
discardingTooLongFrame = false;
fail(ctx, tooLongFrameLength);
} else {
// Keep discarding.
}
}
/**
* Extract the sub-region of the specified buffer. This method is called by
* {@link #decode(ChannelHandlerContext, Channel, ChannelBuffer)} for each

View File

@ -1,35 +0,0 @@
/*
* 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.handler.codec.http;
import java.text.SimpleDateFormat;
import java.util.Locale;
import java.util.TimeZone;
/**
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev$, $Date$
*/
final class CookieDateFormat extends SimpleDateFormat {
private static final long serialVersionUID = 1789486337887402640L;
CookieDateFormat() {
super("E, d-MMM-y HH:mm:ss z", Locale.ENGLISH);
setTimeZone(TimeZone.getTimeZone("GMT"));
}
}

View File

@ -134,7 +134,7 @@ public class CookieDecoder {
} else if (CookieHeaderNames.EXPIRES.equalsIgnoreCase(name)) {
try {
long maxAgeMillis =
new CookieDateFormat().parse(value).getTime() -
new HttpHeaderDateFormat().parse(value).getTime() -
System.currentTimeMillis();
if (maxAgeMillis <= 0) {
maxAge = 0;

View File

@ -107,7 +107,7 @@ public class CookieEncoder {
if (cookie.getMaxAge() >= 0) {
if (cookie.getVersion() == 0) {
addUnquoted(sb, CookieHeaderNames.EXPIRES,
new CookieDateFormat().format(
new HttpHeaderDateFormat().format(
new Date(System.currentTimeMillis() +
cookie.getMaxAge() * 1000L)));
} else {
@ -167,7 +167,10 @@ public class CookieEncoder {
}
}
sb.setLength(sb.length() - 1);
if (sb.length() > 0) {
sb.setLength(sb.length() - 1);
}
return sb.toString();
}
@ -205,7 +208,8 @@ public class CookieEncoder {
}
}
sb.setLength(sb.length() - 1);
if(sb.length() > 0)
sb.setLength(sb.length() - 1);
return sb.toString();
}

View File

@ -51,7 +51,7 @@ public class DefaultHttpDataFactory implements HttpDataFactory {
* Keep all HttpDatas until cleanAllHttpDatas() is called.
*/
private ConcurrentHashMap<HttpRequest, List<HttpData>> requestFileDeleteMap =
new ConcurrentHashMap<HttpRequest, List<HttpData>>();
new ConcurrentHashMap<HttpRequest, List<HttpData>>();
/**
* HttpData will be in memory if less than default size (16KB).
* The type will be Mixed.
@ -88,10 +88,10 @@ public class DefaultHttpDataFactory implements HttpDataFactory {
* @return the associated list of Files for the request
*/
private List<HttpData> getList(HttpRequest request) {
List<HttpData> list = requestFileDeleteMap.get(request);
List<HttpData> list = requestFileDeleteMap.get(request);
if (list == null) {
list = new ArrayList<HttpData>();
requestFileDeleteMap.put(request, list);
list = new ArrayList<HttpData>();
requestFileDeleteMap.put(request, list);
}
return list;
}
@ -170,33 +170,33 @@ public class DefaultHttpDataFactory implements HttpDataFactory {
@Override
public void removeHttpDataFromClean(HttpRequest request, InterfaceHttpData data) {
if (data instanceof HttpData) {
List<HttpData> fileToDelete = getList(request);
List<HttpData> fileToDelete = getList(request);
fileToDelete.remove(data);
}
}
@Override
public void cleanRequestHttpDatas(HttpRequest request) {
List<HttpData> fileToDelete = requestFileDeleteMap.remove(request);
if (fileToDelete != null) {
for (HttpData data: fileToDelete) {
data.delete();
}
fileToDelete.clear();
}
List<HttpData> fileToDelete = requestFileDeleteMap.remove(request);
if (fileToDelete != null) {
for (HttpData data: fileToDelete) {
data.delete();
}
fileToDelete.clear();
}
}
@Override
public void cleanAllHttpDatas() {
for (HttpRequest request : requestFileDeleteMap.keySet()) {
List<HttpData> fileToDelete = requestFileDeleteMap.get(request);
if (fileToDelete != null) {
for (HttpData data: fileToDelete) {
data.delete();
}
fileToDelete.clear();
}
requestFileDeleteMap.remove(request);
}
for (HttpRequest request : requestFileDeleteMap.keySet()) {
List<HttpData> fileToDelete = requestFileDeleteMap.get(request);
if (fileToDelete != null) {
for (HttpData data: fileToDelete) {
data.delete();
}
fileToDelete.clear();
}
requestFileDeleteMap.remove(request);
}
}
}

View File

@ -15,6 +15,7 @@
*/
package org.jboss.netty.handler.codec.http;
import java.io.File;
import java.nio.charset.Charset;
/**
@ -156,7 +157,8 @@ public class DiskFileUpload extends AbstractDiskHttpData implements FileUpload {
@Override
protected String getDiskFilename() {
return filename;
File file = new File(filename);
return file.getName();
}
@Override

View File

@ -136,11 +136,20 @@ public class HttpClientCodec implements ChannelUpstreamHandler,
// message-body, even though the presence of entity-header fields
// might lead one to believe they do.
if (HttpMethod.HEAD.equals(method)) {
// Interesting edge case:
// Zero-byte chunk will appear if Transfer-Encoding of the
// response is 'chunked'. This is probably because of the
// trailing headers.
return !msg.isChunked();
return true;
// The following code was inserted to work around the servers
// that behave incorrectly. It has been commented out
// because it does not work with well behaving servers.
// Please note, even if the 'Transfer-Encoding: chunked'
// header exists in the HEAD response, the response should
// have absolutely no content.
//
//// Interesting edge case:
//// Some poorly implemented servers will send a zero-byte
//// chunk if Transfer-Encoding of the response is 'chunked'.
////
//// return !msg.isChunked();
}
break;
case 'C':

View File

@ -62,7 +62,7 @@ public abstract class HttpContentDecoder extends SimpleChannelUpstreamHandler {
Object msg = e.getMessage();
if (msg instanceof HttpResponse && ((HttpResponse) msg).getStatus().getCode() == 100) {
// 100-continue response must be passed through.
ctx.sendDownstream(e);
ctx.sendUpstream(e);
} else if (msg instanceof HttpMessage) {
HttpMessage m = (HttpMessage) msg;

View File

@ -102,7 +102,8 @@ public abstract class HttpContentEncoder extends SimpleChannelHandler {
throw new IllegalStateException("cannot send more responses than requests");
}
if ((encoder = newContentEncoder(acceptEncoding)) != null) {
boolean hasContent = m.isChunked() || m.getContent().readable();
if (hasContent && (encoder = newContentEncoder(acceptEncoding)) != null) {
// Encode the content and remove or replace the existing headers
// so that the message looks like a decoded message.
m.setHeader(

View File

@ -0,0 +1,92 @@
/*
* 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.handler.codec.http;
import java.text.ParsePosition;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;
import java.util.TimeZone;
/**
* This DateFormat decodes 3 formats of {@link Date}, but only encodes the one,
* the first:
* <ul>
* <li>Sun, 06 Nov 1994 08:49:37 GMT: standard specification, the only one with
* valid generation</li>
* <li>Sun, 06 Nov 1994 08:49:37 GMT: obsolete specification</li>
* <li>Sun Nov 6 08:49:37 1994: obsolete specification</li>
* </ul>
*
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @author <a href="http://www.rogiel.com/">Rogiel Josias Sulzbach</a>
* @version $Rev$, $Date$
*/
final class HttpHeaderDateFormat extends SimpleDateFormat {
private static final long serialVersionUID = -925286159755905325L;
private final SimpleDateFormat format1 = new HttpHeaderDateFormatObsolete1();
private final SimpleDateFormat format2 = new HttpHeaderDateFormatObsolete2();
/**
* Standard date format<p>
* Sun, 06 Nov 1994 08:49:37 GMT -> E, d MMM yyyy HH:mm:ss z
*/
HttpHeaderDateFormat() {
super("E, dd MMM yyyy HH:mm:ss z", Locale.ENGLISH);
setTimeZone(TimeZone.getTimeZone("GMT"));
}
@Override
public Date parse(String text, ParsePosition pos) {
Date date = super.parse(text, pos);
if (date == null) {
date = format1.parse(text, pos);
}
if (date == null) {
date = format2.parse(text, pos);
}
return date;
}
/**
* First obsolete format<p>
* Sunday, 06-Nov-94 08:49:37 GMT -> E, d-MMM-y HH:mm:ss z
*/
private final class HttpHeaderDateFormatObsolete1 extends SimpleDateFormat {
private static final long serialVersionUID = -3178072504225114298L;
HttpHeaderDateFormatObsolete1() {
super("E, dd-MMM-y HH:mm:ss z", Locale.ENGLISH);
setTimeZone(TimeZone.getTimeZone("GMT"));
}
}
/**
* Second obsolete format
* <p>
* Sun Nov 6 08:49:37 1994 -> EEE, MMM d HH:mm:ss yyyy
*/
private final class HttpHeaderDateFormatObsolete2 extends SimpleDateFormat {
private static final long serialVersionUID = 3010674519968303714L;
HttpHeaderDateFormatObsolete2() {
super("E MMM d HH:mm:ss yyyy", Locale.ENGLISH);
setTimeZone(TimeZone.getTimeZone("GMT"));
}
}
}

View File

@ -15,6 +15,9 @@
*/
package org.jboss.netty.handler.codec.http;
import java.text.ParseException;
import java.util.Calendar;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@ -529,6 +532,10 @@ public class HttpHeaders {
/**
* Sets a new header with the specified name and value. If there is an
* existing header with the same name, the existing header is removed.
* If the specified value is not a {@link String}, it is converted into a
* {@link String} by {@link Object#toString()}, except for {@link Date}
* and {@link Calendar} which are formatted to the date format defined in
* <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>.
*/
public static void setHeader(HttpMessage message, String name, Object value) {
message.setHeader(name, value);
@ -537,6 +544,16 @@ public class HttpHeaders {
/**
* Sets a new header with the specified name and values. If there is an
* existing header with the same name, the existing header is removed.
* This method can be represented approximately as the following code:
* <pre>
* removeHeader(message, name);
* for (Object v: values) {
* if (v == null) {
* break;
* }
* addHeader(message, name, v);
* }
* </pre>
*/
public static void setHeader(HttpMessage message, String name, Iterable<?> values) {
message.setHeader(name, values);
@ -544,11 +561,29 @@ public class HttpHeaders {
/**
* Adds a new header with the specified name and value.
* If the specified value is not a {@link String}, it is converted into a
* {@link String} by {@link Object#toString()}, except for {@link Date}
* and {@link Calendar} which are formatted to the date format defined in
* <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>.
*/
public static void addHeader(HttpMessage message, String name, Object value) {
message.addHeader(name, value);
}
/**
* Removes the header with the specified name.
*/
public static void removeHeader(HttpMessage message, String name) {
message.removeHeader(name);
}
/**
* Removes all headers from the specified message.
*/
public static void clearHeaders(HttpMessage message) {
message.clearHeaders();
}
/**
* Returns the integer header value with the specified header name. If
* there are more than one header value for the specified header name, the
@ -561,7 +596,7 @@ public class HttpHeaders {
public static int getIntHeader(HttpMessage message, String name) {
String value = getHeader(message, name);
if (value == null) {
throw new NumberFormatException("null");
throw new NumberFormatException("header not found: " + name);
}
return Integer.parseInt(value);
}
@ -610,17 +645,104 @@ public class HttpHeaders {
message.addHeader(name, value);
}
/**
* Returns the date header value with the specified header name. If
* there are more than one header value for the specified header name, the
* first value is returned.
*
* @return the header value
* @throws ParseException
* if there is no such header or the header value is not a formatted date
*/
public static Date getDateHeader(HttpMessage message, String name) throws ParseException {
String value = getHeader(message, name);
if (value == null) {
throw new ParseException("header not found: " + name, 0);
}
return new HttpHeaderDateFormat().parse(value);
}
/**
* Returns the date header value with the specified header name. If
* there are more than one header value for the specified header name, the
* first value is returned.
*
* @return the header value or the {@code defaultValue} if there is no such
* header or the header value is not a formatted date
*/
public static Date getDateHeader(HttpMessage message, String name, Date defaultValue) {
final String value = getHeader(message, name);
if (value == null) {
return defaultValue;
}
try {
return new HttpHeaderDateFormat().parse(value);
} catch (ParseException e) {
return defaultValue;
}
}
/**
* Sets a new date header with the specified name and value. If there
* is an existing header with the same name, the existing header is removed.
* The specified value is formatted as defined in
* <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>
*/
public static void setDateHeader(HttpMessage message, String name, Date value) {
if (value != null) {
message.setHeader(name, new HttpHeaderDateFormat().format(value));
} else {
message.setHeader(name, null);
}
}
/**
* Sets a new date header with the specified name and values. If there
* is an existing header with the same name, the existing header is removed.
* The specified values are formatted as defined in
* <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>
*/
public static void setDateHeader(HttpMessage message, String name, Iterable<Date> values) {
message.setHeader(name, values);
}
/**
* Adds a new date header with the specified name and value. The specified
* value is formatted as defined in <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>
*/
public static void addDateHeader(HttpMessage message, String name, Date value) {
message.addHeader(name, value);
}
/**
* Returns the length of the content. Please note that this value is
* not retrieved from {@link HttpMessage#getContent()} but from the
* {@code "Content-Length"} header, and thus they are independent from each
* other.
*
* @return the content length or {@code 0} if this message does not have
* the {@code "Content-Length"} header
* @return the content length
*
* @throws NumberFormatException
* if the message does not have the {@code "Content-Length"} header
* or its value is not a number
*/
public static long getContentLength(HttpMessage message) {
return getContentLength(message, 0L);
String value = getHeader(message, Names.CONTENT_LENGTH);
if (value != null) {
return Long.parseLong(value);
}
// We know the content length if it's a Web Socket message even if
// Content-Length header is missing.
long webSocketContentLength = getWebSocketContentLength(message);
if (webSocketContentLength >= 0) {
return webSocketContentLength;
}
// Otherwise we don't.
throw new NumberFormatException("header not found: " + Names.CONTENT_LENGTH);
}
/**
@ -630,14 +752,35 @@ public class HttpHeaders {
* other.
*
* @return the content length or {@code defaultValue} if this message does
* not have the {@code "Content-Length"} header
* not have the {@code "Content-Length"} header or its value is not
* a number
*/
public static long getContentLength(HttpMessage message, long defaultValue) {
String contentLength = message.getHeader(Names.CONTENT_LENGTH);
if (contentLength != null) {
return Long.parseLong(contentLength);
try {
return Long.parseLong(contentLength);
} catch (NumberFormatException e) {
return defaultValue;
}
}
// We know the content length if it's a Web Socket message even if
// Content-Length header is missing.
long webSocketContentLength = getWebSocketContentLength(message);
if (webSocketContentLength >= 0) {
return webSocketContentLength;
}
// Otherwise we don't.
return defaultValue;
}
/**
* Returns the content length of the specified web socket message. If the
* specified message is not a web socket message, {@code -1} is returned.
*/
private static int getWebSocketContentLength(HttpMessage message) {
// WebSockset messages have constant content-lengths.
if (message instanceof HttpRequest) {
HttpRequest req = (HttpRequest) message;
@ -655,7 +798,8 @@ public class HttpHeaders {
}
}
return defaultValue;
// Not a web socket message
return -1;
}
/**
@ -687,6 +831,37 @@ public class HttpHeaders {
message.setHeader(Names.HOST, value);
}
/**
* Returns the value of the {@code "Date"} header.
*
* @throws ParseException
* if there is no such header or the header value is not a formatted date
*/
public static Date getDate(HttpMessage message) throws ParseException {
return getDateHeader(message, Names.DATE);
}
/**
* Returns the value of the {@code "Date"} header. If there is no such
* header or the header is not a formatted date, the {@code defaultValue}
* is returned.
*/
public static Date getDate(HttpMessage message, Date defaultValue) {
return getDateHeader(message, Names.DATE, defaultValue);
}
/**
* Sets the {@code "Date"} header.
*/
public static void setDate(HttpMessage message, Date value) {
if (value != null) {
message.setHeader(Names.DATE, new HttpHeaderDateFormat().format(value));
} else {
message.setHeader(Names.DATE, null);
}
}
/**
* Returns {@code true} if and only if the specified message contains the
* {@code "Expect: 100-continue"} header.
@ -976,6 +1151,18 @@ public class HttpHeaders {
if (value == null) {
return null;
}
if (value instanceof String) {
return (String) value;
}
if (value instanceof Number) {
return value.toString();
}
if (value instanceof Date) {
return new HttpHeaderDateFormat().format((Date) value);
}
if (value instanceof Calendar) {
return new HttpHeaderDateFormat().format(((Calendar) value).getTime());
}
return value.toString();
}

View File

@ -15,6 +15,8 @@
*/
package org.jboss.netty.handler.codec.http;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -100,18 +102,36 @@ public interface HttpMessage {
/**
* Adds a new header with the specified name and value.
* If the specified value is not a {@link String}, it is converted into a
* {@link String} by {@link Object#toString()}, except for {@link Date}
* and {@link Calendar} which are formatted to the date format defined in
* <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>.
*/
void addHeader(String name, Object value);
/**
* Sets a new header with the specified name and value. If there is an
* existing header with the same name, the existing header is removed.
* If the specified value is not a {@link String}, it is converted into a
* {@link String} by {@link Object#toString()}, except for {@link Date}
* and {@link Calendar} which are formatted to the date format defined in
* <a href="http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1">RFC2616</a>.
*/
void setHeader(String name, Object value);
/**
* Sets a new header with the specified name and values. If there is an
* existing header with the same name, the existing header is removed.
* This method can be represented approximately as the following code:
* <pre>
* m.removeHeader(name);
* for (Object v: values) {
* if (v == null) {
* break;
* }
* m.addHeader(name, v);
* }
* </pre>
*/
void setHeader(String name, Iterable<?> values);

View File

@ -634,7 +634,7 @@ public class HttpPostRequestDecoder {
}
case FIELD: {
// Now get value according to Content-Type and Charset
Charset localCharset = null;
Charset localCharset = null;
Attribute charsetAttribute = currentFieldAttributes
.get(HttpHeaders.Values.CHARSET);
if (charsetAttribute != null) {
@ -854,7 +854,7 @@ public class HttpPostRequestDecoder {
Attribute attribute;
try {
attribute = factory.createAttribute(request,
contents[0].trim(),
contents[0].trim(),
decodeAttribute(cleanString(contents[i]), charset));
} catch (NullPointerException e) {
throw new ErrorDataDecoderException(e);

View File

@ -73,8 +73,10 @@ public class MixedAttribute implements Attribute {
if (attribute.length() + buffer.readableBytes() > limitSize) {
DiskAttribute diskAttribute = new DiskAttribute(attribute
.getName());
diskAttribute.addContent(((MemoryAttribute) attribute)
.getChannelBuffer(), false);
if (((MemoryAttribute) attribute).getChannelBuffer() != null) {
diskAttribute.addContent(((MemoryAttribute) attribute)
.getChannelBuffer(), last);
}
attribute = diskAttribute;
}
}

View File

@ -62,8 +62,10 @@ public class MixedFileUpload implements FileUpload {
.getContentType(), fileUpload
.getContentTransferEncoding(), fileUpload.getCharset(),
definedSize);
diskFileUpload.addContent(((MemoryFileUpload) fileUpload)
.getChannelBuffer(), false);
if (((MemoryFileUpload) fileUpload).getChannelBuffer() != null){
diskFileUpload.addContent(((MemoryFileUpload) fileUpload)
.getChannelBuffer(), last);
}
fileUpload = diskFileUpload;
}
}

View File

@ -78,9 +78,12 @@ public class ProtobufEncoder extends OneToOneEncoder {
@Override
protected Object encode(
ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
if (!(msg instanceof MessageLite)) {
return msg;
if (msg instanceof MessageLite) {
return wrappedBuffer(((MessageLite) msg).toByteArray());
}
return wrappedBuffer(((MessageLite) msg).toByteArray());
if (msg instanceof MessageLite.Builder) {
return wrappedBuffer(((MessageLite.Builder) msg).build().toByteArray());
}
return msg;
}
}

View File

@ -21,6 +21,8 @@ import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectStreamClass;
import java.io.StreamCorruptedException;
import java.util.HashMap;
import java.util.Map;
/**
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
@ -31,6 +33,7 @@ import java.io.StreamCorruptedException;
*/
class CompactObjectInputStream extends ObjectInputStream {
private final Map<String, Class<?>> classCache = new HashMap<String, Class<?>>();
private final ClassLoader classLoader;
CompactObjectInputStream(InputStream in) throws IOException {
@ -65,7 +68,7 @@ class CompactObjectInputStream extends ObjectInputStream {
case CompactObjectOutputStream.TYPE_THIN_DESCRIPTOR:
String className = readUTF();
Class<?> clazz = loadClass(className);
return ObjectStreamClass.lookup(clazz);
return ObjectStreamClass.lookupAny(clazz);
default:
throw new StreamCorruptedException(
"Unexpected class descriptor type: " + type);
@ -74,16 +77,33 @@ class CompactObjectInputStream extends ObjectInputStream {
@Override
protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {
// Query the cache first.
String className = desc.getName();
try {
return loadClass(className);
} catch (ClassNotFoundException ex) {
return super.resolveClass(desc);
Class<?> clazz = classCache.get(className);
if (clazz != null) {
return clazz;
}
// And then try to resolve.
try {
clazz = loadClass(className);
} catch (ClassNotFoundException ex) {
clazz = super.resolveClass(desc);
}
classCache.put(className, clazz);
return clazz;
}
protected Class<?> loadClass(String className) throws ClassNotFoundException {
// Query the cache first.
Class<?> clazz;
clazz = classCache.get(className);
if (clazz != null) {
return clazz;
}
// And then try to load.
ClassLoader classLoader = this.classLoader;
if (classLoader == null) {
classLoader = Thread.currentThread().getContextClassLoader();
@ -94,6 +114,8 @@ class CompactObjectInputStream extends ObjectInputStream {
} else {
clazz = Class.forName(className);
}
classCache.put(className, clazz);
return clazz;
}
}

View File

@ -167,6 +167,17 @@ public abstract class CIDR implements Comparable<CIDR>
*/
public abstract boolean contains(InetAddress inetAddress);
/* (non-Javadoc)
* @see java.lang.Object#equals(java.lang.Object)
*/
@Override
public boolean equals(Object arg0) {
if (!(arg0 instanceof CIDR)) {
return false;
}
return (this.compareTo((CIDR) arg0) == 0);
}
/** Convert an IPv4 or IPv6 textual representation into an
* InetAddress.
* @param addr

View File

@ -329,6 +329,8 @@ public class SslHandler extends FrameDecoder
ChannelHandlerContext ctx = this.ctx;
Channel channel = ctx.getChannel();
ChannelFuture handshakeFuture;
Exception exception = null;
synchronized (handshakeLock) {
if (handshaking) {
return this.handshakeFuture;
@ -338,17 +340,24 @@ public class SslHandler extends FrameDecoder
engine.beginHandshake();
runDelegatedTasks();
handshakeFuture = this.handshakeFuture = future(channel);
} catch (SSLException e) {
} catch (Exception e) {
handshakeFuture = this.handshakeFuture = failedFuture(channel, e);
exception = e;
}
}
}
try {
wrapNonAppData(ctx, channel);
} catch (SSLException e) {
handshakeFuture.setFailure(e);
if (exception == null) { // Began handshake successfully.
try {
wrapNonAppData(ctx, channel);
} catch (SSLException e) {
fireExceptionCaught(ctx, e);
handshakeFuture.setFailure(e);
}
} else { // Failed to initiate handshake.
fireExceptionCaught(ctx, exception);
}
return handshakeFuture;
}
@ -363,6 +372,7 @@ public class SslHandler extends FrameDecoder
engine.closeOutbound();
return wrapNonAppData(ctx, channel);
} catch (SSLException e) {
fireExceptionCaught(ctx, e);
return failedFuture(channel, e);
}
}
@ -659,6 +669,11 @@ public class SslHandler extends FrameDecoder
channel, future, msg, channel.getRemoteAddress());
offerEncryptedWriteRequest(encryptedWrite);
offered = true;
} else if (result.getStatus() == Status.CLOSED) {
// SSLEngine has been closed already.
// Any further write attempts should be denied.
success = false;
break;
} else {
final HandshakeStatus handshakeStatus = result.getHandshakeStatus();
handleRenegotiation(handshakeStatus);
@ -1072,6 +1087,8 @@ public class SslHandler extends FrameDecoder
public void operationComplete(ChannelFuture closeNotifyFuture) throws Exception {
if (!(closeNotifyFuture.getCause() instanceof ClosedChannelException)) {
Channels.close(context, e.getFuture());
} else {
e.getFuture().setSuccess();
}
}
}

View File

@ -23,6 +23,12 @@ import java.io.PushbackInputStream;
/**
* A {@link ChunkedInput} that fetches data from an {@link InputStream} chunk by
* chunk.
* <p>
* Please note that the {@link InputStream} instance that feeds data into
* {@link ChunkedStream} must implement {@link InputStream#available()} as
* accurately as possible, rather than using the default implementation.
* Otherwise, {@link ChunkedStream} will generate many too small chunks or
* block unnecessarily often.
*
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>

View File

@ -17,6 +17,7 @@ package org.jboss.netty.handler.stream;
import static org.jboss.netty.channel.Channels.*;
import java.nio.channels.ClosedChannelException;
import java.util.Queue;
import org.jboss.netty.buffer.ChannelBuffers;
@ -150,32 +151,40 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
ctx.sendUpstream(e);
}
private synchronized void discard(ChannelHandlerContext ctx) {
for (;;) {
if (currentEvent == null) {
currentEvent = queue.poll();
}
private void discard(ChannelHandlerContext ctx) {
ClosedChannelException cause = null;
boolean fireExceptionCaught = false;
synchronized (this) {
for (;;) {
if (currentEvent == null) {
currentEvent = queue.poll();
}
if (currentEvent == null) {
break;
}
if (currentEvent == null) {
break;
}
MessageEvent currentEvent = this.currentEvent;
this.currentEvent = null;
MessageEvent currentEvent = this.currentEvent;
this.currentEvent = null;
Object m = currentEvent.getMessage();
if (m instanceof ChunkedInput) {
closeInput((ChunkedInput) m);
Object m = currentEvent.getMessage();
if (m instanceof ChunkedInput) {
closeInput((ChunkedInput) m);
}
// Trigger a ClosedChannelException
Channels.write(
ctx, currentEvent.getFuture(), ChannelBuffers.EMPTY_BUFFER,
currentEvent.getRemoteAddress());
} else {
// Trigger a ClosedChannelException
ctx.sendDownstream(currentEvent);
if (cause == null) {
cause = new ClosedChannelException();
}
currentEvent.getFuture().setFailure(cause);
fireExceptionCaught = true;
currentEvent = null;
}
currentEvent = null;
}
if (fireExceptionCaught) {
Channels.fireExceptionCaught(ctx.getChannel(), cause);
}
}

View File

@ -21,6 +21,8 @@ import java.util.concurrent.TimeUnit;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandler.Sharable;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
@ -73,14 +75,16 @@ import org.jboss.netty.util.TimerTask;
* public class MyPipelineFactory implements {@link ChannelPipelineFactory} {
*
* private final {@link Timer} timer;
* private final {@link ChannelHandler} idleStateHandler;
*
* public MyPipelineFactory({@link Timer} timer) {
* this.timer = timer;
* this.idleStateHandler = <b>new {@link IdleStateHandler}(timer, 60, 30, 0), // timer must be shared.</b>
* }
*
* public {@link ChannelPipeline} getPipeline() {
* return {@link Channels}.pipeline(
* <b>new {@link IdleStateHandler}(timer, 60, 30, 0), // timer must be shared.</b>
* idleStateHandler,
* new MyHandler());
* }
* }
@ -120,6 +124,7 @@ import org.jboss.netty.util.TimerTask;
* @apiviz.uses org.jboss.netty.util.HashedWheelTimer
* @apiviz.has org.jboss.netty.handler.timeout.IdleStateEvent oneway - - triggers
*/
@Sharable
public class IdleStateHandler extends SimpleChannelUpstreamHandler
implements LifeCycleAwareChannelHandler,
ExternalResourceReleasable {
@ -127,15 +132,8 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
final Timer timer;
final long readerIdleTimeMillis;
volatile Timeout readerIdleTimeout;
volatile long lastReadTime;
final long writerIdleTimeMillis;
volatile Timeout writerIdleTimeout;
volatile long lastWriteTime;
final long allIdleTimeMillis;
volatile Timeout allIdleTimeout;
/**
* Creates a new instance.
@ -249,7 +247,7 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
@Override
public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
destroy();
destroy(ctx);
}
@Override
@ -270,14 +268,15 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
@Override
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
destroy();
destroy(ctx);
ctx.sendUpstream(e);
}
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
lastReadTime = System.currentTimeMillis();
State state = (State) ctx.getAttachment();
state.lastReadTime = System.currentTimeMillis();
ctx.sendUpstream(e);
}
@ -285,42 +284,47 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
public void writeComplete(ChannelHandlerContext ctx, WriteCompletionEvent e)
throws Exception {
if (e.getWrittenAmount() > 0) {
lastWriteTime = System.currentTimeMillis();
State state = (State) ctx.getAttachment();
state.lastWriteTime = System.currentTimeMillis();
}
ctx.sendUpstream(e);
}
private void initialize(ChannelHandlerContext ctx) {
lastReadTime = lastWriteTime = System.currentTimeMillis();
State state = new State();
ctx.setAttachment(state);
state.lastReadTime = state.lastWriteTime = System.currentTimeMillis();
if (readerIdleTimeMillis > 0) {
readerIdleTimeout = timer.newTimeout(
state.readerIdleTimeout = timer.newTimeout(
new ReaderIdleTimeoutTask(ctx),
readerIdleTimeMillis, TimeUnit.MILLISECONDS);
}
if (writerIdleTimeMillis > 0) {
writerIdleTimeout = timer.newTimeout(
state.writerIdleTimeout = timer.newTimeout(
new WriterIdleTimeoutTask(ctx),
writerIdleTimeMillis, TimeUnit.MILLISECONDS);
}
if (allIdleTimeMillis > 0) {
allIdleTimeout = timer.newTimeout(
state.allIdleTimeout = timer.newTimeout(
new AllIdleTimeoutTask(ctx),
allIdleTimeMillis, TimeUnit.MILLISECONDS);
}
}
private void destroy() {
if (readerIdleTimeout != null) {
readerIdleTimeout.cancel();
readerIdleTimeout = null;
private void destroy(ChannelHandlerContext ctx) {
State state = (State) ctx.getAttachment();
if (state.readerIdleTimeout != null) {
state.readerIdleTimeout.cancel();
state.readerIdleTimeout = null;
}
if (writerIdleTimeout != null) {
writerIdleTimeout.cancel();
writerIdleTimeout = null;
if (state.writerIdleTimeout != null) {
state.writerIdleTimeout.cancel();
state.writerIdleTimeout = null;
}
if (allIdleTimeout != null) {
allIdleTimeout.cancel();
allIdleTimeout = null;
if (state.allIdleTimeout != null) {
state.allIdleTimeout.cancel();
state.allIdleTimeout = null;
}
}
@ -343,12 +347,13 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
return;
}
State state = (State) ctx.getAttachment();
long currentTime = System.currentTimeMillis();
long lastReadTime = IdleStateHandler.this.lastReadTime;
long lastReadTime = state.lastReadTime;
long nextDelay = readerIdleTimeMillis - (currentTime - lastReadTime);
if (nextDelay <= 0) {
// Reader is idle - set a new timeout and notify the callback.
readerIdleTimeout =
state.readerIdleTimeout =
timer.newTimeout(this, readerIdleTimeMillis, TimeUnit.MILLISECONDS);
try {
channelIdle(ctx, IdleState.READER_IDLE, lastReadTime);
@ -357,7 +362,7 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
}
} else {
// Read occurred before the timeout - set a new timeout with shorter delay.
readerIdleTimeout =
state.readerIdleTimeout =
timer.newTimeout(this, nextDelay, TimeUnit.MILLISECONDS);
}
}
@ -378,12 +383,13 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
return;
}
State state = (State) ctx.getAttachment();
long currentTime = System.currentTimeMillis();
long lastWriteTime = IdleStateHandler.this.lastWriteTime;
long lastWriteTime = state.lastWriteTime;
long nextDelay = writerIdleTimeMillis - (currentTime - lastWriteTime);
if (nextDelay <= 0) {
// Writer is idle - set a new timeout and notify the callback.
writerIdleTimeout =
state.writerIdleTimeout =
timer.newTimeout(this, writerIdleTimeMillis, TimeUnit.MILLISECONDS);
try {
channelIdle(ctx, IdleState.WRITER_IDLE, lastWriteTime);
@ -392,7 +398,7 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
}
} else {
// Write occurred before the timeout - set a new timeout with shorter delay.
writerIdleTimeout =
state.writerIdleTimeout =
timer.newTimeout(this, nextDelay, TimeUnit.MILLISECONDS);
}
}
@ -412,13 +418,14 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
return;
}
State state = (State) ctx.getAttachment();
long currentTime = System.currentTimeMillis();
long lastIoTime = Math.max(lastReadTime, lastWriteTime);
long lastIoTime = Math.max(state.lastReadTime, state.lastWriteTime);
long nextDelay = allIdleTimeMillis - (currentTime - lastIoTime);
if (nextDelay <= 0) {
// Both reader and writer are idle - set a new timeout and
// notify the callback.
allIdleTimeout =
state.allIdleTimeout =
timer.newTimeout(this, allIdleTimeMillis, TimeUnit.MILLISECONDS);
try {
channelIdle(ctx, IdleState.ALL_IDLE, lastIoTime);
@ -428,9 +435,23 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
} else {
// Either read or write occurred before the timeout - set a new
// timeout with shorter delay.
allIdleTimeout =
state.allIdleTimeout =
timer.newTimeout(this, nextDelay, TimeUnit.MILLISECONDS);
}
}
}
private static final class State {
State() {
super();
}
volatile Timeout readerIdleTimeout;
volatile long lastReadTime;
volatile Timeout writerIdleTimeout;
volatile long lastWriteTime;
volatile Timeout allIdleTimeout;
}
}

View File

@ -20,6 +20,8 @@ import static org.jboss.netty.channel.Channels.*;
import java.util.concurrent.TimeUnit;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandler.Sharable;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
@ -42,15 +44,17 @@ import org.jboss.netty.util.TimerTask;
* public class MyPipelineFactory implements {@link ChannelPipelineFactory} {
*
* private final {@link Timer} timer;
* private final {@link ChannelHandler} timeoutHandler;
*
* public MyPipelineFactory({@link Timer} timer) {
* this.timer = timer;
* this.timeoutHandler = <b>new {@link ReadTimeoutHandler}(timer, 30), // timer must be shared.</b>
* }
*
* public {@link ChannelPipeline} getPipeline() {
* // An example configuration that implements 30-second read timeout:
* return {@link Channels}.pipeline(
* <b>new {@link ReadTimeoutHandler}(timer, 30), // timer must be shared.</b>
* timeoutHandler,
* new MyHandler());
* }
* }
@ -77,6 +81,7 @@ import org.jboss.netty.util.TimerTask;
* @apiviz.uses org.jboss.netty.util.HashedWheelTimer
* @apiviz.has org.jboss.netty.handler.timeout.TimeoutException oneway - - raises
*/
@Sharable
public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
implements LifeCycleAwareChannelHandler,
ExternalResourceReleasable {
@ -85,9 +90,6 @@ public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
final Timer timer;
final long timeoutMillis;
volatile Timeout timeout;
private volatile ReadTimeoutTask task;
volatile long lastReadTime;
/**
* Creates a new instance.
@ -159,7 +161,7 @@ public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
@Override
public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
destroy();
destroy(ctx);
}
@Override
@ -180,35 +182,32 @@ public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
@Override
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
destroy();
destroy(ctx);
ctx.sendUpstream(e);
}
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
updateLastReadTime();
State state = (State) ctx.getAttachment();
state.lastReadTime = System.currentTimeMillis();
ctx.sendUpstream(e);
}
private void initialize(ChannelHandlerContext ctx) {
updateLastReadTime();
task = new ReadTimeoutTask(ctx);
State state = new State();
ctx.setAttachment(state);
if (timeoutMillis > 0) {
timeout = timer.newTimeout(task, timeoutMillis, TimeUnit.MILLISECONDS);
state.timeout = timer.newTimeout(new ReadTimeoutTask(ctx), timeoutMillis, TimeUnit.MILLISECONDS);
}
}
private void updateLastReadTime() {
lastReadTime = System.currentTimeMillis();
}
private void destroy() {
if (timeout != null) {
timeout.cancel();
private void destroy(ChannelHandlerContext ctx) {
State state = (State) ctx.getAttachment();
if (state.timeout != null) {
state.timeout.cancel();
state.timeout = null;
}
timeout = null;
task = null;
}
protected void readTimedOut(ChannelHandlerContext ctx) throws Exception {
@ -233,22 +232,34 @@ public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
return;
}
State state = (State) ctx.getAttachment();
long currentTime = System.currentTimeMillis();
long nextDelay = timeoutMillis - (currentTime - lastReadTime);
long nextDelay = timeoutMillis - (currentTime - state.lastReadTime);
if (nextDelay <= 0) {
// Read timed out - set a new timeout and notify the callback.
ReadTimeoutHandler.this.timeout =
state.timeout =
timer.newTimeout(this, timeoutMillis, TimeUnit.MILLISECONDS);
try {
// FIXME This should be called from an I/O thread.
// To be fixed in Netty 4.
readTimedOut(ctx);
} catch (Throwable t) {
fireExceptionCaught(ctx, t);
}
} else {
// Read occurred before the timeout - set a new timeout with shorter delay.
ReadTimeoutHandler.this.timeout =
state.timeout =
timer.newTimeout(this, nextDelay, TimeUnit.MILLISECONDS);
}
}
}
private static final class State {
volatile Timeout timeout;
volatile long lastReadTime = System.currentTimeMillis();
State() {
super();
}
}
}

View File

@ -24,6 +24,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@ -472,11 +473,15 @@ public class HashedWheelTimer implements Timer {
private final class HashedWheelTimeout implements Timeout {
private static final int ST_INIT = 0;
private static final int ST_CANCELLED = 1;
private static final int ST_EXPIRED = 2;
private final TimerTask task;
final long deadline;
volatile int stopIndex;
volatile long remainingRounds;
private volatile boolean cancelled;
private final AtomicInteger state = new AtomicInteger(ST_INIT);
HashedWheelTimeout(TimerTask task, long deadline) {
this.task = task;
@ -495,28 +500,26 @@ public class HashedWheelTimer implements Timer {
@Override
public void cancel() {
if (isExpired()) {
if (!state.compareAndSet(ST_INIT, ST_CANCELLED)) {
// TODO return false
return;
}
cancelled = true;
// Might be called more than once, but doesn't matter.
wheel[stopIndex].remove(this);
}
@Override
public boolean isCancelled() {
return cancelled;
return state.get() == ST_CANCELLED;
}
@Override
public boolean isExpired() {
return cancelled || System.currentTimeMillis() > deadline;
return state.get() != ST_INIT;
}
public void expire() {
if (cancelled) {
if (!state.compareAndSet(ST_INIT, ST_EXPIRED)) {
return;
}

View File

@ -200,6 +200,19 @@ public class ChannelBuffersTest {
assertSame(EMPTY_BUFFER, copiedBuffer(new ChannelBuffer[] { buffer(0), buffer(0) }));
}
@Test
public void testCompare2() {
assertTrue(ChannelBuffers.compare(
ChannelBuffers.wrappedBuffer(new byte[]{(byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF}),
ChannelBuffers.wrappedBuffer(new byte[]{(byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00}))
> 0);
assertTrue(ChannelBuffers.compare(
ChannelBuffers.wrappedBuffer(new byte[]{(byte) 0xFF}),
ChannelBuffers.wrappedBuffer(new byte[]{(byte) 0x00}))
> 0);
}
@Test(expected = NullPointerException.class)
public void shouldDisallowNullEndian1() {
buffer(null, 0);

View File

@ -53,6 +53,7 @@ public class FakeChannelConfig implements SocketChannelConfig {
private ChannelPipelineFactory pipelineFactory =
new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() throws Exception {
return Channels.pipeline();
}
@ -60,87 +61,108 @@ public class FakeChannelConfig implements SocketChannelConfig {
private int writeTimeout = 3000;
@Override
public int getReceiveBufferSize() {
return receiveBufferSize;
}
@Override
public void setReceiveBufferSize(int receiveBufferSize) {
this.receiveBufferSize = receiveBufferSize;
}
@Override
public int getSendBufferSize() {
return sendBufferSize;
}
@Override
public void setSendBufferSize(int sendBufferSize) {
this.sendBufferSize = sendBufferSize;
}
@Override
public int getSoLinger() {
return soLinger;
}
@Override
public void setSoLinger(int soLinger) {
this.soLinger = soLinger;
}
@Override
public int getTrafficClass() {
return trafficClass;
}
@Override
public void setTrafficClass(int trafficClass) {
this.trafficClass = trafficClass;
}
@Override
public boolean isKeepAlive() {
return keepAlive;
}
@Override
public void setKeepAlive(boolean keepAlive) {
this.keepAlive = keepAlive;
}
@Override
public boolean isReuseAddress() {
return reuseAddress;
}
@Override
public void setReuseAddress(boolean reuseAddress) {
this.reuseAddress = reuseAddress;
}
@Override
public boolean isTcpNoDelay() {
return tcpNoDelay;
}
@Override
public void setTcpNoDelay(boolean tcpNoDelay) {
this.tcpNoDelay = tcpNoDelay;
}
@Override
public void setPerformancePreferences(int connectionTime, int latency,
int bandwidth) {
// do nothing
}
@Override
public ChannelBufferFactory getBufferFactory() {
return bufferFactory;
}
@Override
public void setBufferFactory(ChannelBufferFactory bufferFactory) {
this.bufferFactory = bufferFactory;
}
@Override
public int getConnectTimeoutMillis() {
return connectTimeout;
}
@Override
public void setConnectTimeoutMillis(int connectTimeoutMillis) {
connectTimeout = connectTimeoutMillis;
}
@Override
public ChannelPipelineFactory getPipelineFactory() {
return pipelineFactory;
}
@Override
public void setPipelineFactory(ChannelPipelineFactory pipelineFactory) {
this.pipelineFactory = pipelineFactory;
}
@ -153,6 +175,7 @@ public class FakeChannelConfig implements SocketChannelConfig {
writeTimeout = writeTimeoutMillis;
}
@Override
public boolean setOption(String key, Object value) {
if (key.equals("pipelineFactory")) {
setPipelineFactory((ChannelPipelineFactory) value);
@ -180,6 +203,7 @@ public class FakeChannelConfig implements SocketChannelConfig {
return true;
}
@Override
public void setOptions(Map<String, Object> options) {
for (Entry<String, Object> e: options.entrySet()) {
setOption(e.getKey(), e.getValue());

View File

@ -31,6 +31,7 @@ public class FakeChannelSink extends AbstractChannelSink {
public Queue<ChannelEvent> events = new LinkedList<ChannelEvent>();
@Override
public void eventSunk(ChannelPipeline pipeline, ChannelEvent e)
throws Exception {
events.add(e);

View File

@ -36,6 +36,7 @@ public class FakeClientSocketChannelFactory implements
createdChannels = new ArrayList<FakeSocketChannel>();
}
@Override
public SocketChannel newChannel(ChannelPipeline pipeline) {
FakeSocketChannel channel =
new FakeSocketChannel(null, this, pipeline,
@ -44,6 +45,7 @@ public class FakeClientSocketChannelFactory implements
return channel;
}
@Override
public void releaseExternalResources() {
// nothing to do
}

View File

@ -52,22 +52,27 @@ public class FakeServerSocketChannel extends AbstractChannel implements
super(null, factory, pipeline, sink);
}
@Override
public ServerSocketChannelConfig getConfig() {
return config;
}
@Override
public InetSocketAddress getLocalAddress() {
return localAddress;
}
@Override
public InetSocketAddress getRemoteAddress() {
return remoteAddress;
}
@Override
public boolean isBound() {
return bound;
}
@Override
public boolean isConnected() {
return connected;
}

View File

@ -43,30 +43,37 @@ public class FakeServerSocketChannelConfig extends DefaultChannelConfig
public ChannelBufferFactory bufferFactory = new HeapChannelBufferFactory();
@Override
public int getBacklog() {
return backlog;
}
@Override
public void setBacklog(int backlog) {
this.backlog = backlog;
}
@Override
public int getReceiveBufferSize() {
return receiveBufferSize;
}
@Override
public void setReceiveBufferSize(int receiveBufferSize) {
this.receiveBufferSize = receiveBufferSize;
}
@Override
public boolean isReuseAddress() {
return reuseAddress;
}
@Override
public void setReuseAddress(boolean reuseAddress) {
this.reuseAddress = reuseAddress;
}
@Override
public void setPerformancePreferences(int connectionTime, int latency,
int bandwidth) {
// ignore

View File

@ -32,11 +32,13 @@ public class FakeServerSocketChannelFactory implements
public FakeServerSocketChannel createdChannel;
@Override
public ServerSocketChannel newChannel(ChannelPipeline pipeline) {
createdChannel = new FakeServerSocketChannel(this, pipeline, sink);
return createdChannel;
}
@Override
public void releaseExternalResources() {
// nothing to do
}

View File

@ -52,22 +52,27 @@ public class FakeSocketChannel extends AbstractChannel implements SocketChannel
this.sink = sink;
}
@Override
public InetSocketAddress getLocalAddress() {
return localAddress;
}
@Override
public SocketChannelConfig getConfig() {
return config;
}
@Override
public InetSocketAddress getRemoteAddress() {
return remoteAddress;
}
@Override
public boolean isBound() {
return bound;
}
@Override
public boolean isConnected() {
return connected;
}

View File

@ -73,7 +73,7 @@ public class HttpTunnelSoakTester {
final ChannelGroup channels;
private final ExecutorService executor;
final ExecutorService executor;
final ScheduledExecutorService scheduledExecutor;
@ -81,9 +81,9 @@ public class HttpTunnelSoakTester {
final DataSender s2cDataSender = new DataSender("S2C");
private DataVerifier c2sVerifier = new DataVerifier("C2S-Verifier");
final DataVerifier c2sVerifier = new DataVerifier("C2S-Verifier");
private DataVerifier s2cVerifier = new DataVerifier("S2C-Verifier");
final DataVerifier s2cVerifier = new DataVerifier("S2C-Verifier");
private static byte[] SEND_STREAM;
@ -160,6 +160,7 @@ public class HttpTunnelSoakTester {
protected ChannelPipelineFactory createClientPipelineFactory() {
return new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("s2cVerifier", s2cVerifier);
@ -173,6 +174,7 @@ public class HttpTunnelSoakTester {
protected ChannelPipelineFactory createServerPipelineFactory() {
return new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("c2sVerifier", c2sVerifier);
@ -180,6 +182,7 @@ public class HttpTunnelSoakTester {
s2cDataSender));
pipeline.addLast("sendStarter",
new SimpleChannelUpstreamHandler() {
@Override
public void channelConnected(
ChannelHandlerContext ctx,
ChannelStateEvent e) throws Exception {
@ -187,7 +190,7 @@ public class HttpTunnelSoakTester {
channels.add(childChannel);
s2cDataSender.setChannel(childChannel);
executor.execute(s2cDataSender);
};
}
});
return pipeline;
}
@ -271,15 +274,15 @@ public class HttpTunnelSoakTester {
}
HttpTunnelClientChannelConfig config =
((HttpTunnelClientChannelConfig) clientChannelFuture
.getChannel().getConfig());
(HttpTunnelClientChannelConfig) clientChannelFuture
.getChannel().getConfig();
config.setWriteBufferHighWaterMark(2 * 1024 * 1024);
config.setWriteBufferLowWaterMark(1024 * 1024);
return (SocketChannel) clientChannelFuture.getChannel();
}
private ChannelBuffer createRandomSizeBuffer(AtomicInteger nextWriteByte) {
ChannelBuffer createRandomSizeBuffer(AtomicInteger nextWriteByte) {
Random random = new Random();
int arraySize = random.nextInt(MAX_WRITE_SIZE) + 1;
@ -309,13 +312,13 @@ public class HttpTunnelSoakTester {
}
private class DataVerifier extends SimpleChannelUpstreamHandler {
private String name;
private final String name;
private int expectedNext = 0;
private int verifiedBytes = 0;
private CountDownLatch completionLatch = new CountDownLatch(1);
private final CountDownLatch completionLatch = new CountDownLatch(1);
public DataVerifier(String name) {
this.name = name;
@ -376,7 +379,7 @@ public class HttpTunnelSoakTester {
private class DataSender implements Runnable {
private AtomicReference<Channel> channel =
private final AtomicReference<Channel> channel =
new AtomicReference<Channel>();
private long totalBytesSent = 0;
@ -387,15 +390,15 @@ public class HttpTunnelSoakTester {
private boolean firstRun = true;
private AtomicBoolean writeEnabled = new AtomicBoolean(true);
private final AtomicBoolean writeEnabled = new AtomicBoolean(true);
private AtomicBoolean running = new AtomicBoolean(false);
private final AtomicBoolean running = new AtomicBoolean(false);
private CountDownLatch finishLatch = new CountDownLatch(1);
private final CountDownLatch finishLatch = new CountDownLatch(1);
private String name;
private final String name;
private AtomicInteger nextWriteByte = new AtomicInteger(0);
private final AtomicInteger nextWriteByte = new AtomicInteger(0);
public DataSender(String name) {
this.name = name;
@ -407,7 +410,7 @@ public class HttpTunnelSoakTester {
public void setWriteEnabled(boolean enabled) {
writeEnabled.set(enabled);
if (enabled && !this.isRunning() && finishLatch.getCount() > 0) {
if (enabled && !isRunning() && finishLatch.getCount() > 0) {
executor.execute(this);
}
}

View File

@ -102,6 +102,7 @@ public class HttpTunnelTest {
clientCaptureHandler = new ClientEndHandler();
clientBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("clientCapture", clientCaptureHandler);
@ -117,6 +118,7 @@ public class HttpTunnelTest {
connectionCaptureHandler = new ServerEndHandler();
serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("capture", connectionCaptureHandler);

View File

@ -37,23 +37,28 @@ public class MockChannelStateListener implements HttpTunnelClientWorkerOwner {
public String serverHostName = null;
@Override
public void fullyEstablished() {
fullyEstablished = true;
}
@Override
public void onConnectRequest(ChannelFuture connectFuture,
InetSocketAddress remoteAddress) {
// not relevant for test
}
@Override
public void onMessageReceived(ChannelBuffer content) {
messages.add(content);
}
@Override
public void onTunnelOpened(String tunnelId) {
this.tunnelId = tunnelId;
}
@Override
public String getServerHostName() {
return serverHostName;
}

View File

@ -28,11 +28,13 @@ import org.jboss.netty.channel.ChannelUpstreamHandler;
public class NullChannelHandler implements ChannelUpstreamHandler,
ChannelDownstreamHandler {
@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
throws Exception {
ctx.sendUpstream(e);
}
@Override
public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e)
throws Exception {
ctx.sendDownstream(e);

View File

@ -16,7 +16,7 @@
package org.jboss.netty.channel.socket.http;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.*;
import java.net.InetSocketAddress;
@ -55,7 +55,7 @@ public class ServerMessageSwitchTest {
private FakeSocketChannel requesterChannel;
private HttpTunnelAcceptedChannelReceiver htunAcceptedChannel;
HttpTunnelAcceptedChannelReceiver htunAcceptedChannel;
@Before
public void setUp() throws Exception {

View File

@ -33,6 +33,7 @@ public class UpstreamEventCatcher implements ChannelUpstreamHandler {
public Queue<ChannelEvent> events = new LinkedList<ChannelEvent>();
@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
throws Exception {
events.add(e);

View File

@ -0,0 +1,49 @@
/*
* 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.handler.codec.frame;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.handler.codec.embedder.CodecEmbedderException;
import org.jboss.netty.handler.codec.embedder.DecoderEmbedder;
import org.jboss.netty.util.CharsetUtil;
import org.junit.Assert;
import org.junit.Test;
/**
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
*/
public class DelimiterBasedFrameDecoderTest {
@Test
public void testTooLongFrameRecovery() throws Exception {
DecoderEmbedder<ChannelBuffer> embedder = new DecoderEmbedder<ChannelBuffer>(
new DelimiterBasedFrameDecoder(1, Delimiters.nulDelimiter()));
for (int i = 0; i < 2; i ++) {
try {
embedder.offer(ChannelBuffers.wrappedBuffer(new byte[] { 1, 2, 0 }));
Assert.fail(CodecEmbedderException.class.getSimpleName() + " must be raised.");
} catch (CodecEmbedderException e) {
Assert.assertTrue(e.getCause() instanceof TooLongFrameException);
// Expected
}
embedder.offer(ChannelBuffers.wrappedBuffer(new byte[] { 'A', 0 }));
ChannelBuffer buf = embedder.poll();
Assert.assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
}
}
}

View File

@ -0,0 +1,49 @@
/*
* 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.handler.codec.frame;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.handler.codec.embedder.CodecEmbedderException;
import org.jboss.netty.handler.codec.embedder.DecoderEmbedder;
import org.jboss.netty.util.CharsetUtil;
import org.junit.Assert;
import org.junit.Test;
/**
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
*/
public class LengthFieldBasedFrameDecoderTest {
@Test
public void testTooLongFrameRecovery() throws Exception {
DecoderEmbedder<ChannelBuffer> embedder = new DecoderEmbedder<ChannelBuffer>(
new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4));
for (int i = 0; i < 2; i ++) {
try {
embedder.offer(ChannelBuffers.wrappedBuffer(new byte[] { 0, 0, 0, 2, 0, 0 }));
Assert.fail(CodecEmbedderException.class.getSimpleName() + " must be raised.");
} catch (CodecEmbedderException e) {
Assert.assertTrue(e.getCause() instanceof TooLongFrameException);
// Expected
}
embedder.offer(ChannelBuffers.wrappedBuffer(new byte[] { 0, 0, 0, 1, 'A' }));
ChannelBuffer buf = embedder.poll();
Assert.assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
}
}
}

View File

@ -15,7 +15,12 @@
*/
package org.jboss.netty.handler.codec.http;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.Date;
import java.util.Iterator;
@ -33,7 +38,7 @@ public class CookieDecoderTest {
@Test
public void testDecodingSingleCookieV0() {
String cookieString = "myCookie=myValue;expires=XXX;path=/apathsomewhere;domain=.adomainsomewhere;secure;";
cookieString = cookieString.replace("XXX", new CookieDateFormat().format(new Date(System.currentTimeMillis() + 50000)));
cookieString = cookieString.replace("XXX", new HttpHeaderDateFormat().format(new Date(System.currentTimeMillis() + 50000)));
CookieDecoder cookieDecoder = new CookieDecoder();
Set<Cookie> cookies = cookieDecoder.decode(cookieString);

View File

@ -15,7 +15,9 @@
*/
package org.jboss.netty.handler.codec.http;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.junit.Assert.assertNotNull;
import java.text.DateFormat;
import java.util.Date;
@ -33,7 +35,7 @@ public class CookieEncoderTest {
@Test
public void testEncodingSingleCookieV0() {
String result = "myCookie=myValue;Expires=XXX;Path=/apathsomewhere;Domain=.adomainsomewhere;Secure";
DateFormat df = new CookieDateFormat();
DateFormat df = new HttpHeaderDateFormat();
Cookie cookie = new DefaultCookie("myCookie", "myValue");
CookieEncoder encoder = new CookieEncoder(true);
encoder.addCookie(cookie);
@ -131,4 +133,16 @@ public class CookieEncoderTest {
String encodedCookie = encoder.encode();
assertEquals(c1 + c2 + c3, encodedCookie);
}
@Test
public void testEncodingWithNoCookies() {
CookieEncoder encoderForServer = new CookieEncoder(true);
String encodedCookie1 = encoderForServer.encode();
CookieEncoder encoderForClient = new CookieEncoder(false);
String encodedCookie2 = encoderForClient.encode();
assertNotNull(encodedCookie1);
assertNotNull(encodedCookie2);
}
}

View File

@ -0,0 +1,58 @@
package org.jboss.netty.handler.codec.http;
import java.text.ParseException;
import java.util.Date;
import junit.framework.Assert;
import org.junit.Test;
public class HttpHeaderDateFormatTest {
/**
* This date is set at "06 Nov 1994 08:49:37 GMT" (same used in example in
* RFC documentation)
* <p>
* http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html
*/
private static final Date DATE = new Date(784111777000L);
@Test
public void testParse() throws ParseException {
HttpHeaderDateFormat format = new HttpHeaderDateFormat();
{
final Date parsed = format.parse("Sun, 6 Nov 1994 08:49:37 GMT");
Assert.assertNotNull(parsed);
Assert.assertEquals(DATE, parsed);
}
{
final Date parsed = format.parse("Sun, 06 Nov 1994 08:49:37 GMT");
Assert.assertNotNull(parsed);
Assert.assertEquals(DATE, parsed);
}
{
final Date parsed = format.parse("Sunday, 06-Nov-94 08:49:37 GMT");
Assert.assertNotNull(parsed);
Assert.assertEquals(DATE, parsed);
}
{
final Date parsed = format.parse("Sunday, 6-Nov-94 08:49:37 GMT");
Assert.assertNotNull(parsed);
Assert.assertEquals(DATE, parsed);
}
{
final Date parsed = format.parse("Sun Nov 6 08:49:37 1994");
Assert.assertNotNull(parsed);
Assert.assertEquals(DATE, parsed);
}
}
@Test
public void testFormat() {
HttpHeaderDateFormat format = new HttpHeaderDateFormat();
final String formatted = format.format(DATE);
Assert.assertNotNull(formatted);
Assert.assertEquals("Sun, 06 Nov 1994 08:49:37 GMT", formatted);
}
}

View File

@ -0,0 +1,39 @@
/*
* Copyright 2011 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.handler.codec.serialization;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
/**
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
*/
public class CompactObjectSerializationTest {
@Test
public void testInterfaceSerialization() throws Exception {
PipedOutputStream pipeOut = new PipedOutputStream();
PipedInputStream pipeIn = new PipedInputStream(pipeOut);
CompactObjectOutputStream out = new CompactObjectOutputStream(pipeOut);
CompactObjectInputStream in = new CompactObjectInputStream(pipeIn);
out.writeObject(List.class);
Assert.assertSame(List.class, in.readObject());
}
}

View File

@ -19,260 +19,260 @@ import org.junit.Test;
public class IpFilterRuleTest extends TestCase
{
public static boolean accept(IpFilterRuleHandler h, InetSocketAddress addr) throws Exception
{
return h.accept(new ChannelHandlerContext()
public static boolean accept(IpFilterRuleHandler h, InetSocketAddress addr) throws Exception
{
return h.accept(new ChannelHandlerContext()
{
@Override
public boolean canHandleDownstream()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean canHandleDownstream()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean canHandleUpstream()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean canHandleUpstream()
{
// TODO Auto-generated method stub
return false;
}
@Override
public Object getAttachment()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Object getAttachment()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Channel getChannel()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Channel getChannel()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelHandler getHandler()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelHandler getHandler()
{
// TODO Auto-generated method stub
return null;
}
@Override
public String getName()
{
// TODO Auto-generated method stub
return null;
}
@Override
public String getName()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelPipeline getPipeline()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelPipeline getPipeline()
{
// TODO Auto-generated method stub
return null;
}
@Override
public void sendDownstream(ChannelEvent e)
{
// TODO Auto-generated method stub
}
@Override
public void sendDownstream(ChannelEvent e)
{
// TODO Auto-generated method stub
}
@Override
public void sendUpstream(ChannelEvent e)
{
// TODO Auto-generated method stub
}
@Override
public void sendUpstream(ChannelEvent e)
{
// TODO Auto-generated method stub
}
@Override
public void setAttachment(Object attachment)
{
// TODO Auto-generated method stub
}
@Override
public void setAttachment(Object attachment)
{
// TODO Auto-generated method stub
}
},
new UpstreamMessageEvent(new Channel()
{
@Override
public ChannelFuture bind(SocketAddress localAddress)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture bind(SocketAddress localAddress)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture close()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture close()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture connect(SocketAddress remoteAddress)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture connect(SocketAddress remoteAddress)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture disconnect()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture disconnect()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture getCloseFuture()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture getCloseFuture()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelConfig getConfig()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelConfig getConfig()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFactory getFactory()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFactory getFactory()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Integer getId()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Integer getId()
{
// TODO Auto-generated method stub
return null;
}
@Override
public int getInterestOps()
{
// TODO Auto-generated method stub
return 0;
}
@Override
public int getInterestOps()
{
// TODO Auto-generated method stub
return 0;
}
@Override
public SocketAddress getLocalAddress()
{
// TODO Auto-generated method stub
return null;
}
@Override
public SocketAddress getLocalAddress()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Channel getParent()
{
// TODO Auto-generated method stub
return null;
}
@Override
public Channel getParent()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelPipeline getPipeline()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelPipeline getPipeline()
{
// TODO Auto-generated method stub
return null;
}
@Override
public SocketAddress getRemoteAddress()
{
// TODO Auto-generated method stub
return null;
}
@Override
public SocketAddress getRemoteAddress()
{
// TODO Auto-generated method stub
return null;
}
@Override
public boolean isBound()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isBound()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isConnected()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isConnected()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isOpen()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isOpen()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isReadable()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isReadable()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isWritable()
{
// TODO Auto-generated method stub
return false;
}
@Override
public boolean isWritable()
{
// TODO Auto-generated method stub
return false;
}
@Override
public ChannelFuture setInterestOps(int interestOps)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture setInterestOps(int interestOps)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture setReadable(boolean readable)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture setReadable(boolean readable)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture unbind()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture unbind()
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture write(Object message)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture write(Object message)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture write(Object message, SocketAddress remoteAddress)
{
// TODO Auto-generated method stub
return null;
}
@Override
public ChannelFuture write(Object message, SocketAddress remoteAddress)
{
// TODO Auto-generated method stub
return null;
}
@Override
public int compareTo(Channel o)
{
// TODO Auto-generated method stub
return 0;
}
@Override
public int compareTo(Channel o)
{
// TODO Auto-generated method stub
return 0;
}
}, h, addr),
addr);
}
@Test
public void testIpFilterRule() throws Exception
{
}
@Test
public void testIpFilterRule() throws Exception
{
IpFilterRuleHandler h = new IpFilterRuleHandler();
h.addAll(new IpFilterRuleList("+n:localhost, -n:*"));
InetSocketAddress addr = new InetSocketAddress(InetAddress.getLocalHost(), 8080);
@ -317,6 +317,6 @@ public class IpFilterRuleTest extends TestCase
addr = new InetSocketAddress(InetAddress.getByName(InetAddress.getLocalHost().getHostName()), 8080);
assertTrue(accept(h, addr));
}
}
}