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.seda;
018    
019    import java.util.ArrayList;
020    import java.util.HashSet;
021    import java.util.List;
022    import java.util.Set;
023    import java.util.concurrent.BlockingQueue;
024    import java.util.concurrent.CopyOnWriteArraySet;
025    import java.util.concurrent.LinkedBlockingQueue;
026    
027    import org.apache.camel.Component;
028    import org.apache.camel.Consumer;
029    import org.apache.camel.Exchange;
030    import org.apache.camel.Processor;
031    import org.apache.camel.Producer;
032    import org.apache.camel.WaitForTaskToComplete;
033    import org.apache.camel.impl.DefaultEndpoint;
034    import org.apache.camel.spi.BrowsableEndpoint;
035    
036    /**
037     * An implementation of the <a
038     * href="http://camel.apache.org/queue.html">Queue components</a> for
039     * asynchronous SEDA exchanges on a {@link BlockingQueue} within a CamelContext
040     *
041     * @version $Revision: 780262 $
042     */
043    public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint {
044        private BlockingQueue<Exchange> queue;
045        private int size = 1000;
046        private int concurrentConsumers = 1;
047        private WaitForTaskToComplete waitForTaskToComplete = WaitForTaskToComplete.IfReplyExpected;
048        private Set<SedaProducer> producers = new CopyOnWriteArraySet<SedaProducer>();
049        private Set<SedaConsumer> consumers = new CopyOnWriteArraySet<SedaConsumer>();
050    
051        public SedaEndpoint() {
052        }
053    
054        public SedaEndpoint(String endpointUri, Component component, BlockingQueue<Exchange> queue) {
055            this(endpointUri, component, queue, 1);
056        }
057    
058        public SedaEndpoint(String endpointUri, Component component, BlockingQueue<Exchange> queue, int concurrentConsumers) {
059            super(endpointUri, component);
060            this.queue = queue;
061            this.concurrentConsumers = concurrentConsumers;
062        }
063    
064        public SedaEndpoint(String endpointUri, BlockingQueue<Exchange> queue) {
065            this(endpointUri, queue, 1);
066        }
067    
068        public SedaEndpoint(String endpointUri, BlockingQueue<Exchange> queue, int concurrentConsumers) {
069            super(endpointUri);
070            this.queue = queue;
071            this.concurrentConsumers = concurrentConsumers;
072        }
073        
074        public Producer createProducer() throws Exception {
075            return new SedaProducer(this, getQueue(), getWaitForTaskToComplete());
076        }
077    
078        public Consumer createConsumer(Processor processor) throws Exception {
079            return new SedaConsumer(this, processor);
080        }
081    
082        public synchronized BlockingQueue<Exchange> getQueue() {
083            if (queue == null) {
084                queue = new LinkedBlockingQueue<Exchange>(size);
085            }
086            return queue;
087        }
088        
089        public void setQueue(BlockingQueue<Exchange> queue) {
090            this.queue = queue;
091        }
092    
093        public int getSize() {
094            return size;
095        }
096    
097        public void setSize(int size) {
098            this.size = size;
099        }
100    
101        public void setConcurrentConsumers(int concurrentConsumers) {
102            this.concurrentConsumers = concurrentConsumers;
103        }
104        
105        public int getConcurrentConsumers() {
106            return concurrentConsumers;
107        }
108    
109        public WaitForTaskToComplete getWaitForTaskToComplete() {
110            return waitForTaskToComplete;
111        }
112    
113        public void setWaitForTaskToComplete(WaitForTaskToComplete waitForTaskToComplete) {
114            this.waitForTaskToComplete = waitForTaskToComplete;
115        }
116    
117        public boolean isSingleton() {
118            return true;
119        }
120    
121        /**
122         * Returns the current pending exchanges
123         */
124        public List<Exchange> getExchanges() {
125            return new ArrayList<Exchange>(getQueue());
126        }
127    
128        /**
129         * Returns the current active consumers on this endpoint
130         */
131        public Set<SedaConsumer> getConsumers() {
132            return new HashSet<SedaConsumer>(consumers);
133        }
134    
135        /**
136         * Returns the current active producers on this endpoint
137         */
138        public Set<SedaProducer> getProducers() {
139            return new HashSet<SedaProducer>(producers);
140        }
141        
142        void onStarted(SedaProducer producer) {
143            producers.add(producer);
144        }
145    
146        void onStopped(SedaProducer producer) {
147            producers.remove(producer);
148        }
149    
150        void onStarted(SedaConsumer consumer) {
151            consumers.add(consumer);
152        }
153    
154        void onStopped(SedaConsumer consumer) {
155            consumers.remove(consumer);
156        }
157    }