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.activemq;
018
019 import java.io.Serializable;
020 import org.slf4j.Logger;
021 import org.slf4j.LoggerFactory;
022
023 /**
024 * Defines the prefetch message policies for different types of consumers
025 *
026 * @org.apache.xbean.XBean element="prefetchPolicy"
027 *
028 */
029 @SuppressWarnings("serial")
030 public class ActiveMQPrefetchPolicy extends Object implements Serializable {
031 public static final int MAX_PREFETCH_SIZE = Short.MAX_VALUE;
032 public static final int DEFAULT_QUEUE_PREFETCH = 1000;
033 public static final int DEFAULT_QUEUE_BROWSER_PREFETCH = 500;
034 public static final int DEFAULT_DURABLE_TOPIC_PREFETCH = 100;
035 public static final int DEFAULT_OPTIMIZE_DURABLE_TOPIC_PREFETCH=1000;
036 public static final int DEFAULT_INPUT_STREAM_PREFETCH=100;
037 public static final int DEFAULT_TOPIC_PREFETCH = MAX_PREFETCH_SIZE;
038
039 private static final Logger LOG = LoggerFactory.getLogger(ActiveMQPrefetchPolicy.class);
040
041 private int queuePrefetch;
042 private int queueBrowserPrefetch;
043 private int topicPrefetch;
044 private int durableTopicPrefetch;
045 private int optimizeDurableTopicPrefetch;
046 private int inputStreamPrefetch;
047 private int maximumPendingMessageLimit;
048
049 /**
050 * Initialize default prefetch policies
051 */
052 public ActiveMQPrefetchPolicy() {
053 this.queuePrefetch = DEFAULT_QUEUE_PREFETCH;
054 this.queueBrowserPrefetch = DEFAULT_QUEUE_BROWSER_PREFETCH;
055 this.topicPrefetch = DEFAULT_TOPIC_PREFETCH;
056 this.durableTopicPrefetch = DEFAULT_DURABLE_TOPIC_PREFETCH;
057 this.optimizeDurableTopicPrefetch = DEFAULT_OPTIMIZE_DURABLE_TOPIC_PREFETCH;
058 this.inputStreamPrefetch = DEFAULT_INPUT_STREAM_PREFETCH;
059 }
060
061 /**
062 * @return Returns the durableTopicPrefetch.
063 */
064 public int getDurableTopicPrefetch() {
065 return durableTopicPrefetch;
066 }
067
068 /**
069 * @param durableTopicPrefetch The durableTopicPrefetch to set.
070 */
071 public void setDurableTopicPrefetch(int durableTopicPrefetch) {
072 this.durableTopicPrefetch = getMaxPrefetchLimit(durableTopicPrefetch);
073 }
074
075 /**
076 * @return Returns the queuePrefetch.
077 */
078 public int getQueuePrefetch() {
079 return queuePrefetch;
080 }
081
082 /**
083 * @param queuePrefetch The queuePrefetch to set.
084 */
085 public void setQueuePrefetch(int queuePrefetch) {
086 this.queuePrefetch = getMaxPrefetchLimit(queuePrefetch);
087 }
088
089 /**
090 * @return Returns the queueBrowserPrefetch.
091 */
092 public int getQueueBrowserPrefetch() {
093 return queueBrowserPrefetch;
094 }
095
096 /**
097 * @param queueBrowserPrefetch The queueBrowserPrefetch to set.
098 */
099 public void setQueueBrowserPrefetch(int queueBrowserPrefetch) {
100 this.queueBrowserPrefetch = getMaxPrefetchLimit(queueBrowserPrefetch);
101 }
102
103 /**
104 * @return Returns the topicPrefetch.
105 */
106 public int getTopicPrefetch() {
107 return topicPrefetch;
108 }
109
110 /**
111 * @param topicPrefetch The topicPrefetch to set.
112 */
113 public void setTopicPrefetch(int topicPrefetch) {
114 this.topicPrefetch = getMaxPrefetchLimit(topicPrefetch);
115 }
116
117 /**
118 * @return Returns the optimizeDurableTopicPrefetch.
119 */
120 public int getOptimizeDurableTopicPrefetch() {
121 return optimizeDurableTopicPrefetch;
122 }
123
124 /**
125 * @param optimizeAcknowledgePrefetch The optimizeDurableTopicPrefetch to
126 * set.
127 */
128 public void setOptimizeDurableTopicPrefetch(int optimizeAcknowledgePrefetch) {
129 this.optimizeDurableTopicPrefetch = optimizeAcknowledgePrefetch;
130 }
131
132 public int getMaximumPendingMessageLimit() {
133 return maximumPendingMessageLimit;
134 }
135
136 /**
137 * Sets how many messages a broker will keep around, above the prefetch
138 * limit, for non-durable topics before starting to discard older messages.
139 */
140 public void setMaximumPendingMessageLimit(int maximumPendingMessageLimit) {
141 this.maximumPendingMessageLimit = maximumPendingMessageLimit;
142 }
143
144 private int getMaxPrefetchLimit(int value) {
145 int result = Math.min(value, MAX_PREFETCH_SIZE);
146 if (result < value) {
147 LOG.warn("maximum prefetch limit has been reset from " + value + " to " + MAX_PREFETCH_SIZE);
148 }
149 return result;
150 }
151
152 public void setAll(int i) {
153 this.durableTopicPrefetch = i;
154 this.queueBrowserPrefetch = i;
155 this.queuePrefetch = i;
156 this.topicPrefetch = i;
157 this.inputStreamPrefetch = 1;
158 this.optimizeDurableTopicPrefetch = i;
159 }
160
161 public int getInputStreamPrefetch() {
162 return inputStreamPrefetch;
163 }
164
165 public void setInputStreamPrefetch(int inputStreamPrefetch) {
166 this.inputStreamPrefetch = getMaxPrefetchLimit(inputStreamPrefetch);
167 }
168
169 public boolean equals(Object object){
170 if (object instanceof ActiveMQPrefetchPolicy){
171 ActiveMQPrefetchPolicy other = (ActiveMQPrefetchPolicy) object;
172 return this.queuePrefetch == other.queuePrefetch &&
173 this.queueBrowserPrefetch == other.queueBrowserPrefetch &&
174 this.topicPrefetch == other.topicPrefetch &&
175 this.durableTopicPrefetch == other.durableTopicPrefetch &&
176 this.optimizeDurableTopicPrefetch == other.optimizeDurableTopicPrefetch &&
177 this.inputStreamPrefetch == other.inputStreamPrefetch;
178 }
179 return false;
180 }
181
182 }