[hornetq-commits] JBoss hornetq SVN: r8488 - in trunk/src/main/org/hornetq/core: journal/impl and 2 other directories.

Author: timfox Date: 2009-12-01 18:48:59 -0500 (Tue, 01 Dec 2009) New Revision: 8488 Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java Log: mainly remove log.infos Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-12-01 23:32:48 UTC (rev 8487) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-12-01 23:48:59 UTC (rev 8488) @@ -153,7 +153,7 @@ private final boolean blockOnPersistentSend; private final int minLargeMessageSize; - + private final int initialMessagePacketSize; private final boolean cacheLargeMessageClient; @@ -253,7 +253,7 @@ this.cacheLargeMessageClient = cacheLargeMessageClient; this.minLargeMessageSize = minLargeMessageSize; - + this.initialMessagePacketSize = initialMessagePacketSize; this.groupID = groupID; @@ -694,9 +694,9 @@ if (consumer != null) { ClientMessageInternal clMessage = (ClientMessageInternal)message.getMessage(); - + clMessage.setDeliveryCount(message.getDeliveryCount()); - + clMessage.setFlowControlSize(message.getPacketSize()); consumer.handleMessage(clMessage); @@ -779,7 +779,7 @@ { return; } - + boolean resetCreditManager = false; // We lock the channel to prevent any packets to be added to the resend @@ -801,13 +801,13 @@ ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request); if (response.isReattached()) - { + { // The session was found on the server - we reattached transparently ok channel.replayCommands(response.getLastReceivedCommandID(), channel.getID()); } else - { + { // The session wasn't found on the server - probably we're failing over onto a backup server where the // session // won't exist or the target server has been restarted - in this case the session will need to be recreated, @@ -831,8 +831,8 @@ { channel1.sendBlocking(createRequest); retry = false; - } - catch(HornetQException e) + } + catch (HornetQException e) { // the session was created while its server was starting, retry it: if (e.getCode() == HornetQException.SESSION_CREATION_REJECTED) @@ -847,9 +847,8 @@ throw e; } } - } while(retry); - - log.info("created session " + name); + } + while (retry); channel.clearCommands(); @@ -869,16 +868,16 @@ Connection conn = channel.getConnection().getTransportConnection(); HornetQBuffer buffer = createConsumerRequest.encode(channel.getConnection()); - + conn.write(buffer, false); int clientWindowSize = entry.getValue().getClientWindowSize(); - + if (clientWindowSize != 0) { SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(), clientWindowSize); - + packet.setChannelID(channel.getID()); buffer = packet.encode(channel.getConnection()); @@ -931,9 +930,9 @@ if (resetCreditManager) { producerCreditManager.reset(); - - //Also need to send more credits for consumers, otherwise the system could hand with the server - //not having any credits to send + + // Also need to send more credits for consumers, otherwise the system could hand with the server + // not having any credits to send } } @@ -1400,15 +1399,14 @@ // consumer if (windowSize != 0) - { + { channel.send(new SessionConsumerFlowCreditMessage(consumerID, windowSize)); } return consumer; } - private ClientProducer internalCreateProducer(final SimpleString address, - final int maxRate) throws HornetQException + private ClientProducer internalCreateProducer(final SimpleString address, final int maxRate) throws HornetQException { checkClosed(); @@ -1419,7 +1417,7 @@ autoCommitSends && blockOnNonPersistentSend, autoCommitSends && blockOnPersistentSend, autoGroup, - groupID == null?null:new SimpleString(groupID), + groupID == null ? null : new SimpleString(groupID), minLargeMessageSize, channel); Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-12-01 23:32:48 UTC (rev 8487) +++ trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-12-01 23:48:59 UTC (rev 8488) @@ -335,8 +335,6 @@ this.compactPercentage = (float)compactPercentage / 100f; } - log.info("creating journal with max io " + maxIO); - this.compactMinFiles = compactMinFiles; this.fileSize = fileSize; Modified: trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java 2009-12-01 23:32:48 UTC (rev 8487) +++ trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java 2009-12-01 23:48:59 UTC (rev 8488) @@ -97,9 +97,6 @@ public TimedBuffer(final int size, final int timeout, final boolean logRates) { - log.info("timed buffer size " + size); - log.info("timed buffer timeout " + timeout); - bufferSize = size; this.logRates = logRates; Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-12-01 23:32:48 UTC (rev 8487) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-12-01 23:48:59 UTC (rev 8488) @@ -241,17 +241,16 @@ if (journalTypeToUse == JournalType.ASYNCIO) { - log.info("AIO journal selected"); + log.info("Using AIO Journal"); journalFF = new AIOSequentialFileFactory(journalDir, config.getJournalBufferSize_AIO(), config.getJournalBufferTimeout_AIO(), config.isLogJournalWriteRate()); - log.info("AIO loaded successfully"); } else if (config.getJournalType() == JournalType.NIO) { - log.info("NIO Journal selected"); + log.info("Using NIO Journal"); journalFF = new NIOSequentialFileFactory(journalDir, true, config.getJournalBufferSize_NIO(), Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-12-01 23:32:48 UTC (rev 8487) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-12-01 23:48:59 UTC (rev 8488) @@ -1187,13 +1187,6 @@ } initialised = true; - - log.info("********** initialised"); - - if (System.getProperty("org.hornetq.opt.routeblast") != null) - { - runRouteBlast(); - } } /** @@ -1481,197 +1474,7 @@ throw new IllegalArgumentException("Error instantiating transformer class \"" + className + "\"", e); } } - - // private void runRouteBlastNoWait() throws Exception - // { - // SimpleString address = new SimpleString("rbnw_address"); - // SimpleString queueName = new SimpleString("rbnw_name"); - // - // createQueue(address, queueName, null, true, false, true); - // - // Queue queue = (Queue)postOffice.getBinding(queueName).getBindable(); - // - // RBConsumer consumer = new RBConsumer(queue); - // - // queue.addConsumer(consumer); - // - // final int bodySize = 1024; - // - // byte[] body = new byte[bodySize]; - // - // final int numMessages = 10000000; - // - // for (int i = 0; i < numMessages; i++) - // { - // final ServerMessage msg = new ServerMessageImpl(storageManager.generateUniqueID(), 1500); - // - // msg.getBodyBuffer().writeBytes(body); - // - // msg.setDestination(address); - // - // msg.setDurable(true); - // - // postOffice.route(msg); - // } - // } - - private LinkedBlockingQueue<RouteBlastRunner> available = new LinkedBlockingQueue<RouteBlastRunner>(); - - private void runRouteBlast() throws Exception - { - log.info("*** running route blast"); - - final int numThreads = 1; - - final int numClients = 1000; - - for (int i = 0; i < numClients; i++) - { - RouteBlastRunner run = new RouteBlastRunner(new SimpleString("fooaddress" + i)); - - run.setup(); - - available.add(run); - } - - log.info("setup, now running"); - - Set<Thread> runners = new HashSet<Thread>(); - - for (int i = 0; i < numThreads; i++) - { - Thread t = new Thread(new Foo()); - - runners.add(t); - - t.start(); - } - - for (Thread t : runners) - { - t.join(); - } - } - - class RouteBlastRunner implements Runnable - { - private SimpleString address; - - private Set<Consumer> consumers = new HashSet<Consumer>(); - - RouteBlastRunner(SimpleString address) - { - this.address = address; - } - - public void setup() throws Exception - { - final int numQueues = 1; - - for (int i = 0; i < numQueues; i++) - { - SimpleString queueName = new SimpleString(address + ".hq.route_blast_queue" + i); - - createQueue(address, queueName, null, true, false, true); - - Queue queue = (Queue)postOffice.getBinding(queueName).getBindable(); - - RBConsumer consumer = new RBConsumer(queue); - - queue.addConsumer(consumer); - - // log.info("added consumer to queue " + queue); - - consumers.add(consumer); - } - } - - public void run() - { - try - { - final int bodySize = 1024; - - byte[] body = new byte[bodySize]; - - final ServerMessage msg = new ServerMessageImpl(storageManager.generateUniqueID(), 1500); - - msg.getBodyBuffer().writeBytes(body); - - msg.setDestination(address); - - msg.setDurable(true); - - postOffice.route(msg); - - storageManager.afterCompleteOperations(new IOAsyncTask() - { - public void onError(int errorCode, String errorMessage) - { - log.error("Error processing IOCallback code = " + errorCode + " message = " + errorMessage); - } - - public void done() - { - available.add(RouteBlastRunner.this); - } - }); - } - catch (Exception e) - { - log.error("Failed to run runner", e); - } - - } - } - - class Foo implements Runnable - { - public void run() - { - for (int i = 0; i < 1000000; i++) - { - try - { - RouteBlastRunner runner = available.take(); - - runner.run(); - } - catch (InterruptedException e) - { - log.error("Interrupted", e); - } - } - } - } - - private class RBConsumer implements Consumer - { - private Queue queue; - - RBConsumer(Queue queue) - { - this.queue = queue; - } - - public Filter getFilter() - { - return null; - } - - public HandleStatus handle(MessageReference reference) throws Exception - { - reference.handled(); - - queue.acknowledge(reference); - - // log.info("acking"); - - return HandleStatus.HANDLED; - } - - } - + // Inner classes // -------------------------------------------------------------------------------- }