1 |
85ad3d82
|
Assos Assos
|
<?php
|
2 |
|
|
|
3 |
|
|
/**
|
4 |
|
|
* @file
|
5 |
|
|
* Definition of FeedsSourceInterface and FeedsSource class.
|
6 |
|
|
*/
|
7 |
|
|
|
8 |
|
|
/**
|
9 |
a192dc0b
|
Assos Assos
|
* Distinguish exceptions occurring when handling locks.
|
10 |
85ad3d82
|
Assos Assos
|
*/
|
11 |
|
|
class FeedsLockException extends Exception {}
|
12 |
|
|
|
13 |
|
|
/**
|
14 |
|
|
* Denote a import or clearing stage. Used for multi page processing.
|
15 |
|
|
*/
|
16 |
|
|
define('FEEDS_START', 'start_time');
|
17 |
|
|
define('FEEDS_FETCH', 'fetch');
|
18 |
|
|
define('FEEDS_PARSE', 'parse');
|
19 |
|
|
define('FEEDS_PROCESS', 'process');
|
20 |
|
|
define('FEEDS_PROCESS_CLEAR', 'process_clear');
|
21 |
41cc1b08
|
Assos Assos
|
define('FEEDS_PROCESS_EXPIRE', 'process_expire');
|
22 |
85ad3d82
|
Assos Assos
|
|
23 |
|
|
/**
|
24 |
|
|
* Declares an interface for a class that defines default values and form
|
25 |
|
|
* descriptions for a FeedSource.
|
26 |
|
|
*/
|
27 |
|
|
interface FeedsSourceInterface {
|
28 |
|
|
|
29 |
|
|
/**
|
30 |
|
|
* Crutch: for ease of use, we implement FeedsSourceInterface for every
|
31 |
|
|
* plugin, but then we need to have a handle which plugin actually implements
|
32 |
|
|
* a source.
|
33 |
|
|
*
|
34 |
|
|
* @see FeedsPlugin class.
|
35 |
|
|
*
|
36 |
|
|
* @return
|
37 |
|
|
* TRUE if a plugin handles source specific configuration, FALSE otherwise.
|
38 |
|
|
*/
|
39 |
|
|
public function hasSourceConfig();
|
40 |
|
|
|
41 |
|
|
/**
|
42 |
|
|
* Return an associative array of default values.
|
43 |
|
|
*/
|
44 |
|
|
public function sourceDefaults();
|
45 |
|
|
|
46 |
|
|
/**
|
47 |
|
|
* Return a Form API form array that defines a form configuring values. Keys
|
48 |
|
|
* correspond to the keys of the return value of sourceDefaults().
|
49 |
|
|
*/
|
50 |
|
|
public function sourceForm($source_config);
|
51 |
|
|
|
52 |
|
|
/**
|
53 |
|
|
* Validate user entered values submitted by sourceForm().
|
54 |
|
|
*/
|
55 |
|
|
public function sourceFormValidate(&$source_config);
|
56 |
|
|
|
57 |
|
|
/**
|
58 |
|
|
* A source is being saved.
|
59 |
|
|
*/
|
60 |
|
|
public function sourceSave(FeedsSource $source);
|
61 |
|
|
|
62 |
|
|
/**
|
63 |
|
|
* A source is being deleted.
|
64 |
|
|
*/
|
65 |
|
|
public function sourceDelete(FeedsSource $source);
|
66 |
|
|
}
|
67 |
|
|
|
68 |
|
|
/**
|
69 |
|
|
* Status of an import or clearing operation on a source.
|
70 |
|
|
*/
|
71 |
|
|
class FeedsState {
|
72 |
|
|
/**
|
73 |
|
|
* Floating point number denoting the progress made. 0.0 meaning no progress
|
74 |
|
|
* 1.0 = FEEDS_BATCH_COMPLETE meaning finished.
|
75 |
|
|
*/
|
76 |
|
|
public $progress;
|
77 |
|
|
|
78 |
|
|
/**
|
79 |
|
|
* Used as a pointer to store where left off. Must be serializable.
|
80 |
|
|
*/
|
81 |
|
|
public $pointer;
|
82 |
|
|
|
83 |
|
|
/**
|
84 |
|
|
* Natural numbers denoting more details about the progress being made.
|
85 |
|
|
*/
|
86 |
|
|
public $total;
|
87 |
|
|
public $created;
|
88 |
|
|
public $updated;
|
89 |
|
|
public $deleted;
|
90 |
41cc1b08
|
Assos Assos
|
public $unpublished;
|
91 |
|
|
public $blocked;
|
92 |
85ad3d82
|
Assos Assos
|
public $skipped;
|
93 |
|
|
public $failed;
|
94 |
|
|
|
95 |
41cc1b08
|
Assos Assos
|
/**
|
96 |
|
|
* IDs of entities to be removed.
|
97 |
|
|
*/
|
98 |
|
|
public $removeList;
|
99 |
|
|
|
100 |
85ad3d82
|
Assos Assos
|
/**
|
101 |
|
|
* Constructor, initialize variables.
|
102 |
|
|
*/
|
103 |
|
|
public function __construct() {
|
104 |
|
|
$this->progress = FEEDS_BATCH_COMPLETE;
|
105 |
|
|
$this->total =
|
106 |
|
|
$this->created =
|
107 |
|
|
$this->updated =
|
108 |
|
|
$this->deleted =
|
109 |
41cc1b08
|
Assos Assos
|
$this->unpublished =
|
110 |
|
|
$this->blocked =
|
111 |
85ad3d82
|
Assos Assos
|
$this->skipped =
|
112 |
|
|
$this->failed = 0;
|
113 |
|
|
}
|
114 |
|
|
|
115 |
|
|
/**
|
116 |
|
|
* Safely report progress.
|
117 |
|
|
*
|
118 |
|
|
* When $total == $progress, the state of the task tracked by this state is
|
119 |
|
|
* regarded to be complete.
|
120 |
|
|
*
|
121 |
|
|
* Handles the following cases gracefully:
|
122 |
|
|
*
|
123 |
|
|
* - $total is 0
|
124 |
|
|
* - $progress is larger than $total
|
125 |
|
|
* - $progress approximates $total so that $finished rounds to 1.0
|
126 |
|
|
*
|
127 |
|
|
* @param $total
|
128 |
|
|
* A natural number that is the total to be worked off.
|
129 |
|
|
* @param $progress
|
130 |
|
|
* A natural number that is the progress made on $total.
|
131 |
|
|
*/
|
132 |
|
|
public function progress($total, $progress) {
|
133 |
|
|
if ($progress > $total) {
|
134 |
|
|
$this->progress = FEEDS_BATCH_COMPLETE;
|
135 |
|
|
}
|
136 |
|
|
elseif ($total) {
|
137 |
a192dc0b
|
Assos Assos
|
$this->progress = (float) $progress / $total;
|
138 |
85ad3d82
|
Assos Assos
|
if ($this->progress == FEEDS_BATCH_COMPLETE && $total != $progress) {
|
139 |
|
|
$this->progress = 0.99;
|
140 |
|
|
}
|
141 |
|
|
}
|
142 |
|
|
else {
|
143 |
|
|
$this->progress = FEEDS_BATCH_COMPLETE;
|
144 |
|
|
}
|
145 |
|
|
}
|
146 |
|
|
}
|
147 |
|
|
|
148 |
|
|
/**
|
149 |
|
|
* This class encapsulates a source of a feed. It stores where the feed can be
|
150 |
|
|
* found and how to import it.
|
151 |
|
|
*
|
152 |
|
|
* Information on how to import a feed is encapsulated in a FeedsImporter object
|
153 |
|
|
* which is identified by the common id of the FeedsSource and the
|
154 |
|
|
* FeedsImporter. More than one FeedsSource can use the same FeedsImporter
|
155 |
|
|
* therefore a FeedsImporter never holds a pointer to a FeedsSource object, nor
|
156 |
|
|
* does it hold any other information for a particular FeedsSource object.
|
157 |
|
|
*
|
158 |
|
|
* Classes extending FeedsPlugin can implement a sourceForm to expose
|
159 |
|
|
* configuration for a FeedsSource object. This is for instance how FeedsFetcher
|
160 |
|
|
* exposes a text field for a feed URL or how FeedsCSVParser exposes a select
|
161 |
|
|
* field for choosing between colon or semicolon delimiters.
|
162 |
|
|
*
|
163 |
|
|
* It is important that a FeedsPlugin does not directly hold information about
|
164 |
|
|
* a source but leave all storage up to FeedsSource. An instance of a
|
165 |
|
|
* FeedsPlugin class only exists once per FeedsImporter configuration, while an
|
166 |
|
|
* instance of a FeedsSource class exists once per feed_nid to be imported.
|
167 |
|
|
*
|
168 |
|
|
* As with FeedsImporter, the idea with FeedsSource is that it can be used
|
169 |
|
|
* without actually saving the object to the database.
|
170 |
|
|
*/
|
171 |
|
|
class FeedsSource extends FeedsConfigurable {
|
172 |
|
|
|
173 |
|
|
// Contains the node id of the feed this source info object is attached to.
|
174 |
|
|
// Equals 0 if not attached to any node - i. e. if used on a
|
175 |
|
|
// standalone import form within Feeds or by other API users.
|
176 |
|
|
protected $feed_nid;
|
177 |
|
|
|
178 |
|
|
// The FeedsImporter object that this source is expected to be used with.
|
179 |
|
|
protected $importer;
|
180 |
|
|
|
181 |
|
|
// A FeedsSourceState object holding the current import/clearing state of this
|
182 |
|
|
// source.
|
183 |
|
|
protected $state;
|
184 |
|
|
|
185 |
|
|
// Fetcher result, used to cache fetcher result when batching.
|
186 |
|
|
protected $fetcher_result;
|
187 |
|
|
|
188 |
|
|
// Timestamp when this source was imported the last time.
|
189 |
|
|
protected $imported;
|
190 |
|
|
|
191 |
41cc1b08
|
Assos Assos
|
// Holds an exception object in case an exception occurs during importing.
|
192 |
|
|
protected $exception;
|
193 |
|
|
|
194 |
85ad3d82
|
Assos Assos
|
/**
|
195 |
|
|
* Instantiate a unique object per class/id/feed_nid. Don't use
|
196 |
|
|
* directly, use feeds_source() instead.
|
197 |
|
|
*/
|
198 |
|
|
public static function instance($importer_id, $feed_nid) {
|
199 |
|
|
$class = variable_get('feeds_source_class', 'FeedsSource');
|
200 |
a192dc0b
|
Assos Assos
|
|
201 |
|
|
$instances = &drupal_static(__METHOD__, array());
|
202 |
|
|
|
203 |
85ad3d82
|
Assos Assos
|
if (!isset($instances[$class][$importer_id][$feed_nid])) {
|
204 |
|
|
$instances[$class][$importer_id][$feed_nid] = new $class($importer_id, $feed_nid);
|
205 |
|
|
}
|
206 |
|
|
return $instances[$class][$importer_id][$feed_nid];
|
207 |
|
|
}
|
208 |
|
|
|
209 |
|
|
/**
|
210 |
|
|
* Constructor.
|
211 |
|
|
*/
|
212 |
|
|
protected function __construct($importer_id, $feed_nid) {
|
213 |
|
|
$this->feed_nid = $feed_nid;
|
214 |
|
|
$this->importer = feeds_importer($importer_id);
|
215 |
|
|
parent::__construct($importer_id);
|
216 |
|
|
$this->load();
|
217 |
|
|
}
|
218 |
|
|
|
219 |
|
|
/**
|
220 |
|
|
* Returns the FeedsImporter object that this source is expected to be used with.
|
221 |
|
|
*/
|
222 |
|
|
public function importer() {
|
223 |
|
|
return $this->importer;
|
224 |
|
|
}
|
225 |
|
|
|
226 |
|
|
/**
|
227 |
|
|
* Preview = fetch and parse a feed.
|
228 |
|
|
*
|
229 |
|
|
* @return
|
230 |
|
|
* FeedsParserResult object.
|
231 |
|
|
*
|
232 |
|
|
* @throws
|
233 |
|
|
* Throws Exception if an error occurs when fetching or parsing.
|
234 |
|
|
*/
|
235 |
|
|
public function preview() {
|
236 |
|
|
$result = $this->importer->fetcher->fetch($this);
|
237 |
|
|
$result = $this->importer->parser->parse($this, $result);
|
238 |
|
|
module_invoke_all('feeds_after_parse', $this, $result);
|
239 |
|
|
return $result;
|
240 |
|
|
}
|
241 |
|
|
|
242 |
|
|
/**
|
243 |
|
|
* Start importing a source.
|
244 |
|
|
*
|
245 |
|
|
* This method starts an import job. Depending on the configuration of the
|
246 |
|
|
* importer of this source, a Batch API job or a background job with Job
|
247 |
|
|
* Scheduler will be created.
|
248 |
|
|
*
|
249 |
|
|
* @throws Exception
|
250 |
|
|
* If processing in background is enabled, the first batch chunk of the
|
251 |
|
|
* import will be executed on the current page request. This means that this
|
252 |
|
|
* method may throw the same exceptions as FeedsSource::import().
|
253 |
|
|
*/
|
254 |
|
|
public function startImport() {
|
255 |
|
|
$config = $this->importer->getConfig();
|
256 |
|
|
if ($config['process_in_background']) {
|
257 |
|
|
$this->startBackgroundJob('import');
|
258 |
|
|
}
|
259 |
|
|
else {
|
260 |
|
|
$this->startBatchAPIJob(t('Importing'), 'import');
|
261 |
|
|
}
|
262 |
|
|
}
|
263 |
|
|
|
264 |
|
|
/**
|
265 |
|
|
* Start deleting all imported items of a source.
|
266 |
|
|
*
|
267 |
|
|
* This method starts a clear job. Depending on the configuration of the
|
268 |
|
|
* importer of this source, a Batch API job or a background job with Job
|
269 |
|
|
* Scheduler will be created.
|
270 |
|
|
*
|
271 |
|
|
* @throws Exception
|
272 |
|
|
* If processing in background is enabled, the first batch chunk of the
|
273 |
|
|
* clear task will be executed on the current page request. This means that
|
274 |
|
|
* this method may throw the same exceptions as FeedsSource::clear().
|
275 |
|
|
*/
|
276 |
|
|
public function startClear() {
|
277 |
|
|
$config = $this->importer->getConfig();
|
278 |
|
|
if ($config['process_in_background']) {
|
279 |
|
|
$this->startBackgroundJob('clear');
|
280 |
|
|
}
|
281 |
|
|
else {
|
282 |
|
|
$this->startBatchAPIJob(t('Deleting'), 'clear');
|
283 |
|
|
}
|
284 |
|
|
}
|
285 |
|
|
|
286 |
|
|
/**
|
287 |
|
|
* Schedule all periodic tasks for this source.
|
288 |
|
|
*/
|
289 |
|
|
public function schedule() {
|
290 |
|
|
$this->scheduleImport();
|
291 |
41cc1b08
|
Assos Assos
|
$this->scheduleExpire();
|
292 |
85ad3d82
|
Assos Assos
|
}
|
293 |
|
|
|
294 |
|
|
/**
|
295 |
|
|
* Schedule periodic or background import tasks.
|
296 |
|
|
*/
|
297 |
|
|
public function scheduleImport() {
|
298 |
|
|
// Check whether any fetcher is overriding the import period.
|
299 |
|
|
$period = $this->importer->config['import_period'];
|
300 |
|
|
$fetcher_period = $this->importer->fetcher->importPeriod($this);
|
301 |
|
|
if (is_numeric($fetcher_period)) {
|
302 |
|
|
$period = $fetcher_period;
|
303 |
|
|
}
|
304 |
|
|
$job = array(
|
305 |
|
|
'type' => $this->id,
|
306 |
|
|
'id' => $this->feed_nid,
|
307 |
|
|
'period' => $period,
|
308 |
|
|
'periodic' => TRUE,
|
309 |
|
|
);
|
310 |
41cc1b08
|
Assos Assos
|
if ($period == FEEDS_SCHEDULE_NEVER && $this->progressImporting() === FEEDS_BATCH_COMPLETE) {
|
311 |
|
|
JobScheduler::get('feeds_source_import')->remove($job);
|
312 |
|
|
}
|
313 |
|
|
elseif ($this->progressImporting() === FEEDS_BATCH_COMPLETE) {
|
314 |
85ad3d82
|
Assos Assos
|
JobScheduler::get('feeds_source_import')->set($job);
|
315 |
|
|
}
|
316 |
|
|
else {
|
317 |
41cc1b08
|
Assos Assos
|
// Feed is not fully imported yet, so we put this job back in the queue
|
318 |
|
|
// immediately for further processing.
|
319 |
|
|
$queue = DrupalQueue::get('feeds_source_import');
|
320 |
|
|
$queue->createItem($job);
|
321 |
|
|
}
|
322 |
|
|
}
|
323 |
|
|
|
324 |
|
|
/**
|
325 |
|
|
* Schedule background expire tasks.
|
326 |
|
|
*/
|
327 |
|
|
public function scheduleExpire() {
|
328 |
|
|
// Schedule as soon as possible if a batch is active.
|
329 |
|
|
$period = $this->progressExpiring() === FEEDS_BATCH_COMPLETE ? 3600 : 0;
|
330 |
|
|
|
331 |
|
|
$job = array(
|
332 |
|
|
'type' => $this->id,
|
333 |
|
|
'id' => $this->feed_nid,
|
334 |
|
|
'period' => $period,
|
335 |
|
|
'periodic' => TRUE,
|
336 |
|
|
);
|
337 |
|
|
if ($this->importer->processor->expiryTime() == FEEDS_EXPIRE_NEVER) {
|
338 |
|
|
JobScheduler::get('feeds_source_expire')->remove($job);
|
339 |
|
|
}
|
340 |
|
|
else {
|
341 |
|
|
JobScheduler::get('feeds_source_expire')->set($job);
|
342 |
85ad3d82
|
Assos Assos
|
}
|
343 |
|
|
}
|
344 |
|
|
|
345 |
|
|
/**
|
346 |
|
|
* Schedule background clearing tasks.
|
347 |
|
|
*/
|
348 |
|
|
public function scheduleClear() {
|
349 |
|
|
$job = array(
|
350 |
|
|
'type' => $this->id,
|
351 |
|
|
'id' => $this->feed_nid,
|
352 |
|
|
'period' => 0,
|
353 |
|
|
'periodic' => TRUE,
|
354 |
|
|
);
|
355 |
|
|
// Remove job if batch is complete.
|
356 |
|
|
if ($this->progressClearing() === FEEDS_BATCH_COMPLETE) {
|
357 |
|
|
JobScheduler::get('feeds_source_clear')->remove($job);
|
358 |
|
|
}
|
359 |
|
|
// Schedule as soon as possible if batch is not complete.
|
360 |
|
|
else {
|
361 |
|
|
JobScheduler::get('feeds_source_clear')->set($job);
|
362 |
|
|
}
|
363 |
|
|
}
|
364 |
|
|
|
365 |
|
|
/**
|
366 |
|
|
* Import a source: execute fetching, parsing and processing stage.
|
367 |
|
|
*
|
368 |
|
|
* This method only executes the current batch chunk, then returns. If you are
|
369 |
|
|
* looking to import an entire source, use FeedsSource::startImport() instead.
|
370 |
|
|
*
|
371 |
|
|
* @return
|
372 |
|
|
* FEEDS_BATCH_COMPLETE if the import process finished. A decimal between
|
373 |
|
|
* 0.0 and 0.9 periodic if import is still in progress.
|
374 |
|
|
*
|
375 |
|
|
* @throws
|
376 |
|
|
* Throws Exception if an error occurs when importing.
|
377 |
|
|
*/
|
378 |
|
|
public function import() {
|
379 |
|
|
$this->acquireLock();
|
380 |
|
|
try {
|
381 |
|
|
// If fetcher result is empty, we are starting a new import, log.
|
382 |
|
|
if (empty($this->fetcher_result)) {
|
383 |
41cc1b08
|
Assos Assos
|
module_invoke_all('feeds_before_import', $this);
|
384 |
85ad3d82
|
Assos Assos
|
$this->state[FEEDS_START] = time();
|
385 |
|
|
}
|
386 |
|
|
|
387 |
|
|
// Fetch.
|
388 |
|
|
if (empty($this->fetcher_result) || FEEDS_BATCH_COMPLETE == $this->progressParsing()) {
|
389 |
|
|
$this->fetcher_result = $this->importer->fetcher->fetch($this);
|
390 |
|
|
// Clean the parser's state, we are parsing an entirely new file.
|
391 |
|
|
unset($this->state[FEEDS_PARSE]);
|
392 |
|
|
}
|
393 |
|
|
|
394 |
|
|
// Parse.
|
395 |
|
|
$parser_result = $this->importer->parser->parse($this, $this->fetcher_result);
|
396 |
|
|
module_invoke_all('feeds_after_parse', $this, $parser_result);
|
397 |
|
|
|
398 |
|
|
// Process.
|
399 |
|
|
$this->importer->processor->process($this, $parser_result);
|
400 |
41cc1b08
|
Assos Assos
|
|
401 |
|
|
// Import finished without exceptions, so unset any potentially previously
|
402 |
|
|
// recorded exceptions.
|
403 |
|
|
unset($this->exception);
|
404 |
85ad3d82
|
Assos Assos
|
}
|
405 |
|
|
catch (Exception $e) {
|
406 |
41cc1b08
|
Assos Assos
|
// $e is stored and re-thrown once we've had a chance to log our progress.
|
407 |
|
|
// Set the exception so that other modules can check if an exception
|
408 |
|
|
// occurred in hook_feeds_after_import().
|
409 |
|
|
$this->exception = $e;
|
410 |
85ad3d82
|
Assos Assos
|
}
|
411 |
|
|
|
412 |
|
|
// Clean up.
|
413 |
|
|
$result = $this->progressImporting();
|
414 |
|
|
if ($result == FEEDS_BATCH_COMPLETE || isset($e)) {
|
415 |
|
|
$this->imported = time();
|
416 |
a192dc0b
|
Assos Assos
|
$this->log('import', 'Imported in @s seconds.', array('@s' => $this->imported - $this->state[FEEDS_START]), WATCHDOG_INFO);
|
417 |
85ad3d82
|
Assos Assos
|
module_invoke_all('feeds_after_import', $this);
|
418 |
|
|
unset($this->fetcher_result, $this->state);
|
419 |
|
|
}
|
420 |
|
|
$this->save();
|
421 |
a192dc0b
|
Assos Assos
|
|
422 |
|
|
$this->releaseLock();
|
423 |
|
|
|
424 |
85ad3d82
|
Assos Assos
|
if (isset($e)) {
|
425 |
|
|
throw $e;
|
426 |
|
|
}
|
427 |
a192dc0b
|
Assos Assos
|
|
428 |
85ad3d82
|
Assos Assos
|
return $result;
|
429 |
|
|
}
|
430 |
|
|
|
431 |
a192dc0b
|
Assos Assos
|
/**
|
432 |
|
|
* Imports a fetcher result all at once in memory.
|
433 |
|
|
*
|
434 |
|
|
* @param FeedsFetcherResult $fetcher_result
|
435 |
|
|
* The fetcher result to process.
|
436 |
|
|
*
|
437 |
|
|
* @throws Exception
|
438 |
|
|
* Thrown if an error occurs when importing.
|
439 |
|
|
*/
|
440 |
|
|
public function pushImport(FeedsFetcherResult $fetcher_result) {
|
441 |
|
|
// Since locks only work during a request, check if an import is active.
|
442 |
|
|
if (!empty($this->fetcher_result) || !empty($this->state)) {
|
443 |
|
|
throw new RuntimeException('The feed is currently importing.');
|
444 |
|
|
}
|
445 |
|
|
|
446 |
|
|
$this->acquireLock();
|
447 |
|
|
$start = time();
|
448 |
|
|
|
449 |
|
|
try {
|
450 |
|
|
module_invoke_all('feeds_before_import', $this);
|
451 |
|
|
|
452 |
|
|
// Parse.
|
453 |
|
|
do {
|
454 |
|
|
$parser_result = $this->importer->parser->parse($this, $fetcher_result);
|
455 |
|
|
module_invoke_all('feeds_after_parse', $this, $parser_result);
|
456 |
|
|
|
457 |
|
|
// Process.
|
458 |
|
|
$this->importer->processor->process($this, $parser_result);
|
459 |
|
|
|
460 |
|
|
} while ($this->progressParsing() !== FEEDS_BATCH_COMPLETE);
|
461 |
|
|
}
|
462 |
|
|
catch (Exception $e) {
|
463 |
|
|
// $e is stored and re-thrown once we've had a chance to log our progress.
|
464 |
|
|
// Set the exception so that other modules can check if an exception
|
465 |
|
|
// occurred in hook_feeds_after_import().
|
466 |
|
|
$this->exception = $e;
|
467 |
|
|
}
|
468 |
|
|
|
469 |
|
|
module_invoke_all('feeds_after_import', $this);
|
470 |
|
|
|
471 |
|
|
$this->imported = time();
|
472 |
|
|
$this->log('import', 'Imported in @s seconds.', array('@s' => $this->imported - $start), WATCHDOG_INFO);
|
473 |
|
|
|
474 |
|
|
unset($this->fetcher_result, $this->state);
|
475 |
|
|
|
476 |
|
|
$this->save();
|
477 |
|
|
|
478 |
|
|
$this->releaseLock();
|
479 |
|
|
|
480 |
|
|
if (isset($e)) {
|
481 |
|
|
throw $e;
|
482 |
|
|
}
|
483 |
|
|
}
|
484 |
|
|
|
485 |
85ad3d82
|
Assos Assos
|
/**
|
486 |
|
|
* Remove all items from a feed.
|
487 |
|
|
*
|
488 |
|
|
* This method only executes the current batch chunk, then returns. If you are
|
489 |
|
|
* looking to delete all items of a source, use FeedsSource::startClear()
|
490 |
|
|
* instead.
|
491 |
|
|
*
|
492 |
|
|
* @return
|
493 |
|
|
* FEEDS_BATCH_COMPLETE if the clearing process finished. A decimal between
|
494 |
|
|
* 0.0 and 0.9 periodic if clearing is still in progress.
|
495 |
|
|
*
|
496 |
|
|
* @throws
|
497 |
|
|
* Throws Exception if an error occurs when clearing.
|
498 |
|
|
*/
|
499 |
|
|
public function clear() {
|
500 |
|
|
$this->acquireLock();
|
501 |
|
|
try {
|
502 |
|
|
$this->importer->fetcher->clear($this);
|
503 |
|
|
$this->importer->parser->clear($this);
|
504 |
|
|
$this->importer->processor->clear($this);
|
505 |
|
|
}
|
506 |
|
|
catch (Exception $e) {
|
507 |
41cc1b08
|
Assos Assos
|
// $e is stored and re-thrown once we've had a chance to log our progress.
|
508 |
85ad3d82
|
Assos Assos
|
}
|
509 |
|
|
$this->releaseLock();
|
510 |
|
|
|
511 |
|
|
// Clean up.
|
512 |
|
|
$result = $this->progressClearing();
|
513 |
|
|
if ($result == FEEDS_BATCH_COMPLETE || isset($e)) {
|
514 |
|
|
module_invoke_all('feeds_after_clear', $this);
|
515 |
|
|
unset($this->state);
|
516 |
|
|
}
|
517 |
|
|
$this->save();
|
518 |
|
|
if (isset($e)) {
|
519 |
|
|
throw $e;
|
520 |
|
|
}
|
521 |
|
|
return $result;
|
522 |
|
|
}
|
523 |
|
|
|
524 |
41cc1b08
|
Assos Assos
|
/**
|
525 |
|
|
* Removes all expired items from a feed.
|
526 |
|
|
*/
|
527 |
|
|
public function expire() {
|
528 |
|
|
$this->acquireLock();
|
529 |
|
|
try {
|
530 |
|
|
$result = $this->importer->processor->expire($this);
|
531 |
|
|
}
|
532 |
|
|
catch (Exception $e) {
|
533 |
|
|
// Will throw after the lock is released.
|
534 |
|
|
}
|
535 |
|
|
$this->releaseLock();
|
536 |
|
|
|
537 |
|
|
if (isset($e)) {
|
538 |
|
|
throw $e;
|
539 |
|
|
}
|
540 |
|
|
|
541 |
|
|
return $result;
|
542 |
|
|
}
|
543 |
|
|
|
544 |
85ad3d82
|
Assos Assos
|
/**
|
545 |
|
|
* Report progress as float between 0 and 1. 1 = FEEDS_BATCH_COMPLETE.
|
546 |
|
|
*/
|
547 |
|
|
public function progressParsing() {
|
548 |
|
|
return $this->state(FEEDS_PARSE)->progress;
|
549 |
|
|
}
|
550 |
|
|
|
551 |
|
|
/**
|
552 |
|
|
* Report progress as float between 0 and 1. 1 = FEEDS_BATCH_COMPLETE.
|
553 |
|
|
*/
|
554 |
|
|
public function progressImporting() {
|
555 |
|
|
$fetcher = $this->state(FEEDS_FETCH);
|
556 |
|
|
$parser = $this->state(FEEDS_PARSE);
|
557 |
|
|
if ($fetcher->progress == FEEDS_BATCH_COMPLETE && $parser->progress == FEEDS_BATCH_COMPLETE) {
|
558 |
|
|
return FEEDS_BATCH_COMPLETE;
|
559 |
|
|
}
|
560 |
|
|
// Fetching envelops parsing.
|
561 |
|
|
// @todo: this assumes all fetchers neatly use total. May not be the case.
|
562 |
|
|
$fetcher_fraction = $fetcher->total ? 1.0 / $fetcher->total : 1.0;
|
563 |
|
|
$parser_progress = $parser->progress * $fetcher_fraction;
|
564 |
|
|
$result = $fetcher->progress - $fetcher_fraction + $parser_progress;
|
565 |
|
|
if ($result == FEEDS_BATCH_COMPLETE) {
|
566 |
|
|
return 0.99;
|
567 |
|
|
}
|
568 |
|
|
return $result;
|
569 |
|
|
}
|
570 |
|
|
|
571 |
|
|
/**
|
572 |
|
|
* Report progress on clearing.
|
573 |
|
|
*/
|
574 |
|
|
public function progressClearing() {
|
575 |
|
|
return $this->state(FEEDS_PROCESS_CLEAR)->progress;
|
576 |
|
|
}
|
577 |
|
|
|
578 |
41cc1b08
|
Assos Assos
|
/**
|
579 |
|
|
* Report progress on expiry.
|
580 |
|
|
*/
|
581 |
|
|
public function progressExpiring() {
|
582 |
|
|
return $this->state(FEEDS_PROCESS_EXPIRE)->progress;
|
583 |
|
|
}
|
584 |
|
|
|
585 |
85ad3d82
|
Assos Assos
|
/**
|
586 |
|
|
* Return a state object for a given stage. Lazy instantiates new states.
|
587 |
|
|
*
|
588 |
|
|
* @todo Rename getConfigFor() accordingly to config().
|
589 |
|
|
*
|
590 |
|
|
* @param $stage
|
591 |
|
|
* One of FEEDS_FETCH, FEEDS_PARSE, FEEDS_PROCESS or FEEDS_PROCESS_CLEAR.
|
592 |
|
|
*
|
593 |
|
|
* @return
|
594 |
|
|
* The FeedsState object for the given stage.
|
595 |
|
|
*/
|
596 |
|
|
public function state($stage) {
|
597 |
|
|
if (!is_array($this->state)) {
|
598 |
|
|
$this->state = array();
|
599 |
|
|
}
|
600 |
|
|
if (!isset($this->state[$stage])) {
|
601 |
|
|
$this->state[$stage] = new FeedsState();
|
602 |
|
|
}
|
603 |
|
|
return $this->state[$stage];
|
604 |
|
|
}
|
605 |
|
|
|
606 |
|
|
/**
|
607 |
|
|
* Count items imported by this source.
|
608 |
|
|
*/
|
609 |
|
|
public function itemCount() {
|
610 |
|
|
return $this->importer->processor->itemCount($this);
|
611 |
|
|
}
|
612 |
|
|
|
613 |
|
|
/**
|
614 |
|
|
* Save configuration.
|
615 |
|
|
*/
|
616 |
|
|
public function save() {
|
617 |
|
|
// Alert implementers of FeedsSourceInterface to the fact that we're saving.
|
618 |
|
|
foreach ($this->importer->plugin_types as $type) {
|
619 |
|
|
$this->importer->$type->sourceSave($this);
|
620 |
|
|
}
|
621 |
|
|
$config = $this->getConfig();
|
622 |
|
|
|
623 |
|
|
// Store the source property of the fetcher in a separate column so that we
|
624 |
|
|
// can do fast lookups on it.
|
625 |
|
|
$source = '';
|
626 |
|
|
if (isset($config[get_class($this->importer->fetcher)]['source'])) {
|
627 |
|
|
$source = $config[get_class($this->importer->fetcher)]['source'];
|
628 |
|
|
}
|
629 |
|
|
$object = array(
|
630 |
|
|
'id' => $this->id,
|
631 |
|
|
'feed_nid' => $this->feed_nid,
|
632 |
|
|
'imported' => $this->imported,
|
633 |
|
|
'config' => $config,
|
634 |
|
|
'source' => $source,
|
635 |
|
|
'state' => isset($this->state) ? $this->state : FALSE,
|
636 |
|
|
'fetcher_result' => isset($this->fetcher_result) ? $this->fetcher_result : FALSE,
|
637 |
|
|
);
|
638 |
|
|
if (db_query_range("SELECT 1 FROM {feeds_source} WHERE id = :id AND feed_nid = :nid", 0, 1, array(':id' => $this->id, ':nid' => $this->feed_nid))->fetchField()) {
|
639 |
|
|
drupal_write_record('feeds_source', $object, array('id', 'feed_nid'));
|
640 |
|
|
}
|
641 |
|
|
else {
|
642 |
|
|
drupal_write_record('feeds_source', $object);
|
643 |
|
|
}
|
644 |
|
|
}
|
645 |
|
|
|
646 |
|
|
/**
|
647 |
|
|
* Load configuration and unpack.
|
648 |
|
|
*
|
649 |
|
|
* @todo Patch CTools to move constants from export.inc to ctools.module.
|
650 |
|
|
*/
|
651 |
|
|
public function load() {
|
652 |
|
|
if ($record = db_query("SELECT imported, config, state, fetcher_result FROM {feeds_source} WHERE id = :id AND feed_nid = :nid", array(':id' => $this->id, ':nid' => $this->feed_nid))->fetchObject()) {
|
653 |
|
|
// While FeedsSource cannot be exported, we still use CTool's export.inc
|
654 |
|
|
// export definitions.
|
655 |
|
|
ctools_include('export');
|
656 |
|
|
$this->export_type = EXPORT_IN_DATABASE;
|
657 |
|
|
$this->imported = $record->imported;
|
658 |
|
|
$this->config = unserialize($record->config);
|
659 |
|
|
if (!empty($record->state)) {
|
660 |
|
|
$this->state = unserialize($record->state);
|
661 |
|
|
}
|
662 |
|
|
if (!is_array($this->state)) {
|
663 |
|
|
$this->state = array();
|
664 |
|
|
}
|
665 |
|
|
if (!empty($record->fetcher_result)) {
|
666 |
|
|
$this->fetcher_result = unserialize($record->fetcher_result);
|
667 |
|
|
}
|
668 |
|
|
}
|
669 |
|
|
}
|
670 |
|
|
|
671 |
|
|
/**
|
672 |
|
|
* Delete configuration. Removes configuration information
|
673 |
|
|
* from database, does not delete configuration itself.
|
674 |
|
|
*/
|
675 |
|
|
public function delete() {
|
676 |
|
|
// Alert implementers of FeedsSourceInterface to the fact that we're
|
677 |
|
|
// deleting.
|
678 |
|
|
foreach ($this->importer->plugin_types as $type) {
|
679 |
|
|
$this->importer->$type->sourceDelete($this);
|
680 |
|
|
}
|
681 |
|
|
db_delete('feeds_source')
|
682 |
|
|
->condition('id', $this->id)
|
683 |
|
|
->condition('feed_nid', $this->feed_nid)
|
684 |
|
|
->execute();
|
685 |
|
|
// Remove from schedule.
|
686 |
|
|
$job = array(
|
687 |
|
|
'type' => $this->id,
|
688 |
|
|
'id' => $this->feed_nid,
|
689 |
|
|
);
|
690 |
|
|
JobScheduler::get('feeds_source_import')->remove($job);
|
691 |
41cc1b08
|
Assos Assos
|
JobScheduler::get('feeds_source_expire')->remove($job);
|
692 |
85ad3d82
|
Assos Assos
|
}
|
693 |
|
|
|
694 |
|
|
/**
|
695 |
|
|
* Only return source if configuration is persistent and valid.
|
696 |
|
|
*
|
697 |
|
|
* @see FeedsConfigurable::existing().
|
698 |
|
|
*/
|
699 |
|
|
public function existing() {
|
700 |
|
|
// If there is no feed nid given, there must be no content type specified.
|
701 |
|
|
// If there is a feed nid given, there must be a content type specified.
|
702 |
|
|
// Ensure that importer is persistent (= defined in code or DB).
|
703 |
|
|
// Ensure that source is persistent (= defined in DB).
|
704 |
|
|
if ((empty($this->feed_nid) && empty($this->importer->config['content_type'])) ||
|
705 |
|
|
(!empty($this->feed_nid) && !empty($this->importer->config['content_type']))) {
|
706 |
|
|
$this->importer->existing();
|
707 |
|
|
return parent::existing();
|
708 |
|
|
}
|
709 |
|
|
throw new FeedsNotExistingException(t('Source configuration not valid.'));
|
710 |
|
|
}
|
711 |
|
|
|
712 |
|
|
/**
|
713 |
|
|
* Returns the configuration for a specific client class.
|
714 |
|
|
*
|
715 |
|
|
* @param FeedsSourceInterface $client
|
716 |
|
|
* An object that is an implementer of FeedsSourceInterface.
|
717 |
|
|
*
|
718 |
|
|
* @return
|
719 |
|
|
* An array stored for $client.
|
720 |
|
|
*/
|
721 |
|
|
public function getConfigFor(FeedsSourceInterface $client) {
|
722 |
|
|
$class = get_class($client);
|
723 |
|
|
return isset($this->config[$class]) ? $this->config[$class] : $client->sourceDefaults();
|
724 |
|
|
}
|
725 |
|
|
|
726 |
|
|
/**
|
727 |
|
|
* Sets the configuration for a specific client class.
|
728 |
|
|
*
|
729 |
|
|
* @param FeedsSourceInterface $client
|
730 |
|
|
* An object that is an implementer of FeedsSourceInterface.
|
731 |
|
|
* @param $config
|
732 |
|
|
* The configuration for $client.
|
733 |
|
|
*
|
734 |
|
|
* @return
|
735 |
|
|
* An array stored for $client.
|
736 |
|
|
*/
|
737 |
|
|
public function setConfigFor(FeedsSourceInterface $client, $config) {
|
738 |
|
|
$this->config[get_class($client)] = $config;
|
739 |
|
|
}
|
740 |
|
|
|
741 |
|
|
/**
|
742 |
|
|
* Return defaults for feed configuration.
|
743 |
|
|
*/
|
744 |
|
|
public function configDefaults() {
|
745 |
|
|
// Collect information from plugins.
|
746 |
|
|
$defaults = array();
|
747 |
|
|
foreach ($this->importer->plugin_types as $type) {
|
748 |
|
|
if ($this->importer->$type->hasSourceConfig()) {
|
749 |
|
|
$defaults[get_class($this->importer->$type)] = $this->importer->$type->sourceDefaults();
|
750 |
|
|
}
|
751 |
|
|
}
|
752 |
|
|
return $defaults;
|
753 |
|
|
}
|
754 |
|
|
|
755 |
|
|
/**
|
756 |
|
|
* Override parent::configForm().
|
757 |
|
|
*/
|
758 |
|
|
public function configForm(&$form_state) {
|
759 |
|
|
// Collect information from plugins.
|
760 |
|
|
$form = array();
|
761 |
|
|
foreach ($this->importer->plugin_types as $type) {
|
762 |
|
|
if ($this->importer->$type->hasSourceConfig()) {
|
763 |
|
|
$class = get_class($this->importer->$type);
|
764 |
|
|
$config = isset($this->config[$class]) ? $this->config[$class] : array();
|
765 |
|
|
$form[$class] = $this->importer->$type->sourceForm($config);
|
766 |
|
|
$form[$class]['#tree'] = TRUE;
|
767 |
|
|
}
|
768 |
|
|
}
|
769 |
|
|
return $form;
|
770 |
|
|
}
|
771 |
|
|
|
772 |
|
|
/**
|
773 |
|
|
* Override parent::configFormValidate().
|
774 |
|
|
*/
|
775 |
|
|
public function configFormValidate(&$values) {
|
776 |
|
|
foreach ($this->importer->plugin_types as $type) {
|
777 |
|
|
$class = get_class($this->importer->$type);
|
778 |
|
|
if (isset($values[$class]) && $this->importer->$type->hasSourceConfig()) {
|
779 |
|
|
$this->importer->$type->sourceFormValidate($values[$class]);
|
780 |
|
|
}
|
781 |
|
|
}
|
782 |
|
|
}
|
783 |
|
|
|
784 |
|
|
/**
|
785 |
|
|
* Writes to feeds log.
|
786 |
|
|
*/
|
787 |
|
|
public function log($type, $message, $variables = array(), $severity = WATCHDOG_NOTICE) {
|
788 |
|
|
feeds_log($this->id, $this->feed_nid, $type, $message, $variables, $severity);
|
789 |
|
|
}
|
790 |
|
|
|
791 |
|
|
/**
|
792 |
|
|
* Background job helper. Starts a background job using Job Scheduler.
|
793 |
|
|
*
|
794 |
|
|
* Execute the first batch chunk of a background job on the current page load,
|
795 |
|
|
* moves the rest of the job processing to a cron powered background job.
|
796 |
|
|
*
|
797 |
|
|
* Executing the first batch chunk is important, otherwise, when a user
|
798 |
|
|
* submits a source for import or clearing, we will leave her without any
|
799 |
|
|
* visual indicators of an ongoing job.
|
800 |
|
|
*
|
801 |
|
|
* @see FeedsSource::startImport().
|
802 |
|
|
* @see FeedsSource::startClear().
|
803 |
|
|
*
|
804 |
|
|
* @param $method
|
805 |
|
|
* Method to execute on importer; one of 'import' or 'clear'.
|
806 |
|
|
*
|
807 |
|
|
* @throws Exception $e
|
808 |
|
|
*/
|
809 |
|
|
protected function startBackgroundJob($method) {
|
810 |
|
|
if (FEEDS_BATCH_COMPLETE != $this->$method()) {
|
811 |
|
|
$job = array(
|
812 |
|
|
'type' => $this->id,
|
813 |
|
|
'id' => $this->feed_nid,
|
814 |
|
|
'period' => 0,
|
815 |
|
|
'periodic' => FALSE,
|
816 |
|
|
);
|
817 |
|
|
JobScheduler::get("feeds_source_{$method}")->set($job);
|
818 |
|
|
}
|
819 |
|
|
}
|
820 |
|
|
|
821 |
|
|
/**
|
822 |
|
|
* Batch API helper. Starts a Batch API job.
|
823 |
|
|
*
|
824 |
|
|
* @see FeedsSource::startImport().
|
825 |
|
|
* @see FeedsSource::startClear().
|
826 |
|
|
* @see feeds_batch()
|
827 |
|
|
*
|
828 |
|
|
* @param $title
|
829 |
|
|
* Title to show to user when executing batch.
|
830 |
|
|
* @param $method
|
831 |
|
|
* Method to execute on importer; one of 'import' or 'clear'.
|
832 |
|
|
*/
|
833 |
|
|
protected function startBatchAPIJob($title, $method) {
|
834 |
|
|
$batch = array(
|
835 |
|
|
'title' => $title,
|
836 |
|
|
'operations' => array(
|
837 |
|
|
array('feeds_batch', array($method, $this->id, $this->feed_nid)),
|
838 |
|
|
),
|
839 |
|
|
'progress_message' => '',
|
840 |
|
|
);
|
841 |
|
|
batch_set($batch);
|
842 |
|
|
}
|
843 |
|
|
|
844 |
|
|
/**
|
845 |
|
|
* Acquires a lock for this source.
|
846 |
|
|
*
|
847 |
|
|
* @throws FeedsLockException
|
848 |
|
|
* If a lock for the requested job could not be acquired.
|
849 |
|
|
*/
|
850 |
|
|
protected function acquireLock() {
|
851 |
|
|
if (!lock_acquire("feeds_source_{$this->id}_{$this->feed_nid}", 60.0)) {
|
852 |
|
|
throw new FeedsLockException(t('Cannot acquire lock for source @id / @feed_nid.', array('@id' => $this->id, '@feed_nid' => $this->feed_nid)));
|
853 |
|
|
}
|
854 |
|
|
}
|
855 |
|
|
|
856 |
|
|
/**
|
857 |
|
|
* Releases a lock for this source.
|
858 |
|
|
*/
|
859 |
|
|
protected function releaseLock() {
|
860 |
|
|
lock_release("feeds_source_{$this->id}_{$this->feed_nid}");
|
861 |
|
|
}
|
862 |
41cc1b08
|
Assos Assos
|
|
863 |
|
|
/**
|
864 |
|
|
* Implements FeedsConfigurable::dependencies().
|
865 |
|
|
*/
|
866 |
|
|
public function dependencies() {
|
867 |
|
|
$dependencies = parent::dependencies();
|
868 |
|
|
return array_merge($dependencies, $this->importer()->dependencies());
|
869 |
|
|
}
|
870 |
|
|
|
871 |
85ad3d82
|
Assos Assos
|
} |