1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
/*
* $Id: AbstractPollingMessageReceiver.java 10961 2008-02-22 19:01:02Z dfeist $
* --------------------------------------------------------------------------------------
* Copyright (c) MuleSource, Inc. All rights reserved. http://www.mulesource.com
*
* The software in this package is published under the terms of the CPAL v1.0
* license, a copy of which has been included with this distribution in the
* LICENSE.txt file.
*/
package org.mule.transport;
import org.mule.api.MuleException;
import org.mule.api.endpoint.InboundEndpoint;
import org.mule.api.lifecycle.CreateException;
import org.mule.api.service.Service;
import org.mule.api.transport.Connector;
import org.mule.config.i18n.CoreMessages;
import org.mule.util.ObjectUtils;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
/**
* <code>AbstractPollingMessageReceiver</code> implements a base class for polling
* message receivers. The receiver provides a {@link #poll()} method that implementations
* must implement to execute their custom code. Note that the receiver will not poll if
* the associated connector is not started.
*/
public abstract class AbstractPollingMessageReceiver extends AbstractMessageReceiver
{
public static final long DEFAULT_POLL_FREQUENCY = 1000;
public static final TimeUnit DEFAULT_POLL_TIMEUNIT = TimeUnit.MILLISECONDS;
public static final long DEFAULT_STARTUP_DELAY = 1000;
private long frequency = DEFAULT_POLL_FREQUENCY;
private TimeUnit timeUnit = DEFAULT_POLL_TIMEUNIT;
// @GuardedBy(itself)
protected final List schedules = new LinkedList();
public AbstractPollingMessageReceiver(Connector connector,
Service service,
final InboundEndpoint endpoint) throws CreateException
{
super(connector, service, endpoint);
}
protected void doStart() throws MuleException
{
try
{
this.schedule();
}
catch (Exception ex)
{
this.stop();
throw new CreateException(CoreMessages.failedToScheduleWork(), ex, this);
}
}
protected void doStop() throws MuleException
{
this.unschedule();
}
/**
* This method registers this receiver for periodic polling ticks with the connectors
* scheduler. Subclasses can override this in case they want to handle their polling
* differently.
*
* @throws RejectedExecutionException
* @throws NullPointerException
* @throws IllegalArgumentException
* @see {@link ScheduledExecutorService#scheduleWithFixedDelay(Runnable, long, long, TimeUnit)}
*/
protected void schedule()
throws RejectedExecutionException, NullPointerException, IllegalArgumentException
{
synchronized (schedules)
{
// we use scheduleWithFixedDelay to prevent queue-up of tasks when
// polling takes longer than the specified frequency, e.g. when the
// polled database or network is slow or returns large amounts of
// data.
ScheduledFuture schedule = connector.getScheduler().scheduleWithFixedDelay(
new PollingReceiverWorkerSchedule(this.createWork()), DEFAULT_STARTUP_DELAY,
this.getFrequency(), this.getTimeUnit());
schedules.add(schedule);
if (logger.isDebugEnabled())
{
logger.debug(ObjectUtils.identityToShortString(this) + " scheduled "
+ ObjectUtils.identityToShortString(schedule) + " with " + frequency
+ " " + getTimeUnit() + " polling frequency");
}
}
}
/**
* This method cancels the schedules which were created in {@link #schedule()}.
*
* @see {@link Future#cancel(boolean)}
*/
protected void unschedule()
{
synchronized (schedules)
{
// cancel our schedules gently: do not interrupt when polling is in progress
for (Iterator i = schedules.iterator(); i.hasNext();)
{
ScheduledFuture schedule = (ScheduledFuture)i.next();
schedule.cancel(false);
i.remove();
if (logger.isDebugEnabled())
{
logger.debug(ObjectUtils.identityToShortString(this) + " cancelled polling schedule: "
+ ObjectUtils.identityToShortString(schedule));
}
}
}
}
protected PollingReceiverWorker createWork()
{
return new PollingReceiverWorker(this);
}
public long getFrequency()
{
return frequency;
}
// TODO a nifty thing would be on-the-fly adjustment (via JMX?) of the
// polling frequency by rescheduling without explicit stop()
public void setFrequency(long value)
{
if (value <= 0)
{
frequency = DEFAULT_POLL_FREQUENCY;
}
else
{
frequency = value;
}
}
public TimeUnit getTimeUnit()
{
return timeUnit;
}
public void setTimeUnit(TimeUnit timeUnit)
{
this.timeUnit = timeUnit;
}
public abstract void poll() throws Exception;
}
|
