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 */ 017package org.apache.activemq.broker.region; 018 019import java.io.IOException; 020import java.util.ArrayList; 021import java.util.List; 022import java.util.concurrent.ConcurrentHashMap; 023import java.util.concurrent.ConcurrentMap; 024 025import javax.jms.JMSException; 026 027import org.apache.activemq.broker.Broker; 028import org.apache.activemq.broker.ConnectionContext; 029import org.apache.activemq.command.ConsumerInfo; 030import org.apache.activemq.command.MessageAck; 031import org.apache.activemq.command.MessageId; 032import org.apache.activemq.filter.MessageEvaluationContext; 033import org.apache.activemq.usage.SystemUsage; 034import org.slf4j.Logger; 035import org.slf4j.LoggerFactory; 036 037public class QueueBrowserSubscription extends QueueSubscription { 038 039 protected static final Logger LOG = LoggerFactory.getLogger(QueueBrowserSubscription.class); 040 041 int queueRefs; 042 boolean browseDone; 043 boolean destinationsAdded; 044 045 private final ConcurrentMap<MessageId, Object> audit = new ConcurrentHashMap<MessageId, Object>(); 046 private long maxMessages; 047 048 public QueueBrowserSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws JMSException { 049 super(broker, usageManager, context, info); 050 } 051 052 @Override 053 protected boolean canDispatch(MessageReference node) { 054 return !((QueueMessageReference) node).isAcked(); 055 } 056 057 @Override 058 public synchronized String toString() { 059 return "QueueBrowserSubscription:" + " consumer=" + info.getConsumerId() + 060 ", destinations=" + destinations.size() + ", dispatched=" + dispatched.size() + 061 ", delivered=" + this.prefetchExtension + ", pending=" + getPendingQueueSize(); 062 } 063 064 synchronized public void destinationsAdded() throws Exception { 065 destinationsAdded = true; 066 checkDone(); 067 } 068 069 public boolean isDuplicate(MessageId messageId) { 070 return audit.putIfAbsent(messageId, Boolean.TRUE) != null; 071 } 072 073 private void checkDone() throws Exception { 074 if (!browseDone && queueRefs == 0 && destinationsAdded) { 075 browseDone = true; 076 add(QueueMessageReference.NULL_MESSAGE); 077 } 078 } 079 080 @Override 081 public boolean matches(MessageReference node, MessageEvaluationContext context) throws IOException { 082 return !browseDone && super.matches(node, context); 083 } 084 085 /** 086 * Since we are a browser we don't really remove the message from the queue. 087 */ 088 @Override 089 protected void acknowledge(ConnectionContext context, final MessageAck ack, final MessageReference n) throws IOException { 090 if (info.isNetworkSubscription()) { 091 super.acknowledge(context, ack, n); 092 } 093 } 094 095 synchronized public void incrementQueueRef() { 096 queueRefs++; 097 } 098 099 synchronized public void decrementQueueRef() throws Exception { 100 if (queueRefs > 0) { 101 queueRefs--; 102 } 103 checkDone(); 104 } 105 106 @Override 107 public List<MessageReference> remove(ConnectionContext context, Destination destination) throws Exception { 108 super.remove(context, destination); 109 // there's no unacked messages that needs to be redelivered 110 // in case of browser 111 return new ArrayList<MessageReference>(); 112 } 113 114 public boolean atMax() { 115 return maxMessages > 0 && getEnqueueCounter() >= maxMessages; 116 } 117 118 public void setMaxMessages(long max) { 119 maxMessages = max; 120 } 121}