KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > de > nava > informa > utils > poller > Poller


1 //
2 // Informa -- RSS Library for Java
3 // Copyright (c) 2002 by Niko Schmuck
4 //
5 // Niko Schmuck
6 // http://sourceforge.net/projects/informa
7 // mailto:niko_schmuck@users.sourceforge.net
8 //
9 // This library is free software.
10 //
11 // You may redistribute it and/or modify it under the terms of the GNU
12 // Lesser General Public License as published by the Free Software Foundation.
13 //
14 // Version 2.1 of the license should be included with this distribution in
15 // the file LICENSE. If the license is not included with this distribution,
16 // you may find a copy at the FSF web site at 'www.gnu.org' or 'www.fsf.org',
17 // or you may write to the Free Software Foundation, 675 Mass Ave, Cambridge,
18 // MA 02139 USA.
19 //
20 // This library is distributed in the hope that it will be useful,
21 // but WITHOUT ANY WARRANTY; without even the implied waranty of
22 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
23 // Lesser General Public License for more details.
24 //
25 // $Id: Poller.java,v 1.3 2004/08/24 18:04:30 spyromus Exp $
26 //
27
28 package de.nava.informa.utils.poller;
29
30 import de.nava.informa.core.ChannelIF;
31 import de.nava.informa.utils.toolkit.WorkerThread;
32 import de.nava.informa.utils.toolkit.WorkerThreadFactoryIF;
33 import de.nava.informa.utils.toolkit.WorkersManager;
34 import de.nava.informa.utils.toolkit.Scheduler;
35 import de.nava.informa.utils.toolkit.ChannelRecord;
36 import de.nava.informa.utils.toolkit.SchedulerCallbackIF;
37
38 /**
39  * Background poller of channels and groups of channels. Default poller period of channel is
40  * set to an hour.
41  * <p>
42  * Each Poller instance has its own scheduler to manage the schedule of updates and
43  * worker threads manager to manage processing of updates.</p>
44  * <p>
45  * Right after the instance of Poller is created you might be interested in adding some
46  * observers (<code>addObserver()</code>) to receive events on channels updates. Registered
47  * observers are added to the composite observer which is following Composite pattern to
48  * combine all of the listeners and present them as single instance convenient for operation.
49  * This instance is passed to all newly created worker threads. Later it will be used by
50  * worker threads to report about changes.</p>
51  * <p>
52  * <i>If you are interested in making changes directly in persistence storage then, please,
53  * look at <code>PersistenceObserver</code> class. It's simple implementation of persistence
54  * changer based on Observer events.</i>
55  * <p>
56  * You also may require to filter newly found items in channels being polled. Using Approvers
57  * you can receive events only when <i>wanted</i> items are detected. Each registered Approver
58  * votes for addition of newly detected item in channel. All of registered Approvers should
59  * vote for addition in order to send event about newly detected item to Observers.</p>
60  * <p>
61  * Finally, you need to register some channels to get benefits from Poller. Using appropriate
62  * methods you can register and unregister channels, request immediate updates and change
63  * polling period for all regitered channels at once.</p>
64  *
65  * @author Aleksey Gureev (spyromus@noizeramp.com)
66  */
67 public class Poller {
68   private static final long DEFAULT_POLL_PERIOD = 3600000; // millis (60 minutes)
69 private static final int DEFAULT_WORKER_THREADS = 5;
70
71   private WorkersManager workersManager;
72   private Scheduler scheduler;
73
74   private CompositeObserver compositeObserver;
75   private CompositeApprover compositeApprover;
76
77   private long globalPollPeriod = DEFAULT_POLL_PERIOD;
78
79   /**
80    * Creates poller with default number of worker threads.
81    */
82   public Poller() {
83     this(DEFAULT_WORKER_THREADS);
84   }
85
86   /**
87    * Creates memory-only poller.
88    *
89    * @param workerThreads desired number of worker threads.
90    */
91   public Poller(int workerThreads) {
92     // Create composite objects.
93 compositeObserver = new CompositeObserver();
94     compositeApprover = new CompositeApprover();
95
96     // Initialize workers manager.
97 workersManager = new WorkersManager(new PollerThreadFactory(), workerThreads);
98
99     // Initialize scheduler.
100 scheduler = new Scheduler(new SchedulerCallback());
101   }
102
103   /**
104    * Sets count on number of poller threads. This method will not kill all currently running
105    * extra threads. It will instruct extra threads to terminate once their job is over or
106    * create new threads if new count is bigger than previous.
107    *
108    * @param count new count on number of poller threads.
109    */
110   public final void setWorkerThreads(int count) {
111     workersManager.setWorkerThreads(count);
112   }
113
114   /**
115    * Register new channel for background poller.
116    *
117    * @param channel channel to register.
118    */
119   public final void registerChannel(ChannelIF channel) {
120     scheduler.schedule(channel, globalPollPeriod, ChannelRecord.PRIO_NORMAL);
121   }
122
123   /**
124    * Register new channel for background poller with custom period.
125    *
126    * @param channel channel.
127    * @param period period in millis.
128    */
129   public final void registerChannel(ChannelIF channel, long period) {
130     scheduler.schedule(channel, period, ChannelRecord.PRIO_NORMAL);
131   }
132
133   /**
134    * Performs immediate update of the channel and reworks the schedule starting from current time.
135    * If channel isn't registered yet it will be registered with normal priority.
136    *
137    * @param channel channel to update.
138    */
139   public final void updateChannel(ChannelIF channel) {
140     scheduler.triggerNow(channel);
141   }
142
143   /**
144    * Unregister the channel from background poller.
145    *
146    * @param channel channel to unregister.
147    */
148   public final void unregisterChannel(ChannelIF channel) {
149     scheduler.unschedule(channel);
150   }
151
152   /**
153    * Adds observer to the list of interested parties.
154    *
155    * @param observer new observer.
156    */
157   public final void addObserver(PollerObserverIF observer) {
158     compositeObserver.add(observer);
159   }
160
161   /**
162    * Removes observer from the list.
163    *
164    * @param observer observer to remove.
165    */
166   public final void removeObserver(PollerObserverIF observer) {
167     compositeObserver.remove(observer);
168   }
169
170   /**
171    * Adds approver to the list.
172    *
173    * @param approver approver.
174    */
175   public final void addApprover(PollerApproverIF approver) {
176     compositeApprover.add(approver);
177   }
178
179   /**
180    * Removes approver from the list.
181    *
182    * @param approver approver object.
183    */
184   public final void removeApprover(PollerApproverIF approver) {
185     compositeApprover.remove(approver);
186   }
187
188   /**
189    * Sets global update period to the specified value. All tasks will be rescheduled.
190    *
191    * @param period period in millis.
192    */
193   public final void setPeriod(long period) {
194     this.globalPollPeriod = period;
195     scheduler.rescheduleAll(period);
196   }
197
198   /**
199    * Callback implementation, which simply passes the request to workers manager.
200    */
201   private class SchedulerCallback implements SchedulerCallbackIF {
202     /**
203      * Invoked by scheduler when time to process channel information comes.
204      *
205      * @param record channel record.
206      */
207     public void process(ChannelRecord record) {
208       workersManager.process(record);
209     }
210   }
211
212   /**
213    * Worker threads factory for Poller.
214    */
215   private class PollerThreadFactory implements WorkerThreadFactoryIF {
216     /**
217      * Creates new worker thread object.
218      *
219      * @return worker thread object.
220      */
221     public WorkerThread create() {
222       return new PollerWorkerThread(compositeObserver, compositeApprover);
223     }
224   }
225 }
226
Popular Tags