[hornetq-commits] JBoss hornetq SVN: r10037 - in trunk: tests/src/org/hornetq/tests/integration/http and 1 other directory.

Author: gaohoward Date: 2010-12-13 08:03:21 -0500 (Mon, 13 Dec 2010) New Revision: 10037 Modified: trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java trunk/tests/src/org/hornetq/tests/integration/http/CoreClientOverHttpTest.java Log: https://issues.jboss.org/browse/JBPAPP-5542 Modified: trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java 2010-12-11 10:45:00 UTC (rev 10036) +++ trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java 2010-12-13 13:03:21 UTC (rev 10037) @@ -347,6 +347,8 @@ if (httpEnabled) { handlers.put("http-decoder", new HttpRequestDecoder()); + + handlers.put("http-aggregator", new HttpChunkAggregator(Integer.MAX_VALUE)); handlers.put("http-encoder", new HttpResponseEncoder()); Modified: trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java 2010-12-11 10:45:00 UTC (rev 10036) +++ trunk/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java 2010-12-13 13:03:21 UTC (rev 10037) @@ -66,6 +66,7 @@ import org.jboss.netty.handler.codec.http.CookieDecoder; import org.jboss.netty.handler.codec.http.CookieEncoder; import org.jboss.netty.handler.codec.http.DefaultHttpRequest; +import org.jboss.netty.handler.codec.http.HttpChunkAggregator; import org.jboss.netty.handler.codec.http.HttpHeaders; import org.jboss.netty.handler.codec.http.HttpMethod; import org.jboss.netty.handler.codec.http.HttpRequest; @@ -354,6 +355,8 @@ handlers.add(new HttpRequestEncoder()); handlers.add(new HttpResponseDecoder()); + + handlers.add(new HttpChunkAggregator(Integer.MAX_VALUE)); handlers.add(new HttpHandler()); } Modified: trunk/tests/src/org/hornetq/tests/integration/http/CoreClientOverHttpTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/http/CoreClientOverHttpTest.java 2010-12-11 10:45:00 UTC (rev 10036) +++ trunk/tests/src/org/hornetq/tests/integration/http/CoreClientOverHttpTest.java 2010-12-13 13:03:21 UTC (rev 10037) @@ -13,6 +13,7 @@ package org.hornetq.tests.integration.http; import java.util.HashMap; +import java.util.Random; import junit.framework.Assert; @@ -127,4 +128,80 @@ server.stop(); } + + //https://issues.jboss.org/browse/JBPAPP-5542 + public void testCoreHttpClient8kPlus() throws Exception + { + final SimpleString QUEUE = new SimpleString("CoreClientOverHttpTestQueue"); + + Configuration conf = new ConfigurationImpl(); + + conf.setSecurityEnabled(false); + + HashMap<String, Object> params = new HashMap<String, Object>(); + params.put(TransportConstants.HTTP_ENABLED_PROP_NAME, true); + conf.getAcceptorConfigurations().add(new TransportConfiguration(NettyAcceptorFactory.class.getName(), params)); + + HornetQServer server = HornetQServers.newHornetQServer(conf, false); + + server.start(); + + ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(NettyConnectorFactory.class.getName(), params)); + ClientSessionFactory sf = locator.createSessionFactory(); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(QUEUE, QUEUE, null, false); + + ClientProducer producer = session.createProducer(QUEUE); + + final int numMessages = 100; + + String[] content = new String[numMessages]; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + content[i] = this.getFixedSizeString(i*1024*8); + message.getBodyBuffer().writeString(content[i]); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(QUEUE); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message2 = consumer.receive(); + + Assert.assertEquals(content[i], message2.getBodyBuffer().readString()); + + message2.acknowledge(); + } + + session.close(); + + locator.close(); + + server.stop(); + } + + private String getFixedSizeString(int size) + { + StringBuffer sb = new StringBuffer(); + Random r = new Random(); + for (int i = 0; i < size; i++) + { + char chr = (char)r.nextInt(256); + sb.append(chr); + } + String result = sb.toString(); + return result; + } + }