? 617054-16_push.patch ? 714692-8_modify_url.patch ? libraries/simplepie.inc Index: feeds.install =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.install,v retrieving revision 1.6 diff -u -p -r1.6 feeds.install --- feeds.install 10 Feb 2010 23:49:35 -0000 1.6 +++ feeds.install 18 Feb 2010 16:56:32 -0000 @@ -187,6 +187,59 @@ function feeds_schema() { 'guid' => array(array('guid', 255)), ), ); + $schema['feeds_push_subscriptions'] = array( + 'description' => 'PubsubHubbub subscriptions.', + 'fields' => array( + 'domain' => array( + 'type' => 'varchar', + 'length' => 128, + 'not null' => TRUE, + 'default' => '', + 'description' => 'Domain of the subscriber.', + ), + 'subscriber_id' => array( + 'type' => 'int', + 'not null' => TRUE, + 'default' => 0, + 'unsigned' => TRUE, + 'description' => 'ID of the subscriber.', + ), + 'timestamp' => array( + 'type' => 'int', + 'unsigned' => FALSE, + 'default' => 0, + 'not null' => TRUE, + 'description' => 'Created timestamp.', + ), + 'hub' => array( + 'type' => 'text', + 'not null' => TRUE, + 'description' => t('The URL of the hub endpoint of this subscription.'), + ), + 'topic' => array( + 'type' => 'text', + 'not null' => TRUE, + 'description' => t('The topic URL (feed URL) of this subscription.'), + ), + 'mode' => array( + 'type' => 'varchar', + 'length' => 64, + 'not null' => TRUE, + 'default' => '', + 'description' => 'Subscribe or unsubscribe. Empty if subscribed.', + ), + 'post_fields' => array( + 'type' => 'text', + 'not null' => FALSE, + 'description' => 'Fields posted.', + 'serialize' => TRUE, + ), + ), + 'primary key' => array('domain', 'subscriber_id'), + 'indexes' => array( + 'timestamp' => array('timestamp'), + ), + ); return $schema; } @@ -362,4 +415,66 @@ function feeds_update_6008() { db_change_field($ret, 'feeds_schedule', 'last_scheduled_time', 'last_executed_time', $spec); return $ret; -} \ No newline at end of file +} + +/** + * Add feeds_push_subscriptions tables. + */ +function feeds_update_6009() { + $ret = array(); + $table = array( + 'description' => 'PubsubHubbub subscriptions.', + 'fields' => array( + 'domain' => array( + 'type' => 'varchar', + 'length' => 128, + 'not null' => TRUE, + 'default' => '', + 'description' => 'Domain of the subscriber.', + ), + 'subscriber_id' => array( + 'type' => 'int', + 'not null' => TRUE, + 'default' => 0, + 'unsigned' => TRUE, + 'description' => 'ID of the subscriber.', + ), + 'timestamp' => array( + 'type' => 'int', + 'unsigned' => FALSE, + 'default' => 0, + 'not null' => TRUE, + 'description' => 'Created timestamp.', + ), + 'hub' => array( + 'type' => 'text', + 'not null' => TRUE, + 'description' => t('The URL of the hub endpoint of this subscription.'), + ), + 'topic' => array( + 'type' => 'text', + 'not null' => TRUE, + 'description' => t('The topic URL (feed URL) of this subscription.'), + ), + 'mode' => array( + 'type' => 'varchar', + 'length' => 64, + 'not null' => TRUE, + 'default' => '', + 'description' => 'Subscribe or unsubscribe. Empty if subscribed.', + ), + 'post_fields' => array( + 'type' => 'text', + 'not null' => FALSE, + 'description' => 'Fields posted.', + 'serialize' => TRUE, + ), + ), + 'primary key' => array('domain', 'subscriber_id'), + 'indexes' => array( + 'timestamp' => array('timestamp'), + ), + ); + db_create_table($ret, 'feeds_push_subscriptions', $table); + return $ret; +} Index: feeds.module =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.module,v retrieving revision 1.29 diff -u -p -r1.29 feeds.module --- feeds.module 10 Feb 2010 23:49:35 -0000 1.29 +++ feeds.module 18 Feb 2010 16:56:32 -0000 @@ -135,6 +135,13 @@ function feeds_menu() { 'file' => 'feeds.pages.inc', ); } + $items['feeds/pubsubhubbub/notify/%feeds_importer'] = array( + 'page callback' => 'feeds_push_receive', + 'page arguments' => array(3, 4), + 'access callback' => TRUE, + 'file' => 'feeds.pages.inc', + 'type' => MENU_CALLBACK, + ); return $items; } @@ -285,8 +292,9 @@ function feeds_nodeapi(&$node, $op, $for if ($op == 'insert' && feeds_importer($importer_id)->config['import_on_create'] && !isset($node->feeds['suppress_import'])) { feeds_batch_set(t('Importing'), 'import', $importer_id, $node->nid); } - // Add import to scheduler. + // Add import and subscribe to scheduler. feeds_scheduler()->add($importer_id, 'import', $node->nid); + feeds_scheduler()->add($importer_id, 'subscribe', $node->nid); // Add expiry to schedule, in case this is the first feed of this // configuration. feeds_scheduler()->add($importer_id, 'expire'); @@ -294,6 +302,7 @@ function feeds_nodeapi(&$node, $op, $for case 'delete': // Remove feed from scheduler and delete source. feeds_scheduler()->remove($importer_id, 'import', $node->nid); + feeds_scheduler()->remove($importer_id, 'subscribe', $node->nid); feeds_source($importer_id, $node->nid)->delete(); break; } Index: feeds.pages.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.pages.inc,v retrieving revision 1.9 diff -u -p -r1.9 feeds.pages.inc --- feeds.pages.inc 10 Feb 2010 23:49:35 -0000 1.9 +++ feeds.pages.inc 18 Feb 2010 16:56:32 -0000 @@ -88,6 +88,7 @@ function feeds_import_form_submit($form, // Add importer to schedule. feeds_scheduler()->add($form['#importer_id'], 'import'); + feeds_scheduler()->add($form['#importer_id'], 'subscribe'); feeds_scheduler()->add($form['#importer_id'], 'expire'); } @@ -139,3 +140,26 @@ function feeds_delete_tab_form_submit($f $form_state['redirect'] = $form['#redirect']; feeds_batch_set(t('Deleting'), 'clear', $form['#importer_id'], empty($form['#feed_nid']) ? 0 : $form['#feed_nid']); } + +/** + * Handle a PubsubHubbub notification. + */ +function feeds_push_receive($importer, $feed_nid = 0) { + if ($importer->fetcher instanceof FeedsPubSubFetcher) { + // A subscription verification has been sent, verify. + if ($_GET['hub_challenge']) { + $importer->fetcher->pushVerifySubscriptionRequest($_GET, $feed_nid); + } + // No subscription notification has ben sent, we are being notified. + else { + try { + feeds_source($importer->id, $feed_nid)->import(); + } + catch (Exception $e) { + // In case of an error, respond with a 503 Service (temporary) unavailable. + header('HTTP/1.1 503 "Not Found"', null, 503); + exit(); + } + } + } +} \ No newline at end of file Index: feeds.plugins.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.plugins.inc,v retrieving revision 1.4 diff -u -p -r1.4 feeds.plugins.inc --- feeds.plugins.inc 25 Jan 2010 20:03:05 -0000 1.4 +++ feeds.plugins.inc 18 Feb 2010 16:56:32 -0000 @@ -76,6 +76,16 @@ function _feeds_feeds_plugins() { 'path' => $path, ), ); + $info['FeedsPubSubFetcher'] = array( + 'name' => 'Pubsubhubbub Fetcher', + 'description' => 'Receive updates from a PubSubHubbub notification hub.', + 'handler' => array( + 'parent' => 'FeedsHTTPFetcher', + 'class' => 'FeedsPubSubFetcher', + 'file' => 'FeedsPubSubFetcher.inc', + 'path' => $path, + ), + ); $info['FeedsCSVParser'] = array( 'name' => 'CSV parser', 'description' => 'Parse data in Comma Separated Value format.', Index: includes/FeedsImporter.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsImporter.inc,v retrieving revision 1.12 diff -u -p -r1.12 FeedsImporter.inc --- includes/FeedsImporter.inc 11 Feb 2010 00:26:49 -0000 1.12 +++ includes/FeedsImporter.inc 18 Feb 2010 16:56:32 -0000 @@ -104,10 +104,13 @@ class FeedsImporter extends FeedsConfigu switch ($job['callback']) { case 'import': return feeds_source($job['id'], $job['feed_nid'])->import(); - break; case 'expire': return $this->expire(); + case 'subscribe': + feeds_source($job['importer_id'], $job['feed_nid'])->fetcher->subscribe(); + break; } + return FEEDS_BATCH_COMPLETE; } /** @@ -123,6 +126,8 @@ class FeedsImporter extends FeedsConfigu return 3600; } return FEEDS_SCHEDULE_NEVER; + case 'subscribe': + return $this->fetcher->subscriptionPeriod(); } } @@ -130,7 +135,7 @@ class FeedsImporter extends FeedsConfigu * Expose available schedule callbacks. */ public function getScheduleCallbacks() { - return array('import', 'expire'); + return array('import', 'expire', 'subscribe'); } /** Index: libraries/PuSHSubscriber.inc =================================================================== RCS file: libraries/PuSHSubscriber.inc diff -N libraries/PuSHSubscriber.inc --- /dev/null 1 Jan 1970 00:00:00 -0000 +++ libraries/PuSHSubscriber.inc 18 Feb 2010 16:56:32 -0000 @@ -0,0 +1,193 @@ +domain = $domain; + $this->subscriber_id = $subscriber_id; + $this->subscriptions = $subscriptions; + } + + /** + * Subscribe to a given URL. Retrieve 'hub' and 'self' links from feed at $url + * and issue a subscription request to the hub. + * + * @param $url + * The URL of the feed to subscribe to. + * @param $callback_url + * The full URL that hub should invoke when feed changes. + */ + public function subscribe($url, $callback_url) { + // Fetch document, find rel=hub and rel=self. + // If present, issue subscription request. + $request = curl_init($url); + curl_setopt($request, CURLOPT_FOLLOWLOCATION, TRUE); + curl_setopt($request, CURLOPT_RETURNTRANSFER, TRUE); + $data = curl_exec($request); + if (curl_getinfo($request, CURLINFO_HTTP_CODE) == 200) { + $xml = new SimpleXMLElement($data); + $xml->registerXPathNamespace('atom', 'http://www.w3.org/2005/Atom'); + if ($hub = @current($xml->xpath("/atom:feed/atom:link[attribute::rel='hub']"))) { + $hub = (string) $hub->attributes()->href; + } + if ($self = @current($xml->xpath("/atom:feed/atom:link[attribute::rel='self']"))) { + $self = (string) $self->attributes()->href; + } + if ($hub && $self) { + $this->request($hub, $self, 'subscribe', $callback_url); + } + } + curl_close($request); + } + + /** + * @todo Unsubscribe from a hub. + * @todo Make sure we unsubscribe with the correct topic URL as it can differ + * from the initial subscription URL. + * + * @param $topic_url + * The URL of the topic to unsubscribe from. + * @param $callback_url + * The callback to unsubscribe. + */ + public function unsubscribe($topic_url, $callback_url) { + $this->subscriptions->delete($this->domain, $this->subscriber_id); + } + + /** + * Issue a subscribe or unsubcribe request to a PubsubHubbub hub. + * + * @param $hub + * The URL of the hub's subscription endpoint. + * @param $topic + * The topic URL of the feed to subscribe to. + * @param $mode + * 'subscribe' or 'unsubscribe'. + * @param $callback_url + * The subscriber's notifications callback URL. + * + * Compare to http://pubsubhubbub.googlecode.com/svn/trunk/pubsubhubbub-core-0.2.html#anchor5 + * + * @todo don't use url() + */ + protected function request($hub, $topic, $mode, $callback_url) { + $post_fields = array( + 'hub.callback' => $callback_url, + 'hub.mode' => $mode, + 'hub.topic' => $topic, + 'hub.verify' => 'sync', + 'hub.lease_seconds' => '', // Permanent subscription. + 'hub.secret' => '', // @todo + 'hub.verify_token' => md5(session_id() . rand()), + ); + $this->subscriptions->save($this->domain, $this->subscriber_id, $hub, $topic, $mode, $post_fields); + // Issue subscription request. + $request = curl_init($hub); + curl_setopt($request, CURLOPT_POST, TRUE); + curl_setopt($request, CURLOPT_POSTFIELDS, $post_fields); + curl_exec($request); + if (in_array(curl_getinfo($request, CURLINFO_HTTP_CODE), array(202, 204))) { + $this->subscriptions->save($this->domain, $this->subscriber_id, $hub, $topic, ''); + } + else { + drupal_set_message(t('Error subscribing to PubsubHubbub hub.'), 'error'); + } + curl_close($request); + } + + /** + * Verify subscription request. + * + * @todo: break out subscription handling in its own abstract class with a + * Drupal implementer. + */ + public function verifySubscriptionRequest($verify_fields) { + if ($subscription = $this->subscriptions->load($this->domain, $this->subscriber_id)) { + if ($verify_fields['hub_verify_token'] == $subscription['post_fields']['hub.verify_token']) { + header('HTTP/1.1 200 "Found"', null, 200); + print $verify_fields['hub_challenge']; + exit(); + } + } + header('HTTP/1.1 404 "Not Found"', null, 404); + exit(); + } + + /** + * Receive a notification. + */ + public function receive() { + if ($_SERVER['REQUEST_METHOD'] == 'POST') { + return file_get_contents('php://input'); + } + return FALSE; + } +} + +/** + * Implement to provide a storage backend for subscriptions. + */ +interface PuSHSubscriptionsInterface { + /** + * Save a subscription. + * + * @param $domain + * A string that defines the domain in which the subscriber_id is unique. + * @param $subscriber_id + * A unique numeric subscriber id. + * @param $hub + * The URL of the hub endpoint. + * @param $topic + * The topic to subscribe to. + * @param $mode + * The temporary mode of the subscription. + * @param $post_fields + * An array of the fields posted to the hub. + */ + public function save($domain, $subscriber_id, $hub, $topic, $mode, $post_fields = array()); + + /** + * Load a subscription. + */ + public function load($domain, $subscriber_id); + + /** + * Delete a subscription. + */ + public function delete($domain, $subscriber_id); +} Index: plugins/FeedsFetcher.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsFetcher.inc,v retrieving revision 1.4 diff -u -p -r1.4 FeedsFetcher.inc --- plugins/FeedsFetcher.inc 20 Dec 2009 23:54:44 -0000 1.4 +++ plugins/FeedsFetcher.inc 18 Feb 2010 16:56:32 -0000 @@ -26,4 +26,27 @@ abstract class FeedsFetcher extends Feed * caches pertaining to this source. */ public function clear(FeedsSource $source) {} + + /** + * Subscribe to a source. Only implement if fetcher requires subscription. + * + * @param FeedsSource $source + * Source information for this subscription. + */ + public function subscribe(FeedsSource $source) {} + + /** + * Unsubscribe from a source. Only implement if fetcher requires subscription. + * + * @param FeedsSource $source + * Source information for unsubscribing. + */ + public function unsubscribe(FeedsSource $source) {} + + /** + * Indicate the time within which a subscription needs to be renewed. + */ + public function subscriptionPeriod() { + return FEEDS_SCHEDULE_NEVER; + } } Index: plugins/FeedsPubSubFetcher.inc =================================================================== RCS file: plugins/FeedsPubSubFetcher.inc diff -N plugins/FeedsPubSubFetcher.inc --- /dev/null 1 Jan 1970 00:00:00 -0000 +++ plugins/FeedsPubSubFetcher.inc 18 Feb 2010 16:56:32 -0000 @@ -0,0 +1,196 @@ +raw = $raw; + parent::__construct(); + } + + /** + * Override FeedsHTTPBatch::getRaw() and simply return the raw content + * populated on instnatiation. + */ + function getRaw() { + return $this->raw; + } +} + +/** + * Publish/Subscribe fetcher. Supports at the moment only PubSubHubbub (PuSH). + */ +class FeedsPubSubFetcher extends FeedsHTTPFetcher { + + /** + * Implementation of FeedsFetcher::fetch(). + */ + public function fetch(FeedsSource $source) { + $source_config = $source->getConfigFor($this); + // Handle fat ping if present, otherwise pass up to HTTP fetcher. + if ($raw = feeds_push_subscriber($this->id, $source->feed_nid)->receive()) { + return new FeedsPubSubBatch($raw); + } + return parent::fetch($source); + } + + /** + * Override sourceSave() - subscribe to hub. + */ + public function sourceSave(FeedsSource $source) { + $this->subscribe($source); + } + + /** + * Override sourceDelete() - unsubscribe from hub. + */ + public function sourceDelete(FeedsSource $source) { + $this->unsubscribe($source); + } + + /** + * Implement FeedsFetcher::subscribe() - subscribe to hub. + */ + public function subscribe(FeedsSource $source) { + $source_config = $source->getConfigFor($this); + $url = $source_config['source']; + if (module_exists('keyauth') && $this->config['use_keyauth']) { + keyauth_include(); + $url = keyauth_sign_url($this->config['keyauth_public'], $url); + } + feeds_push_subscriber($this->id, $source->feed_nid)->subscribe($url, url('feeds/pubsubhubbub/notify/'. $this->id .'/'. $source->feed_nid, array('absolute' => TRUE))); + } + + /** + * Implement FeedsFetcher::unsubscribe() - unsubscribe from hub. + */ + public function unsubscribe(FeedsSource $source) { + $source_config = $source->getConfigFor($this); + feeds_push_subscriber($this->id, $source->feed_nid)->unsubscribe($source_config['source'], url('feeds/pubsubhubbub/notify/'. $this->id .'/'. $source->feed_nid, array('absolute' => TRUE))); + } + + /** + * Implement FeedsFetcher::subscriptionPeriod(). + * Indicate how often a subscription needs to be renewed. + * + * @todo subscription_period should actually be retrieved from the hub's + * response to a subscription. This will mean a different subscription + * period per source, hence a major change to FeedsScheduler which is + * currently assuming a fixed period per task! + */ + public function subscriptionPeriod() { + return $this->config['subscription_period']; + } + + /** + * Return defaults for configuration. + */ + public function configDefaults() { + $defaults = parent::configDefaults(); + return $defaults + array( + 'subscription_period' => 3600*24, // Renew subscription in 24 hours. + ); + } + + /** + * Override parent::configForm(). + */ + public function configForm(&$form_state) { + $form = parent::configForm($form_state); + $period = drupal_map_assoc(array(0, 900, 1800, 3600, 10800, 21600, 43200, 86400, 259200, 604800, 2419200), 'format_interval'); + $period[FEEDS_SCHEDULE_NEVER] = t('Never renew'); + $period[0] = t('Renew as often as possible'); + $form['subscription_period'] = array( + '#type' => 'select', + '#title' => t('Renew subscription after'), + '#options' => $period, + '#description' => t('This is the minimum time that must elapse before a subscription is renewed.'), + '#default_value' => $this->config['subscription_period'], + ); + return $form; + } + + /** + * Verify a PubSubHubbub subscription request. + */ + public function pushVerifySubscriptionRequest($verify_fields, $feed_nid) { + feeds_push_subscriber($this->id, $feed_nid)->verifySubscriptionRequest($verify_fields); + } +} + +/** + * Create a PubSubHubbub subscriber. + * + * @return PushSubscriber + * A PushSubscriber object. + */ +function feeds_push_subscriber($id, $subscriber_id) { + return PushSubscriber::instance($id, $subscriber_id, PuSHSubscriptions::instance()); +} + +/** + * Implement a PuSHSubscriptionsInterface. + */ +class PuSHSubscriptions implements PuSHSubscriptionsInterface { + /** + * Singleton. + */ + public function instance() { + static $subscriptions; + if (empty($subscriptions)) { + $subscriptions = new PuSHSubscriptions(); + } + return $subscriptions; + } + + /** + * Protect constructor. + */ + protected function __construct() { + } + + /** + * Save a subscription. + */ + public function save($domain, $subscriber_id, $hub, $topic, $mode, $post_fields = array()) { + $save = array( + 'domain' => $domain, + 'subscriber_id' => $subscriber_id, + 'hub' => $hub, + 'topic' => $topic, + 'mode' => $mode, + 'post_fields' => $post_fields, + 'timestamp' => time(), + ); + $this->delete($domain, $subscriber_id); + drupal_write_record('feeds_push_subscriptions', $save); + } + + /** + * Load a subscription. + */ + public function load($domain, $subscriber_id) { + if ($request = db_fetch_array(db_query("SELECT * FROM {feeds_push_subscriptions} WHERE domain = '%s' AND subscriber_id = %d", $domain, $subscriber_id))) { + $request['post_fields'] = unserialize($request['post_fields']); + } + return $request; + } + + /** + * Delete a subscription. + */ + public function delete($domain, $subscriber_id) { + db_query("DELETE FROM {feeds_push_subscriptions} WHERE domain = '%s' AND subscriber_id = %d", $domain, $subscriber_id); + } +}