You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2008/04/21 16:04:00 UTC
svn commit: r650148 [3/10] - in
/incubator/qpid/branches/broker-queue-refactor/java: broker/ broker/etc/
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/ack/
broker/src/main/java/org/apache/qpid/server/configura...
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,448 @@
+package org.apache.qpid.server.exchange.topic;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.AMQShortStringTokenizer;
+
+import java.util.*;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public class TopicParser
+{
+ private static final byte TOPIC_DELIMITER = (byte)'.';
+
+ private final TopicWordDictionary _dictionary = new TopicWordDictionary();
+
+ private static class Position
+ {
+ private final TopicWord _word;
+ private final boolean _selfTransition;
+ private final int _position;
+ private final boolean _endState;
+
+
+ public Position(final int position, final TopicWord word, final boolean selfTransition, final boolean endState)
+ {
+ _position = position;
+ _word = word;
+ _selfTransition = selfTransition;
+ _endState = endState;
+ }
+
+
+ }
+
+ private static final Position ERROR_POSITION = new Position(Integer.MAX_VALUE,null, true, false);
+
+ private static class SimpleState
+ {
+ Set<Position> _positions;
+ Map<TopicWord, SimpleState> _nextState;
+ }
+
+
+ TopicMatcherDFAState createStateMachine(AMQShortString bindingKey, TopicMatcherResult result)
+ {
+ List<TopicWord> wordList = createTopicWordList(bindingKey);
+ int wildCards = 0;
+ for(TopicWord word : wordList)
+ {
+ if(word == TopicWord.WILDCARD_WORD)
+ {
+ wildCards++;
+ }
+ }
+ if(wildCards == 0)
+ {
+ TopicMatcherDFAState[] states = new TopicMatcherDFAState[wordList.size()+1];
+ states[states.length-1] = new TopicMatcherDFAState(Collections.EMPTY_MAP, Collections.singleton(result));
+ for(int i = states.length-2; i >= 0; i--)
+ {
+ states[i] = new TopicMatcherDFAState(Collections.singletonMap(wordList.get(i),states[i+1]),Collections.EMPTY_SET);
+
+ }
+ return states[0];
+ }
+ else if(wildCards == wordList.size())
+ {
+ Map<TopicWord,TopicMatcherDFAState> stateMap = new HashMap<TopicWord,TopicMatcherDFAState>();
+ TopicMatcherDFAState state = new TopicMatcherDFAState(stateMap, Collections.singleton(result));
+ stateMap.put(TopicWord.ANY_WORD, state);
+ return state;
+ }
+
+
+ int positionCount = wordList.size() - wildCards;
+
+ Position[] positions = new Position[positionCount+1];
+
+ int lastWord;
+
+ if(wordList.get(wordList.size()-1)== TopicWord.WILDCARD_WORD)
+ {
+ lastWord = wordList.size()-1;
+ positions[positionCount] = new Position(positionCount, TopicWord.ANY_WORD, true, true);
+ }
+ else
+ {
+ lastWord = wordList.size();
+ positions[positionCount] = new Position(positionCount, TopicWord.ANY_WORD, false, true);
+ }
+
+
+
+ int pos = 0;
+ int wordPos = 0;
+
+
+ while(wordPos < lastWord)
+ {
+ TopicWord word = wordList.get(wordPos++);
+
+ if(word == TopicWord.WILDCARD_WORD)
+ {
+ int nextWordPos = wordPos++;
+ word = wordList.get(nextWordPos);
+
+ positions[pos] = new Position(pos++,word,true,false);
+ }
+ else
+ {
+ positions[pos] = new Position(pos++,word,false,false);
+ }
+
+ }
+
+ // from each position you transition to a set of other positions.
+ // we approach this by examining steps of increasing length - so we
+ // look how far we can go from the start position in 1 word, 2 words, etc...
+
+ Map<Set<Position>,SimpleState> stateMap = new HashMap<Set<Position>,SimpleState>();
+
+
+ SimpleState state = new SimpleState();
+ state._positions = Collections.singleton( positions[0] );
+ stateMap.put(state._positions, state);
+
+ calculateNextStates(state, stateMap, positions);
+
+ SimpleState[] simpleStates = stateMap.values().toArray(new SimpleState[stateMap.size()]);
+ HashMap<TopicWord, TopicMatcherDFAState>[] dfaStateMaps = new HashMap[simpleStates.length];
+ Map<SimpleState, TopicMatcherDFAState> simple2DFAMap = new HashMap<SimpleState, TopicMatcherDFAState>();
+
+ for(int i = 0; i < simpleStates.length; i++)
+ {
+
+ Collection<TopicMatcherResult> results;
+ boolean endState = false;
+
+ for(Position p : simpleStates[i]._positions)
+ {
+ if(p._endState)
+ {
+ endState = true;
+ break;
+ }
+ }
+
+ if(endState)
+ {
+ results = Collections.singleton(result);
+ }
+ else
+ {
+ results = Collections.EMPTY_SET;
+ }
+
+ dfaStateMaps[i] = new HashMap<TopicWord, TopicMatcherDFAState>();
+ simple2DFAMap.put(simpleStates[i], new TopicMatcherDFAState(dfaStateMaps[i],results));
+
+ }
+ for(int i = 0; i < simpleStates.length; i++)
+ {
+ SimpleState simpleState = simpleStates[i];
+
+ Map<TopicWord, SimpleState> nextSimpleStateMap = simpleState._nextState;
+ for(Map.Entry<TopicWord, SimpleState> stateMapEntry : nextSimpleStateMap.entrySet())
+ {
+ dfaStateMaps[i].put(stateMapEntry.getKey(), simple2DFAMap.get(stateMapEntry.getValue()));
+ }
+
+ }
+
+ return simple2DFAMap.get(state);
+
+ }
+
+
+
+ private void calculateNextStates(final SimpleState state,
+ final Map<Set<Position>, SimpleState> stateMap,
+ final Position[] positions)
+ {
+ Map<TopicWord, Set<Position>> transitions = new HashMap<TopicWord,Set<Position>>();
+
+ for(Position pos : state._positions)
+ {
+ if(pos._selfTransition)
+ {
+ Set<Position> dest = transitions.get(TopicWord.ANY_WORD);
+ if(dest == null)
+ {
+ dest = new HashSet<Position>();
+ transitions.put(TopicWord.ANY_WORD,dest);
+ }
+ dest.add(pos);
+ }
+
+ final int nextPos = pos._position + 1;
+ Position nextPosition = nextPos == positions.length ? ERROR_POSITION : positions[nextPos];
+
+ Set<Position> dest = transitions.get(pos._word);
+ if(dest == null)
+ {
+ dest = new HashSet<Position>();
+ transitions.put(pos._word,dest);
+ }
+ dest.add(nextPosition);
+
+ }
+
+ Set<Position> anyWordTransitions = transitions.get(TopicWord.ANY_WORD);
+ if(anyWordTransitions != null)
+ {
+ for(Set<Position> dest : transitions.values())
+ {
+ dest.addAll(anyWordTransitions);
+ }
+ }
+
+ state._nextState = new HashMap<TopicWord, SimpleState>();
+
+ for(Map.Entry<TopicWord,Set<Position>> dest : transitions.entrySet())
+ {
+
+ if(dest.getValue().size()>1)
+ {
+ dest.getValue().remove(ERROR_POSITION);
+ }
+ Position loopingTerminal = null;
+ for(Position destPos : dest.getValue())
+ {
+ if(destPos._selfTransition && destPos._endState)
+ {
+ loopingTerminal = destPos;
+ break;
+ }
+ }
+
+ if(loopingTerminal!=null)
+ {
+ dest.setValue(Collections.singleton(loopingTerminal));
+ }
+
+ SimpleState stateForEntry = stateMap.get(dest.getValue());
+ if(stateForEntry == null)
+ {
+ stateForEntry = new SimpleState();
+ stateForEntry._positions = dest.getValue();
+ stateMap.put(dest.getValue(),stateForEntry);
+ calculateNextStates(stateForEntry,
+ stateMap,
+ positions);
+ }
+ state._nextState.put(dest.getKey(),stateForEntry);
+
+
+
+ }
+
+ // remove redundant transitions
+ SimpleState anyWordState = state._nextState.get(TopicWord.ANY_WORD);
+ if(anyWordState != null)
+ {
+ List<TopicWord> removeList = new ArrayList<TopicWord>();
+ for(Map.Entry<TopicWord,SimpleState> entry : state._nextState.entrySet())
+ {
+ if(entry.getValue() == anyWordState && entry.getKey() != TopicWord.ANY_WORD)
+ {
+ removeList.add(entry.getKey());
+ }
+ }
+ for(TopicWord removeKey : removeList)
+ {
+ state._nextState.remove(removeKey);
+ }
+ }
+
+
+ }
+
+ private List<TopicWord> createTopicWordList(final AMQShortString bindingKey)
+ {
+ AMQShortStringTokenizer tokens = bindingKey.tokenize(TOPIC_DELIMITER);
+ TopicWord previousWord = null;
+
+ List<TopicWord> wordList = new ArrayList<TopicWord>();
+
+ while(tokens.hasMoreTokens())
+ {
+ TopicWord nextWord = _dictionary.getOrCreateWord(tokens.nextToken());
+ if(previousWord == TopicWord.WILDCARD_WORD)
+ {
+
+ if(nextWord == TopicWord.WILDCARD_WORD)
+ {
+ // consecutive wildcards can be merged
+ // i.e. subsequent wildcards can be discarded
+ continue;
+ }
+ else if(nextWord == TopicWord.ANY_WORD)
+ {
+ // wildcard and anyword can be reordered to always put anyword first
+ wordList.set(wordList.size()-1,TopicWord.ANY_WORD);
+ nextWord = TopicWord.WILDCARD_WORD;
+ }
+ }
+ wordList.add(nextWord);
+ previousWord = nextWord;
+
+ }
+ return wordList;
+ }
+
+
+ public static void main(String[] args)
+ {
+ printMatches("a.#.b.#","a.b.b.b.b.b.b.b.c");
+
+
+ printMatches("","");
+ printMatches("a","a");
+ printMatches("a","");
+ printMatches("","a");
+ printMatches("a.b","a.b");
+ printMatches("a","a.b");
+ printMatches("a.b","a");
+ printMatches("*","a");
+ printMatches("*.b","a.b");
+ printMatches("*.*","a.b");
+ printMatches("a.*","a.b");
+ printMatches("a.*.#","a.b");
+ printMatches("a.#.b","a.b");
+
+ printMatches("#.b","a");
+ printMatches("#.b","a.b");
+ printMatches("#.a.b","a.b");
+
+
+ printMatches("#","");
+ printMatches("#","a");
+ printMatches("#","a.b");
+ printMatches("#.#","a.b");
+ printMatches("#.*","a.b");
+
+ printMatches("#.a.b","a.b");
+ printMatches("a.b.#","a.b");
+ printMatches("a.#","a.b");
+ printMatches("#.*.#","a.b");
+ printMatches("#.*.b.#","a.b");
+ printMatches("#.a.*.#","a.b");
+ printMatches("#.a.#.b.#","a.b");
+ printMatches("#.*.#.*.#","a.b");
+ printMatches("*.#.*.#","a.b");
+ printMatches("#.*.#.*","a.b");
+
+
+ printMatches(new String[]{"a.#.b.#","a.*.#.b.#"},"a.b.b.b.b.b.b.b.c");
+
+
+ printMatches(new String[]{"a.b", "a.c"},"a.b");
+ printMatches(new String[]{"a.#", "a.c", "#.b"},"a.b");
+ printMatches(new String[]{"a.#", "a.c", "#.b", "#", "*.*"},"a.b");
+
+ printMatches(new String[]{"a.b.c.d.e.#", "a.b.c.d.#", "a.b.c.d.*", "a.b.c.#", "#.e", "a.*.c.d.e","#.c.*.#.*.*"},"a.b.c.d.e");
+ printMatches(new String[]{"a.b.c.d.e.#", "a.b.c.d.#", "a.b.c.d.*", "a.b.c.#", "#.e", "a.*.c.d.e","#.c.*.#.*.*"},"a.b.c.d.f.g");
+
+
+
+
+ }
+
+ private static void printMatches(final String[] bindingKeys, final String routingKey)
+ {
+ TopicMatcherDFAState sm = null;
+ Map<TopicMatcherResult, String> resultMap = new HashMap<TopicMatcherResult, String>();
+
+ TopicParser parser = new TopicParser();
+
+ for(int i = 0; i < bindingKeys.length; i++)
+ {
+ TopicMatcherResult r = new TopicMatcherResult();
+ resultMap.put(r, bindingKeys[i]);
+ AMQShortString bindingKeyShortString = new AMQShortString(bindingKeys[i]);
+
+ if(i==0)
+ {
+ sm = parser.createStateMachine(bindingKeyShortString, r);
+ }
+ else
+ {
+ sm = sm.mergeStateMachines(parser.createStateMachine(bindingKeyShortString, r));
+ }
+ }
+ AMQShortString routingKeyShortString = new AMQShortString(routingKey);
+
+ Collection<TopicMatcherResult> results = sm.parse(parser._dictionary, routingKeyShortString);
+ Collection<String> resultStrings = new ArrayList<String>();
+
+ for(TopicMatcherResult result : results)
+ {
+ resultStrings.add(resultMap.get(result));
+ }
+
+ final ArrayList<String> nonMatches = new ArrayList<String>(Arrays.asList(bindingKeys));
+ nonMatches.removeAll(resultStrings);
+ System.out.println("\""+routingKeyShortString+"\" matched with " + resultStrings + " DID NOT MATCH with " + nonMatches);
+
+
+ }
+
+ private static void printMatches(String bindingKey, String routingKey)
+ {
+ printMatches(new String[] { bindingKey }, routingKey);
+ }
+
+
+ private static boolean matches(String bindingKey, String routingKey)
+ {
+ AMQShortString bindingKeyShortString = new AMQShortString(bindingKey);
+ AMQShortString routingKeyShortString = new AMQShortString(routingKey);
+ TopicParser parser = new TopicParser();
+
+ final TopicMatcherResult result = new TopicMatcherResult();
+
+ TopicMatcherDFAState sm = parser.createStateMachine(bindingKeyShortString, result);
+ return !sm.parse(parser._dictionary,routingKeyShortString).isEmpty();
+
+ }
+
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,34 @@
+package org.apache.qpid.server.exchange.topic;
+
+import org.apache.qpid.framing.AMQShortString;
+
+import java.util.Map;
+import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public final class TopicWord
+{
+ public static final TopicWord ANY_WORD = new TopicWord();
+ public static final TopicWord WILDCARD_WORD = new TopicWord();
+
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,63 @@
+package org.apache.qpid.server.exchange.topic;
+
+import org.apache.qpid.framing.AMQShortString;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public class TopicWordDictionary
+{
+ private final ConcurrentHashMap<AMQShortString,TopicWord> _dictionary =
+ new ConcurrentHashMap<AMQShortString,TopicWord>();
+
+
+
+ public TopicWordDictionary()
+ {
+ _dictionary.put(new AMQShortString("*"), TopicWord.ANY_WORD);
+ _dictionary.put(new AMQShortString("#"), TopicWord.WILDCARD_WORD);
+ }
+
+
+
+
+ public TopicWord getOrCreateWord(AMQShortString name)
+ {
+ TopicWord word = _dictionary.putIfAbsent(name, new TopicWord());
+ if(word == null)
+ {
+ word = _dictionary.get(name);
+ }
+ return word;
+ }
+
+
+ public TopicWord getWord(AMQShortString name)
+ {
+ TopicWord word = _dictionary.get(name);
+ if(word == null)
+ {
+ word = TopicWord.ANY_WORD;
+ }
+ return word;
+ }
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,57 @@
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+package org.apache.qpid.server.flow;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Set;
+import java.util.HashSet;
+
+public abstract class AbstractFlowCreditManager implements FlowCreditManager
+{
+ protected final AtomicBoolean _suspended = new AtomicBoolean(false);
+ private final Set<FlowCreditManagerListener> _listeners = new HashSet<FlowCreditManagerListener>();
+
+ public final void addStateListener(FlowCreditManagerListener listener)
+ {
+ _listeners.add(listener);
+ }
+
+ public final boolean removeListener(FlowCreditManagerListener listener)
+ {
+ return _listeners.remove(listener);
+ }
+
+ private void notifyListeners(final boolean suspended)
+ {
+ for(FlowCreditManagerListener listener : _listeners)
+ {
+ listener.creditStateChanged(!suspended);
+ }
+ }
+
+ protected final void setSuspended(final boolean suspended)
+ {
+ if(_suspended.compareAndSet(!suspended, suspended))
+ {
+ notifyListeners(suspended);
+ }
+ }
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,77 @@
+package org.apache.qpid.server.flow;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Set;
+import java.util.HashSet;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public class BytesOnlyCreditManager extends AbstractFlowCreditManager
+{
+ private final AtomicLong _bytesCredit;
+
+ public BytesOnlyCreditManager(long initialCredit)
+ {
+ _bytesCredit = new AtomicLong(initialCredit);
+ }
+
+ public void addCredit(long messageCredit, long bytesCredit)
+ {
+ _bytesCredit.addAndGet(bytesCredit);
+ setSuspended(false);
+ }
+
+ public void removeAllCredit()
+ {
+ _bytesCredit.set(0L);
+ }
+
+ public boolean hasCredit()
+ {
+ return _bytesCredit.get() > 0L;
+ }
+
+ public boolean useCreditForMessage(AMQMessage msg)
+ {
+ final long msgSize = msg.getSize();
+ if(hasCredit())
+ {
+ if(_bytesCredit.addAndGet(-msgSize) >= 0)
+ {
+ return true;
+ }
+ else
+ {
+ _bytesCredit.addAndGet(msgSize);
+ setSuspended(true);
+ return false;
+ }
+ }
+ else
+ {
+ return false;
+ }
+
+ }
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,44 @@
+package org.apache.qpid.server.flow;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public interface FlowCreditManager
+{
+
+ public static interface FlowCreditManagerListener
+ {
+ void creditStateChanged(boolean hasCredit);
+ }
+
+ void addStateListener(FlowCreditManagerListener listener);
+
+ boolean removeListener(FlowCreditManagerListener listener);
+
+ public void addCredit(long messageCredit, long bytesCredit);
+
+ public void removeAllCredit();
+
+ public boolean hasCredit();
+
+ public boolean useCreditForMessage(AMQMessage msg);
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,44 @@
+package org.apache.qpid.server.flow;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public class LimitlessCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
+{
+ public void addCredit(long messageCredit, long bytesCredit)
+ {
+ }
+
+ public void removeAllCredit()
+ {
+ }
+
+ public boolean hasCredit()
+ {
+ return true;
+ }
+
+ public boolean useCreditForMessage(AMQMessage msg)
+ {
+ return true;
+ }
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,79 @@
+package org.apache.qpid.server.flow;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public class MessageAndBytesCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
+{
+ private long _messageCredit;
+ private long _bytesCredit;
+
+ MessageAndBytesCreditManager(final long messageCredit, final long bytesCredit)
+ {
+ _messageCredit = messageCredit;
+ _bytesCredit = bytesCredit;
+ }
+
+ public synchronized void addCredit(long messageCredit, long bytesCredit)
+ {
+ _messageCredit += messageCredit;
+ _bytesCredit += bytesCredit;
+ setSuspended(hasCredit());
+ }
+
+ public synchronized void removeAllCredit()
+ {
+ _messageCredit = 0L;
+ _bytesCredit = 0L;
+ setSuspended(true);
+ }
+
+ public synchronized boolean hasCredit()
+ {
+ return (_messageCredit > 0L) && ( _bytesCredit > 0L );
+ }
+
+ public synchronized boolean useCreditForMessage(AMQMessage msg)
+ {
+ if(_messageCredit == 0L)
+ {
+ setSuspended(true);
+ return false;
+ }
+ else
+ {
+ final long msgSize = msg.getSize();
+ if(msgSize > _bytesCredit)
+ {
+ setSuspended(true);
+ return false;
+ }
+ _messageCredit--;
+ _bytesCredit -= msgSize;
+ setSuspended(false);
+ return true;
+ }
+
+ }
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,76 @@
+package org.apache.qpid.server.flow;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public class MessageOnlyCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
+{
+ private final AtomicLong _messageCredit;
+
+ MessageOnlyCreditManager(final long initialCredit)
+ {
+ _messageCredit = new AtomicLong(initialCredit);
+ }
+
+ public void addCredit(long messageCredit, long bytesCredit)
+ {
+ setSuspended(false);
+ _messageCredit.addAndGet(messageCredit);
+ }
+
+ public void removeAllCredit()
+ {
+ setSuspended(true);
+ _messageCredit.set(0L);
+ }
+
+ public boolean hasCredit()
+ {
+ return _messageCredit.get() > 0L;
+ }
+
+ public boolean useCreditForMessage(AMQMessage msg)
+ {
+ if(hasCredit())
+ {
+ if(_messageCredit.addAndGet(-1L) >= 0)
+ {
+ setSuspended(false);
+ return true;
+ }
+ else
+ {
+ _messageCredit.addAndGet(1L);
+ setSuspended(true);
+ return false;
+ }
+ }
+ else
+ {
+ setSuspended(true);
+ return false;
+ }
+
+ }
+}
Added: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java?rev=650148&view=auto
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java (added)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java Mon Apr 21 07:03:35 2008
@@ -0,0 +1,177 @@
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+package org.apache.qpid.server.flow;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+public class Pre0_10CreditManager extends AbstractFlowCreditManager implements FlowCreditManager
+{
+
+ private volatile long _bytesCreditLimit;
+ private volatile long _messageCreditLimit;
+
+ private volatile long _bytesCredit;
+ private volatile long _messageCredit;
+
+ public Pre0_10CreditManager(long bytesCreditLimit, long messageCreditLimit)
+ {
+ _bytesCreditLimit = bytesCreditLimit;
+ _messageCreditLimit = messageCreditLimit;
+ _bytesCredit = bytesCreditLimit;
+ _messageCredit = messageCreditLimit;
+ }
+
+
+ public synchronized void setCreditLimits(final long bytesCreditLimit, final long messageCreditLimit)
+ {
+ long bytesCreditChange = bytesCreditLimit - _bytesCreditLimit;
+ long messageCreditChange = messageCreditLimit - _messageCreditLimit;
+
+
+
+ if(bytesCreditChange != 0L)
+ {
+ if(bytesCreditLimit == 0L)
+ {
+ _bytesCredit = 0;
+ }
+ else
+ {
+ _bytesCredit += bytesCreditChange;
+ }
+ }
+
+
+ if(messageCreditChange != 0L)
+ {
+ if(messageCreditLimit == 0L)
+ {
+ _messageCredit = 0;
+ }
+ else
+ {
+ _messageCredit += messageCreditChange;
+ }
+ }
+
+
+ _bytesCreditLimit = bytesCreditLimit;
+ _messageCreditLimit = messageCreditLimit;
+
+ setSuspended(!hasCredit());
+
+ }
+
+
+ public synchronized void addCredit(final long messageCredit, final long bytesCredit)
+ {
+ final long messageCreditLimit = _messageCreditLimit;
+ if(messageCreditLimit != 0L)
+ {
+ long newCredit = _messageCredit + messageCredit;
+ _messageCredit = newCredit > messageCreditLimit ? messageCreditLimit : newCredit;
+ }
+
+
+ final long bytesCreditLimit = _bytesCreditLimit;
+ if(bytesCreditLimit != 0L)
+ {
+ long newCredit = _bytesCredit + bytesCredit;
+ _bytesCredit = newCredit > bytesCreditLimit ? bytesCreditLimit : newCredit;
+ }
+
+ setSuspended(!hasCredit());
+
+ }
+
+ public synchronized void removeAllCredit()
+ {
+ _bytesCredit = 0L;
+ _messageCredit = 0L;
+ setSuspended(!hasCredit());
+ }
+
+ public synchronized boolean hasCredit()
+ {
+ return (_bytesCreditLimit == 0L || _bytesCredit > 0)
+ && (_messageCreditLimit == 0L || _messageCredit > 0);
+ }
+
+ public synchronized boolean useCreditForMessage(final AMQMessage msg)
+ {
+ if(_messageCreditLimit != 0L)
+ {
+ if(_messageCredit != 0L)
+ {
+ if(_bytesCreditLimit == 0L)
+ {
+ _messageCredit--;
+
+ return true;
+ }
+ else
+ {
+ if((_bytesCredit >= msg.getSize()) || (_bytesCredit == _bytesCreditLimit))
+ {
+ _messageCredit--;
+ _bytesCredit -= msg.getSize();
+
+ return true;
+ }
+ else
+ {
+ setSuspended(true);
+ return false;
+ }
+ }
+ }
+ else
+ {
+ setSuspended(true);
+ return false;
+ }
+ }
+ else
+ {
+ if(_bytesCreditLimit == 0L)
+ {
+
+ return true;
+ }
+ else
+ {
+ if((_bytesCredit >= msg.getSize()) || (_bytesCredit == _bytesCreditLimit))
+ {
+ _bytesCredit -= msg.getSize();
+
+ return true;
+ }
+ else
+ {
+ setSuspended(true);
+ return false;
+ }
+ }
+
+ }
+
+ }
+}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java Mon Apr 21 07:03:35 2008
@@ -4,10 +4,6 @@
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.AMQException;
/**
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java Mon Apr 21 07:03:35 2008
@@ -21,11 +21,9 @@
package org.apache.qpid.server.handler;
import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.BasicCancelBody;
import org.apache.qpid.framing.BasicCancelOkBody;
import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.state.AMQStateManager;
@@ -65,7 +63,7 @@
" nowait:" + body.getNowait());
}
- channel.unsubscribeConsumer(session, body.getConsumerTag());
+ channel.unsubscribeConsumer(body.getConsumerTag());
if (!body.getNowait())
{
MethodRegistry methodRegistry = session.getMethodRegistry();
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java Mon Apr 21 07:03:35 2008
@@ -111,7 +111,7 @@
try
{
- AMQShortString consumerTag = channel.subscribeToQueue(consumerTagName, queue, session, !body.getNoAck(),
+ AMQShortString consumerTag = channel.subscribeToQueue(consumerTagName, queue, !body.getNoAck(),
body.getArguments(), body.getNoLocal(), body.getExclusive());
if (!body.getNowait())
{
@@ -121,8 +121,7 @@
}
- //now allow queue to start async processing of any backlog of messages
- queue.deliverAsync();
+
}
catch (org.apache.qpid.AMQInvalidArgumentException ise)
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java Mon Apr 21 07:03:35 2008
@@ -29,6 +29,7 @@
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueueImpl;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.state.AMQStateManager;
@@ -65,7 +66,6 @@
else
{
AMQQueue queue = body.getQueue() == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.getQueue());
-
if (queue == null)
{
_log.info("No queue for '" + body.getQueue() + "'");
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java Mon Apr 21 07:03:35 2008
@@ -91,7 +91,7 @@
MessagePublishInfo info = session.getMethodRegistry().getProtocolVersionMethodConverter().convertToInfo(body);
info.setExchange(exchange);
- channel.setPublishFrame(info, session, e);
+ channel.setPublishFrame(info, e);
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java Mon Apr 21 07:03:35 2008
@@ -22,10 +22,8 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.BasicQosBody;
-import org.apache.qpid.framing.BasicQosOkBody;
import org.apache.qpid.framing.MethodRegistry;
import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
@@ -49,8 +47,8 @@
throw body.getChannelNotFoundException(channelId);
}
- channel.setPrefetchCount(body.getPrefetchCount());
- channel.setPrefetchSize(body.getPrefetchSize());
+ channel.setCredit(body.getPrefetchSize(), body.getPrefetchCount());
+
MethodRegistry methodRegistry = session.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createBasicQosOkBody();
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java Mon Apr 21 07:03:35 2008
@@ -22,9 +22,8 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.BasicRejectBody;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.ack.UnacknowledgedMessage;
+import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
@@ -49,16 +48,6 @@
{
AMQProtocolSession session = stateManager.getProtocolSession();
-
-
-// if (_logger.isDebugEnabled())
-// {
-// _logger.debug("Rejecting:" + evt.getMethod().deliveryTag +
-// ": Requeue:" + evt.getMethod().requeue +
-//// ": Resend:" + evt.getMethod().resend +
-// " on channel:" + channelId);
-// }
-
AMQChannel channel = session.getChannel(channelId);
if (channel == null)
@@ -76,7 +65,7 @@
long deliveryTag = body.getDeliveryTag();
- UnacknowledgedMessage message = channel.getUnacknowledgedMessageMap().get(deliveryTag);
+ QueueEntry message = channel.getUnacknowledgedMessageMap().get(deliveryTag);
if (message == null)
{
@@ -85,11 +74,16 @@
}
else
{
- if (message.isQueueDeleted() || message.getQueue().isDeleted())
+ if (message.isQueueDeleted())
{
_logger.warn("Message's Queue as already been purged, unable to Reject. " +
"Dropping message should use Dead Letter Queue");
- //sendtoDeadLetterQueue(msg)
+ message = channel.getUnacknowledgedMessageMap().remove(deliveryTag);
+ if(message != null)
+ {
+ message.discard(channel.getStoreContext());
+ }
+ //sendtoDeadLetterQueue(msg)
return;
}
@@ -111,7 +105,7 @@
// If we haven't requested message to be resent to this consumer then reject it from ever getting it.
//if (!evt.getMethod().resend)
{
- message.entry.reject();
+ message.reject();
}
if (body.getRequeue())
@@ -121,6 +115,7 @@
else
{
_logger.warn("Dropping message as requeue not required and there is no dead letter queue");
+ message = channel.getUnacknowledgedMessageMap().remove(deliveryTag);
//sendtoDeadLetterQueue(AMQMessage message)
// message.queue = channel.getDefaultDeadLetterQueue();
// channel.requeue(deliveryTag);
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java Mon Apr 21 07:03:35 2008
@@ -22,12 +22,10 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.*;
-import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java Mon Apr 21 07:03:35 2008
@@ -112,7 +112,7 @@
if (!exch.isBound(routingKey, body.getArguments(), queue))
{
- queue.bind(routingKey, body.getArguments(), exch);
+ queue.bind(exch, routingKey, body.getArguments());
}
}
catch (AMQInvalidRoutingKeyException rke)
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java Mon Apr 21 07:03:35 2008
@@ -34,9 +34,10 @@
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.security.access.Permission;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.store.MessageStore;
@@ -123,7 +124,7 @@
{
Exchange defaultExchange = exchangeRegistry.getDefaultExchange();
- queue.bind(queueName, null, defaultExchange);
+ queue.bind(defaultExchange, queueName, null);
_logger.info("Queue " + queueName + " bound to default exchange(" + defaultExchange.getName() + ")");
}
}
@@ -173,7 +174,7 @@
{
final QueueRegistry registry = virtualHost.getQueueRegistry();
AMQShortString owner = body.getExclusive() ? session.getContextKey() : null;
- final AMQQueue queue = new AMQQueue(queueName, body.getDurable(), owner, body.getAutoDelete(), virtualHost);
+ final AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(queueName, body.getDurable(), owner, body.getAutoDelete(), virtualHost);
if (body.getExclusive() && !body.getDurable())
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java Mon Apr 21 07:03:35 2008
@@ -24,11 +24,10 @@
import org.apache.qpid.framing.QueueDeleteBody;
import org.apache.qpid.framing.QueueDeleteOkBody;
import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.store.MessageStore;
@@ -104,11 +103,12 @@
}
else
{
-
+
//Perform ACLs
virtualHost.getAccessManager().authorise(session, Permission.DELETE, body, queue);
- int purged = queue.delete(body.getIfUnused(), body.getIfEmpty());
+ int purged = queue.delete();
+
if (queue.isDurable())
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java Mon Apr 21 07:03:35 2008
@@ -23,14 +23,12 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.QueuePurgeBody;
-import org.apache.qpid.framing.QueuePurgeOkBody;
import org.apache.qpid.framing.MethodRegistry;
import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java Mon Apr 21 07:03:35 2008
@@ -42,7 +42,7 @@
final AMQQueue queue;
- final AMQShortString routingKey;
+ final AMQShortString routingKey;
if (body.getQueue() == null)
{
@@ -84,7 +84,7 @@
try
{
- queue.unBind(routingKey, body.getArguments(), exch);
+ queue.unBind(exch, routingKey, body.getArguments());
}
catch (AMQInvalidRoutingKeyException rke)
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java Mon Apr 21 07:03:35 2008
@@ -23,10 +23,8 @@
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.TxCommitBody;
-import org.apache.qpid.framing.TxCommitOkBody;
import org.apache.qpid.framing.MethodRegistry;
import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.state.AMQStateManager;
@@ -70,7 +68,7 @@
AMQMethodBody responseBody = methodRegistry.createTxCommitOkBody();
session.writeFrame(responseBody.generateFrame(channelId));
- channel.processReturns(session);
+ channel.processReturns();
}
catch (AMQException e)
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java Mon Apr 21 07:03:35 2008
@@ -78,9 +78,9 @@
final AMQMessageHandle messageHandle = message.getMessageHandle();
final StoreContext storeContext = message.getStoreContext();
- final Long messageId = message.getMessageId();
- final int bodyCount = messageHandle.getBodyCount(storeContext,messageId);
+
+ final int bodyCount = messageHandle.getBodyCount(storeContext);
if(bodyCount == 0)
{
@@ -97,7 +97,7 @@
// Optimise the case where we have a single content body. In that case we create a composite block
// so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
//
- ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
+ ContentChunk cb = messageHandle.getContentChunk(storeContext, 0);
AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
AMQDataBlock[] blocks = new AMQDataBlock[]{deliver, contentHeader, firstContentBody};
@@ -109,7 +109,7 @@
//
for(int i = 1; i < bodyCount; i++)
{
- cb = messageHandle.getContentChunk(storeContext,messageId, i);
+ cb = messageHandle.getContentChunk(storeContext, i);
writeFrame(new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
}
@@ -125,7 +125,6 @@
final AMQMessageHandle messageHandle = message.getMessageHandle();
final StoreContext storeContext = message.getStoreContext();
- final long messageId = message.getMessageId();
AMQDataBlock deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);
@@ -133,7 +132,7 @@
AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,
message.getContentHeaderBody());
- final int bodyCount = messageHandle.getBodyCount(storeContext,messageId);
+ final int bodyCount = messageHandle.getBodyCount(storeContext);
if(bodyCount == 0)
{
SmallCompositeAMQDataBlock compositeBlock = new SmallCompositeAMQDataBlock(deliver,
@@ -148,7 +147,7 @@
// Optimise the case where we have a single content body. In that case we create a composite block
// so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
//
- ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
+ ContentChunk cb = messageHandle.getContentChunk(storeContext, 0);
AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
AMQDataBlock[] blocks = new AMQDataBlock[]{deliver, contentHeader, firstContentBody};
@@ -160,7 +159,7 @@
//
for(int i = 1; i < bodyCount; i++)
{
- cb = messageHandle.getContentChunk(storeContext, messageId, i);
+ cb = messageHandle.getContentChunk(storeContext, i);
writeFrame(new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java Mon Apr 21 07:03:35 2008
@@ -56,9 +56,9 @@
final AMQMessageHandle messageHandle = message.getMessageHandle();
final StoreContext storeContext = message.getStoreContext();
- final Long messageId = message.getMessageId();
- final int bodyCount = messageHandle.getBodyCount(storeContext,messageId);
+
+ final int bodyCount = messageHandle.getBodyCount(storeContext);
if(bodyCount == 0)
{
@@ -75,7 +75,7 @@
// Optimise the case where we have a single content body. In that case we create a composite block
// so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
//
- ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
+ ContentChunk cb = messageHandle.getContentChunk(storeContext, 0);
AMQBody firstContentBody = PROTOCOL_METHOD_CONVERTER.convertToBody(cb);
@@ -87,14 +87,13 @@
//
for(int i = 1; i < bodyCount; i++)
{
- cb = messageHandle.getContentChunk(storeContext,messageId, i);
+ cb = messageHandle.getContentChunk(storeContext, i);
writeFrame(new AMQFrame(channelId, PROTOCOL_METHOD_CONVERTER.convertToBody(cb)));
}
}
-
-
+
}
private AMQDataBlock createContentHeaderBlock(final int channelId, final ContentHeaderBody contentHeaderBody)
@@ -111,14 +110,13 @@
final AMQMessageHandle messageHandle = message.getMessageHandle();
final StoreContext storeContext = message.getStoreContext();
- final long messageId = message.getMessageId();
AMQFrame deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);
AMQDataBlock contentHeader = createContentHeaderBlock(channelId, message.getContentHeaderBody());
- final int bodyCount = messageHandle.getBodyCount(storeContext,messageId);
+ final int bodyCount = messageHandle.getBodyCount(storeContext);
if(bodyCount == 0)
{
SmallCompositeAMQDataBlock compositeBlock = new SmallCompositeAMQDataBlock(deliver,
@@ -133,7 +131,7 @@
// Optimise the case where we have a single content body. In that case we create a composite block
// so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
//
- ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
+ ContentChunk cb = messageHandle.getContentChunk(storeContext, 0);
AMQDataBlock firstContentBody = new AMQFrame(channelId, PROTOCOL_METHOD_CONVERTER.convertToBody(cb));
AMQDataBlock[] blocks = new AMQDataBlock[]{deliver, contentHeader, firstContentBody};
@@ -145,7 +143,7 @@
//
for(int i = 1; i < bodyCount; i++)
{
- cb = messageHandle.getContentChunk(storeContext, messageId, i);
+ cb = messageHandle.getContentChunk(storeContext, i);
writeFrame(new AMQFrame(channelId, PROTOCOL_METHOD_CONVERTER.convertToBody(cb)));
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java Mon Apr 21 07:03:35 2008
@@ -198,7 +198,7 @@
}
private void frameReceived(AMQFrame frame) throws AMQException
- {
+ {
int channelId = frame.getChannel();
AMQBody body = frame.getBodyFrame();
@@ -373,7 +373,7 @@
AMQChannel channel = getAndAssertChannel(channelId);
- channel.publishContentHeader(body, this);
+ channel.publishContentHeader(body);
}
@@ -381,7 +381,7 @@
{
AMQChannel channel = getAndAssertChannel(channelId);
- channel.publishContentBody(body, this);
+ channel.publishContentBody(body);
}
public void heartbeatBodyReceived(int channelId, HeartbeatBody body)
@@ -536,7 +536,7 @@
{
try
{
- channel.close(this);
+ channel.close();
markChannelAwaitingCloseOk(channelId);
}
finally
@@ -602,7 +602,7 @@
{
for (AMQChannel channel : _channelMap.values())
{
- channel.close(this);
+ channel.close();
}
_channelMap.clear();
@@ -633,7 +633,7 @@
public String toString()
{
- return "AMQProtocolSession(" + _minaProtocolSession.getRemoteAddress() + ")";
+ return _minaProtocolSession.getRemoteAddress() + "("+(getAuthorizedID() == null ? "?" : getAuthorizedID().getName()+")");
}
public String dump()
@@ -739,7 +739,7 @@
public Object getClientIdentifier()
{
- return _minaProtocolSession.getRemoteAddress();
+ return (_minaProtocolSession != null) ? _minaProtocolSession.getRemoteAddress() : null;
}
public VirtualHost getVirtualHost()