6:21 a.m.
Author: timfox
Date: 2010-07-28 07:21:06 -0400 (Wed, 28 Jul 2010)
New Revision: 9484
Modified:
trunk/src/main/org/hornetq/core/buffers/impl/ResetLimitWrappedHornetQBuffer.java
trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java
trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java
Log:
a few tweaks + fix message copy issue
Modified:
trunk/src/main/org/hornetq/core/buffers/impl/ResetLimitWrappedHornetQBuffer.java
===================================================================
---
trunk/src/main/org/hornetq/core/buffers/impl/ResetLimitWrappedHornetQBuffer.java 2010-07-28
07:52:35 UTC (rev 9483)
+++
trunk/src/main/org/hornetq/core/buffers/impl/ResetLimitWrappedHornetQBuffer.java 2010-07-28
11:21:06 UTC (rev 9484)
@@ -44,9 +44,9 @@
{
writerIndex(limit);
}
+
+ buffer.readerIndex(limit);
- readerIndex(limit);
-
this.message = message;
}
@@ -74,6 +74,8 @@
@Override
public void readerIndex(int readerIndex)
{
+ changed();
+
if (readerIndex < limit)
{
readerIndex = limit;
@@ -85,6 +87,8 @@
@Override
public void resetReaderIndex()
{
+ changed();
+
buffer.readerIndex(limit);
}
Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-07-28 07:52:35 UTC
(rev 9483)
+++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-07-28 11:21:06 UTC
(rev 9484)
@@ -422,7 +422,7 @@
}
}
- public void resetCopied()
+ public synchronized void resetCopied()
{
copied = false;
}
Modified: trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java 2010-07-28
07:52:35 UTC (rev 9483)
+++ trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java 2010-07-28
11:21:06 UTC (rev 9484)
@@ -337,7 +337,7 @@
*/
public ChannelPipeline getPipeline() throws Exception
{
- ChannelPipeline pipeline = new DefaultChannelPipeline();
+ List<ChannelHandler> handlers = new ArrayList<ChannelHandler>();
if (sslEnabled)
{
@@ -347,37 +347,40 @@
SslHandler handler = new SslHandler(engine);
- pipeline.addLast("ssl", handler);
+ handlers.add(handler);
}
if (httpEnabled)
{
- pipeline.addLast("http-decoder", new HttpRequestDecoder());
+ handlers.add(new HttpRequestDecoder());
- pipeline.addLast("http-encoder", new HttpResponseEncoder());
+ handlers.add(new HttpResponseEncoder());
- pipeline.addLast("http-handler", new
HttpAcceptorHandler(httpKeepAliveRunnable, httpResponseTime));
+ handlers.add(new HttpAcceptorHandler(httpKeepAliveRunnable,
httpResponseTime));
}
if (protocol == ProtocolType.CORE)
{
// Core protocol uses it's own optimised decoder
- pipeline.addLast("hornetq-decoder", new
HornetQFrameDecoder2());
+
+ handlers.add(new HornetQFrameDecoder2());
}
else if (protocol == ProtocolType.STOMP_WS)
{
- pipeline.addLast("http-decoder", new HttpRequestDecoder());
- pipeline.addLast("http-aggregator", new
HttpChunkAggregator(65536));
- pipeline.addLast("http-encoder", new HttpResponseEncoder());
- pipeline.addLast("hornetq-decoder", new
HornetQFrameDecoder(decoder));
- pipeline.addLast("websocket-handler", new
WebSocketServerHandler());
+ handlers.add(new HttpRequestDecoder());
+ handlers.add(new HttpChunkAggregator(65536));
+ handlers.add(new HttpResponseEncoder());
+ handlers.add(new HornetQFrameDecoder(decoder));
+ handlers.add(new WebSocketServerHandler());
}
else
{
- pipeline.addLast("hornetq-decoder", new
HornetQFrameDecoder(decoder));
+ handlers.add(new HornetQFrameDecoder(decoder));
}
- pipeline.addLast("handler", new
HornetQServerChannelHandler(channelGroup, handler, new Listener()));
+ handlers.add(new HornetQServerChannelHandler(channelGroup, handler, new
Listener()));
+
+ ChannelPipeline pipeline = new StaticChannelPipeline(handlers.toArray(new
ChannelHandler[handlers.size()]));
return pipeline;
}