3 * StatusNet - the distributed open-source microblogging tool
4 * Copyright (C) 2010, StatusNet, Inc.
6 * This program is free software: you can redistribute it and/or modify
7 * it under the terms of the GNU Affero General Public License as published by
8 * the Free Software Foundation, either version 3 of the License, or
9 * (at your option) any later version.
11 * This program is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 * GNU Affero General Public License for more details.
16 * You should have received a copy of the GNU Affero General Public License
17 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 * PuSH feed subscription record
23 * @author Brion Vibber <brion@status.net>
25 class HubSub extends Memcached_DataObject
27 public $__table = 'hubsub';
29 public $hashkey; // sha1(topic . '|' . $callback); (topic, callback) key is too long for myisam in utf8
39 public /*static*/ function staticGet($topic, $callback)
41 return parent::staticGet(__CLASS__, 'hashkey', self::hashkey($topic, $callback));
44 protected static function hashkey($topic, $callback)
46 return sha1($topic . '|' . $callback);
50 * return table definition for DB_DataObject
52 * DB_DataObject needs to know something about the table to manipulate
53 * instances. This method provides all the DB_DataObject needs to know.
55 * @return array array of column definitions
60 return array('hashkey' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
61 'topic' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
62 'callback' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
63 'secret' => DB_DATAOBJECT_STR,
64 'lease' => DB_DATAOBJECT_INT,
65 'sub_start' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
66 'sub_end' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
67 'created' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL,
68 'modified' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL);
71 static function schemaDef()
73 return array(new ColumnDef('hashkey', 'char',
77 new ColumnDef('topic', 'varchar',
81 new ColumnDef('callback', 'varchar',
83 new ColumnDef('secret', 'text',
85 new ColumnDef('lease', 'int',
87 new ColumnDef('sub_start', 'datetime',
89 new ColumnDef('sub_end', 'datetime',
91 new ColumnDef('created', 'datetime',
93 new ColumnDef('modified', 'datetime',
99 return array_keys($this->keyTypes());
102 function sequenceKey()
104 return array(false, false, false);
108 * return key definitions for DB_DataObject
110 * DB_DataObject needs to know about keys that the table has; this function
113 * @return array key definitions
118 return array('hashkey' => 'K');
122 * Validates a requested lease length, sets length plus
123 * subscription start & end dates.
125 * Does not save to database -- use before insert() or update().
127 * @param int $length in seconds
129 function setLease($length)
131 assert(is_int($length));
137 // We want to garbage collect dead subscriptions!
139 } elseif( $length < $min) {
141 } else if ($length > $max) {
145 $this->lease = $length;
146 $this->start_sub = common_sql_now();
147 $this->end_sub = common_sql_date(time() + $length);
151 * Schedule a future verification ping to the subscriber.
152 * If queues are disabled, will be immediate.
154 * @param string $mode 'subscribe' or 'unsubscribe'
155 * @param string $token hub.verify_token value, if provided by client
157 function scheduleVerify($mode, $token=null, $retries=null)
159 if ($retries === null) {
160 $retries = intval(common_config('ostatus', 'hub_retries'));
162 $data = array('sub' => clone($this),
165 'retries' => $retries);
166 $qm = QueueManager::get();
167 $qm->enqueue($data, 'hubconf');
171 * Send a verification ping to subscriber, and if confirmed apply the changes.
172 * This may create, update, or delete the database record.
174 * @param string $mode 'subscribe' or 'unsubscribe'
175 * @param string $token hub.verify_token value, if provided by client
176 * @throws ClientException on failure
178 function verify($mode, $token=null)
180 assert($mode == 'subscribe' || $mode == 'unsubscribe');
182 $challenge = common_good_rand(32);
183 $params = array('hub.mode' => $mode,
184 'hub.topic' => $this->topic,
185 'hub.challenge' => $challenge);
186 if ($mode == 'subscribe') {
187 $params['hub.lease_seconds'] = $this->lease;
189 if ($token !== null) {
190 $params['hub.verify_token'] = $token;
193 // Any existing query string parameters must be preserved
194 $url = $this->callback;
195 if (strpos('?', $url) !== false) {
200 $url .= http_build_query($params, '', '&');
202 $request = new HTTPClient();
203 $response = $request->get($url);
204 $status = $response->getStatus();
206 if ($status >= 200 && $status < 300) {
207 common_log(LOG_INFO, "Verified $mode of $this->callback:$this->topic");
209 throw new ClientException("Hub subscriber verification returned HTTP $status");
212 $old = HubSub::staticGet($this->topic, $this->callback);
213 if ($mode == 'subscribe') {
217 $ok = $this->insert();
219 } else if ($mode == 'unsubscribe') {
223 // That's ok, we're already unsubscribed.
229 * Insert wrapper; transparently set the hash key from topic and callback columns.
230 * @return mixed success
234 $this->hashkey = self::hashkey($this->topic, $this->callback);
235 $this->created = common_sql_now();
236 $this->modified = common_sql_now();
237 return parent::insert();
241 * Update wrapper; transparently update modified column.
242 * @return boolean success
244 function update($old=null)
246 $this->modified = common_sql_now();
247 return parent::update($old);
251 * Schedule delivery of a 'fat ping' to the subscriber's callback
252 * endpoint. If queues are disabled, this will run immediately.
254 * @param string $atom well-formed Atom feed
255 * @param int $retries optional count of retries if POST fails; defaults to hub_retries from config or 0 if unset
257 function distribute($atom, $retries=null)
259 if ($retries === null) {
260 $retries = intval(common_config('ostatus', 'hub_retries'));
263 // We dare not clone() as when the clone is discarded it'll
264 // destroy the result data for the parent query.
265 // @fixme use clone() again when it's safe to copy an
266 // individual item from a multi-item query again.
267 $sub = HubSub::staticGet($this->topic, $this->callback);
268 $data = array('sub' => $sub,
270 'retries' => $retries);
271 common_log(LOG_INFO, "Queuing PuSH: $this->topic to $this->callback");
272 $qm = QueueManager::get();
273 $qm->enqueue($data, 'hubout');
277 * Send a 'fat ping' to the subscriber's callback endpoint
278 * containing the given Atom feed chunk.
280 * Determination of which items to send should be done at
281 * a higher level; don't just shove in a complete feed!
283 * @param string $atom well-formed Atom feed
284 * @throws Exception (HTTP or general)
288 $headers = array('Content-Type: application/atom+xml');
290 $hmac = hash_hmac('sha1', $atom, $this->secret);
291 $headers[] = "X-Hub-Signature: sha1=$hmac";
295 common_log(LOG_INFO, "About to push feed to $this->callback for $this->topic, HMAC $hmac");
297 $request = new HTTPClient();
298 $request->setBody($atom);
299 $response = $request->post($this->callback, $headers);
301 if ($response->isOk()) {
304 throw new Exception("Callback returned status: " .
305 $response->getStatus() .
307 trim($response->getBody()));