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.servicemix.eip.support.resequence;
018    
019    import javax.jbi.messaging.MessageExchange;
020    
021    /**
022     * Compares {@link MessageExchange} sequence elements based on sequence numbers
023     * defined by their in-{@link NormalizedMessage}s. This comparator works on
024     * sequence numbers of type {@link Long}. Sequence numbers must be stored as
025     * {@link NormalizedMessage} properties. The property name under which the
026     * sequence number is stored is configured via this comparator's
027     * <code>sequenceNumberKey</code> property.
028     * 
029     * @author Martin Krasser
030     * 
031     * @org.apache.xbean.XBean element="default-comparator"
032     */
033    public class DefaultComparator implements SequenceElementComparator<MessageExchange> {
034    
035        public static final String SEQUENCE_NUMBER_KEY = "org.apache.servicemix.eip.sequence.number";
036        
037        private static final String IN = "in";
038        
039        private String sequenceNumberKey;
040        
041        private boolean sequenceNumberAsString;
042        
043        public DefaultComparator() {
044            sequenceNumberKey = SEQUENCE_NUMBER_KEY;
045            sequenceNumberAsString = false;
046        }
047        
048        public String getSequenceNumberKey() {
049            return sequenceNumberKey;
050        }
051    
052        public void setSequenceNumberKey(String sequenceNumberPropertyName) {
053            this.sequenceNumberKey = sequenceNumberPropertyName;
054        }
055    
056        public boolean isSequenceNumberAsString() {
057            return sequenceNumberAsString;
058        }
059    
060        public void setSequenceNumberAsString(boolean sequenceNumberAsString) {
061            this.sequenceNumberAsString = sequenceNumberAsString;
062        }
063    
064        public boolean predecessor(MessageExchange o1, MessageExchange o2) {
065            long n1 = getSequenceNumber(o1).longValue();
066            long n2 = getSequenceNumber(o2).longValue();
067            return n1 == (n2 - 1L);
068        }
069    
070        public boolean successor(MessageExchange o1, MessageExchange o2) {
071            long n1 = getSequenceNumber(o1).longValue();
072            long n2 = getSequenceNumber(o2).longValue();
073            return n2 == (n1 - 1L);
074        }
075    
076        public int compare(MessageExchange o1, MessageExchange o2) {
077            Long n1 = getSequenceNumber(o1);
078            Long n2 = getSequenceNumber(o2);
079            return n1.compareTo(n2);
080        }
081    
082        private Long getSequenceNumber(MessageExchange exchange) {
083            Object number = exchange.getMessage(IN).getProperty(sequenceNumberKey);
084            if (sequenceNumberAsString) {
085                return new Long((String)number);
086            } else {
087                return (Long)number;
088            }
089        }
090        
091    }