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.common.endpoints;
018
019 import java.util.Date;
020 import java.util.concurrent.atomic.AtomicBoolean;
021
022 import javax.jbi.JBIException;
023 import javax.jbi.servicedesc.ServiceEndpoint;
024 import javax.xml.namespace.QName;
025
026 import org.apache.servicemix.common.DefaultComponent;
027 import org.apache.servicemix.common.ServiceUnit;
028 import org.apache.servicemix.common.scheduler.ScheduleIterator;
029 import org.apache.servicemix.common.scheduler.Scheduler;
030 import org.apache.servicemix.common.scheduler.SchedulerTask;
031 import org.apache.servicemix.executors.Executor;
032
033 /**
034 * An implementation inheritence class for an endpoint which polls some resource at periodic intervals to decide if
035 * there is an event to process.
036 *
037 * @version $Revision: 464478 $
038 */
039 public abstract class PollingEndpoint extends ConsumerEndpoint {
040
041 private Executor executor;
042 private Scheduler scheduler;
043 private Date firstTime;
044 private long period = 5000;
045 private long delay;
046 private SchedulerTask schedulerTask;
047 private ScheduleIterator scheduleIterator;
048 private boolean started;
049 private boolean scheduleExecutedFlag;
050 private boolean concurrentPolling;
051 private AtomicBoolean pollActive = new AtomicBoolean(false);
052
053 public PollingEndpoint() {
054 }
055
056 public PollingEndpoint(ServiceUnit serviceUnit, QName service, String endpoint) {
057 super(serviceUnit, service, endpoint);
058 }
059
060 public PollingEndpoint(DefaultComponent component, ServiceEndpoint endpoint) {
061 super(component.getServiceUnit(), endpoint.getServiceName(), endpoint.getEndpointName());
062 }
063
064 /**
065 * Polls the underlying resource to see if some event is required
066 *
067 * @throws JBIException
068 */
069 public abstract void poll() throws Exception;
070
071
072 // Properties
073 // -------------------------------------------------------------------------
074 public Executor getExecutor() {
075 return executor;
076 }
077
078 public long getDelay() {
079 return delay;
080 }
081
082 /**
083 * Sets the amount of time the endpoint waits before making the first poll.
084 *
085 * @param delay a long specifying the number of milliseconds to wait
086 * @org.apache.xbean.Property description="the number of milliseconds to wait before the first poll"
087 */
088 public void setDelay(long delay) {
089 this.delay = delay;
090 }
091
092 public Date getFirstTime() {
093 return firstTime;
094 }
095
096 /**
097 * Sets the date on which the first poll will be executed. If a delay is
098 * also set using <code>setDelay</code>, the delay interval will be added
099 * after the date specified,
100 *
101 * @param firstTime a <code>Date</code> specifying when to make the
102 * first polling attempt
103 * @org.apache.xbean.Property description="the date of the first polling attempt. The date is specified using the <code>YYYY-MM-DD</code> format. The <code>delay</code> value is added after the date."
104 */
105 public void setFirstTime(Date firstTime) {
106 this.firstTime = firstTime;
107 }
108
109 public long getPeriod() {
110 return period;
111 }
112
113 /**
114 * returns if more than one poll can be active at a time
115 *
116 * @return Returns the concurrentPolling flag.
117 * @org.apache.xbean.Property description="returns if more than one poll can be active at a time"
118 */
119 public boolean isConcurrentPolling() {
120 return this.concurrentPolling;
121 }
122
123 /**
124 * sets if more than one poll can be active at a time (true means yes)
125 *
126 * @param concurrentPolling The concurrentPolling to set.
127 * @org.apache.xbean.Property description="sets if more than one poll can be active at a time (true means yes)"
128 */
129 public void setConcurrentPolling(boolean concurrentPolling) {
130 this.concurrentPolling = concurrentPolling;
131 }
132
133 /**
134 * Sets the number of milliseconds between polling attempts.
135 *
136 * @param period a long specifying the gap between polling attempts
137 * @org.apache.xbean.Property description="the number of milliseconds between polling attempts"
138 */
139 public void setPeriod(long period) {
140 this.period = period;
141 }
142
143 public Scheduler getScheduler() {
144 return scheduler;
145 }
146
147 public void setScheduler(Scheduler scheduler) {
148 this.scheduler = scheduler;
149 }
150
151 public synchronized void start() throws Exception {
152 if (!started) {
153 started = true;
154
155 if (scheduler == null) {
156 scheduler = new Scheduler(true);
157 }
158 if (scheduleIterator == null) {
159 scheduleIterator = new PollingEndpoint.PollScheduleIterator();
160 }
161
162 if (executor == null) {
163 executor = getServiceUnit().getComponent().getExecutor();
164 }
165 if (schedulerTask != null) {
166 schedulerTask.cancel();
167 }
168 schedulerTask = new PollingEndpoint.PollSchedulerTask();
169 this.scheduler.schedule(schedulerTask, scheduleIterator);
170 }
171 super.start();
172 }
173
174 public synchronized void stop() throws Exception {
175 if (schedulerTask != null) {
176 schedulerTask.cancel();
177 schedulerTask = null;
178 }
179 scheduleExecutedFlag = false;
180 started = false;
181 scheduler.cancel();
182 scheduler = null;
183 scheduleIterator = null;
184 executor = null;
185 super.stop();
186 }
187
188 // Implementation methods
189 // -------------------------------------------------------------------------
190
191 private class PollSchedulerTask extends SchedulerTask {
192 public void run() {
193 try {
194 if (!isConcurrentPolling() && pollActive.get()) {
195 // do not disturb the active poll cycle
196 return;
197 }
198
199 // lets run the work inside the JCA worker pools to ensure
200 // the threads are setup correctly when we actually do stuff
201 getExecutor().execute(new Runnable() {
202 public void run() {
203 try {
204 // set busy marker
205 pollActive.set(true);
206
207 // call poll implementation
208 poll();
209
210 // release busy marker
211 pollActive.set(false);
212 }
213 catch (Exception e) {
214 handlePollException(e);
215 pollActive.set(false);
216 }
217 }
218 });
219 }
220 catch (Throwable e) {
221 logger.error("Failed to schedule work: " + e, e);
222 pollActive.set(false);
223 }
224 }
225 }
226
227 protected void handlePollException(Exception e) {
228 logger.error("Caught exception while polling: " + e, e);
229 }
230
231
232 private class PollScheduleIterator implements ScheduleIterator {
233 public Date nextExecution() {
234 long nextTime = System.currentTimeMillis();
235 if (scheduleExecutedFlag) {
236 nextTime += period;
237 }
238 else {
239 if (firstTime != null) {
240 nextTime = firstTime.getTime();
241 }
242 nextTime += delay;
243 scheduleExecutedFlag = true;
244 }
245 return (started) ? new Date(nextTime) : null;
246 }
247 }
248 }