001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.camel.processor;
018
019 import java.text.NumberFormat;
020 import java.util.concurrent.atomic.AtomicInteger;
021
022 import org.apache.camel.Exchange;
023 import org.apache.commons.logging.Log;
024
025 /**
026 * A logger for logging message throughput.
027 *
028 * @version $Revision: 659760 $
029 */
030 public class ThroughputLogger extends Logger {
031 private int groupSize = 100;
032 private long startTime;
033 private long groupStartTime;
034 private AtomicInteger receivedCounter = new AtomicInteger();
035 private NumberFormat numberFormat = NumberFormat.getNumberInstance();
036 private String action = "Received";
037 private String logMessage;
038
039 public ThroughputLogger() {
040 }
041
042 public ThroughputLogger(Log log) {
043 super(log);
044 }
045
046 public ThroughputLogger(Log log, LoggingLevel level) {
047 super(log, level);
048 }
049
050 public ThroughputLogger(String logName) {
051 super(logName);
052 }
053
054 public ThroughputLogger(String logName, LoggingLevel level) {
055 super(logName, level);
056 }
057
058 public ThroughputLogger(String logName, LoggingLevel level, int groupSize) {
059 super(logName, level);
060 setGroupSize(groupSize);
061 }
062
063 public ThroughputLogger(String logName, int groupSize) {
064 super(logName);
065 setGroupSize(groupSize);
066 }
067
068 public ThroughputLogger(int groupSize) {
069 setGroupSize(groupSize);
070 }
071
072 @Override
073 public void process(Exchange exchange) {
074 if (startTime == 0) {
075 startTime = System.currentTimeMillis();
076 }
077 int receivedCount = receivedCounter.incrementAndGet();
078 if (receivedCount % groupSize == 0) {
079 logMessage = createLogMessage(exchange, receivedCount);
080 super.process(exchange);
081 }
082 }
083
084 public int getGroupSize() {
085 return groupSize;
086 }
087
088 public void setGroupSize(int groupSize) {
089 if (groupSize == 0) {
090 throw new IllegalArgumentException("groupSize cannot be zero!");
091 }
092 this.groupSize = groupSize;
093 }
094
095 public NumberFormat getNumberFormat() {
096 return numberFormat;
097 }
098
099 public void setNumberFormat(NumberFormat numberFormat) {
100 this.numberFormat = numberFormat;
101 }
102
103 public String getAction() {
104 return action;
105 }
106
107 public void setAction(String action) {
108 this.action = action;
109 }
110
111 @Override
112 protected Object logMessage(Exchange exchange) {
113 return logMessage;
114 }
115
116 protected String createLogMessage(Exchange exchange, int receivedCount) {
117 long time = System.currentTimeMillis();
118 if (groupStartTime == 0) {
119 groupStartTime = startTime;
120 }
121
122 double rate = messagesPerSecond(groupSize, groupStartTime, time);
123 double average = messagesPerSecond(receivedCount, startTime, time);
124
125 groupStartTime = time;
126
127 return getAction() + ": " + receivedCount + " messages so far. Last group took: " + (time - groupStartTime)
128 + " millis which is: " + numberFormat.format(rate)
129 + " messages per second. average: " + numberFormat.format(average);
130 }
131
132 // timeOneMessage = elapsed / messageCount
133 // messagePerSend = 1000 / timeOneMessage
134 protected double messagesPerSecond(long messageCount, long startTime, long endTime) {
135 double rate = messageCount * 1000.0;
136 rate /= endTime - startTime;
137 return rate;
138 }
139 }