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.camel.component.dataset;
018    
019    import java.util.HashMap;
020    import java.util.Map;
021    
022    import org.apache.camel.Exchange;
023    import org.apache.camel.Message;
024    import org.apache.camel.Processor;
025    import org.apache.camel.util.ExchangeHelper;
026    
027    /**
028     * Base class for DataSet
029     *
030     * @version $Revision: 13710 $
031     */
032    public abstract class DataSetSupport implements DataSet {
033        private Map<String, Object> defaultHeaders;
034        private Processor outputTransformer;
035        private long size = 10;
036        private long reportCount = -1;
037    
038        public DataSetSupport() {
039        }
040    
041        public DataSetSupport(int size) {
042            setSize(size);
043        }
044    
045        public void populateMessage(Exchange exchange, long messageIndex) throws Exception {
046            Message in = exchange.getIn();
047            in.setBody(createMessageBody(messageIndex));
048            in.setHeaders(getDefaultHeaders());
049            applyHeaders(exchange, messageIndex);
050    
051            if (outputTransformer != null) {
052                outputTransformer.process(exchange);
053            }
054        }
055    
056        public void assertMessageExpected(DataSetEndpoint dataSetEndpoint, Exchange expected, Exchange actual, long index) throws Exception {
057            Object expectedBody = expected.getIn().getBody();
058            Object actualBody = actual.getIn().getBody();
059            if (expectedBody != null) {
060                // lets coerce to the correct type
061                actualBody = ExchangeHelper.getMandatoryInBody(actual, expectedBody.getClass());
062            }
063            DataSetEndpoint.assertEquals("message body", expectedBody, actualBody, actual);
064        }
065    
066        // Properties
067        //-------------------------------------------------------------------------
068    
069        public long getSize() {
070            return size;
071        }
072    
073        public void setSize(long size) {
074            this.size = size;
075        }
076    
077        public long getReportCount() {
078            if (reportCount <= 0) {
079                reportCount = getSize() / 5;
080            }
081            // report cannot be 0 then default to the size
082            if (reportCount == 0) {
083                reportCount = getSize();
084            }
085            return reportCount;
086        }
087    
088        /**
089         * Sets the number of messages in a group on which we will report that messages have been received.
090         */
091        public void setReportCount(long reportCount) {
092            this.reportCount = reportCount;
093        }
094    
095        public Map<String, Object> getDefaultHeaders() {
096            if (defaultHeaders == null) {
097                defaultHeaders = new HashMap<String, Object>();
098                populateDefaultHeaders(defaultHeaders);
099            }
100            return defaultHeaders;
101        }
102    
103        public void setDefaultHeaders(Map<String, Object> defaultHeaders) {
104            this.defaultHeaders = defaultHeaders;
105        }
106    
107        public Processor getOutputTransformer() {
108            return outputTransformer;
109        }
110    
111        public void setOutputTransformer(Processor outputTransformer) {
112            this.outputTransformer = outputTransformer;
113        }
114    
115        // Implementation methods
116        //-------------------------------------------------------------------------
117    
118        protected abstract Object createMessageBody(long messageIndex);
119    
120        /**
121         * Allows derived classes to add some custom headers for a given message
122         */
123        protected void applyHeaders(Exchange exchange, long messageIndex) {
124        }
125    
126        /**
127         * Allows derived classes to customize a default set of properties
128         */
129        protected void populateDefaultHeaders(Map<String, Object> map) {
130        }
131    }