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.memory.list; 018 019 import java.util.ArrayList; 020 import java.util.Iterator; 021 import java.util.LinkedList; 022 import java.util.List; 023 import org.apache.activemq.broker.region.MessageReference; 024 import org.apache.activemq.command.ActiveMQDestination; 025 import org.apache.activemq.command.Message; 026 import org.apache.activemq.filter.DestinationFilter; 027 import org.slf4j.Logger; 028 import org.slf4j.LoggerFactory; 029 030 /** 031 * A simple fixed size {@link MessageList} where there is a single, fixed size 032 * list that all messages are added to for simplicity. Though this will lead to 033 * possibly slow recovery times as many more messages than is necessary will 034 * have to be iterated through for each subscription. 035 * 036 * 037 */ 038 public class SimpleMessageList implements MessageList { 039 private static final Logger LOG = LoggerFactory.getLogger(SimpleMessageList.class); 040 private final LinkedList<MessageReference> list = new LinkedList<MessageReference>(); 041 private int maximumSize = 100 * 64 * 1024; 042 private int size; 043 private final Object lock = new Object(); 044 045 public SimpleMessageList() { 046 } 047 048 public SimpleMessageList(int maximumSize) { 049 this.maximumSize = maximumSize; 050 } 051 052 public void add(MessageReference node) { 053 int delta = node.getMessageHardRef().getSize(); 054 synchronized (lock) { 055 list.add(node); 056 size += delta; 057 while (size > maximumSize) { 058 MessageReference evicted = list.removeFirst(); 059 size -= evicted.getMessageHardRef().getSize(); 060 } 061 } 062 } 063 064 public List<MessageReference> getMessages(ActiveMQDestination destination) { 065 return getList(); 066 } 067 068 public Message[] browse(ActiveMQDestination destination) { 069 List<Message> result = new ArrayList<Message>(); 070 DestinationFilter filter = DestinationFilter.parseFilter(destination); 071 synchronized (lock) { 072 for (Iterator<MessageReference> i = list.iterator(); i.hasNext();) { 073 MessageReference ref = i.next(); 074 Message msg; 075 msg = ref.getMessage(); 076 if (filter.matches(msg.getDestination())) { 077 result.add(msg); 078 } 079 080 } 081 } 082 return result.toArray(new Message[result.size()]); 083 } 084 085 /** 086 * Returns a copy of the list 087 */ 088 public List<MessageReference> getList() { 089 synchronized (lock) { 090 return new ArrayList<MessageReference>(list); 091 } 092 } 093 094 public int getSize() { 095 synchronized (lock) { 096 return size; 097 } 098 } 099 100 public void clear() { 101 synchronized (lock) { 102 list.clear(); 103 size = 0; 104 } 105 } 106 107 }