moksha/api/streams/datastream.py | 4 - moksha/api/widgets/feed/feed.py | 6 - moksha/api/widgets/feed/live.py | 6 + moksha/hub/hub.py | 19 +++- moksha/hub/reactor.py | 6 - moksha/lib/cache.py | 11 ++ moksha/streams/feed/feedstream.py | 152 +++++++++++++++++--------------------- start-moksha | 1 8 files changed, 106 insertions(+), 99 deletions(-)
New commits: commit 1dd06e1be96beb0088e58e06e9e775063e14b6ec Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 23:11:08 2009 -0500
Fix a bug on platform detection on Mac OSX
diff --git a/moksha/hub/reactor.py b/moksha/hub/reactor.py index c92ab1b..8e919ce 100644 --- a/moksha/hub/reactor.py +++ b/moksha/hub/reactor.py @@ -26,12 +26,12 @@ try: if 'linux' in sys.platform: from twisted.internet import epollreactor epollreactor.install() - elif 'win' in sys.platform: - from twisted.internet import iocpreactor - iocpreactor.install() elif 'freebsd' in sys.platform or 'darwin' in sys.platform: from twisted.internet import kqreactor kqreactor.install() + elif 'win' in sys.platform: + from twisted.internet import iocpreactor + iocpreactor.install() except AssertionError: # reactor already installed pass
commit ad33591fa573ea41a1dd083b21c9895dfc9a339b Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:24:27 2009 -0500
Have our MokshaHub.send_message support lists of topics.
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py index e41ffa3..d0d9f3e 100644 --- a/moksha/hub/hub.py +++ b/moksha/hub/hub.py @@ -66,17 +66,22 @@ class MokshaHub(StompHub, AMQPHub): def send_message(self, topic, message, jsonify=True): """ Send a message to a specific topic.
- :topic: The stop to send the message to + :topic: A topic or list of topics to send the message to. :message: The message body. Can be a string, list, or dict. :jsonify: To automatically encode non-strings to JSON
""" - if jsonify and not isinstance(message, basestring): - message = json.encode(message) - if self.amqp_broker: - AMQPHub.send_message(self, topic, message, routing_key=topic) - elif self.stomp_broker: - StompHub.send_message(self, topic, message) + if not isinstance(topic, list): + topics = [topic] + else: + topics = topic + for topic in topics: + if jsonify and not isinstance(message, basestring): + message = json.encode(message) + if self.amqp_broker: + AMQPHub.send_message(self, topic, message, routing_key=topic) + elif self.stomp_broker: + StompHub.send_message(self, topic, message)
def close(self): if self.amqp_broker:
commit ce3aadd4a503b9b374d3345e75d5968b67c910d8 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:22:06 2009 -0500
Send our feed messages to two topics for now
diff --git a/moksha/streams/feed/feedstream.py b/moksha/streams/feed/feedstream.py index eb72445..11496f7 100644 --- a/moksha/streams/feed/feedstream.py +++ b/moksha/streams/feed/feedstream.py @@ -200,8 +200,8 @@ class FeederProtocol(object): entry_title = entry.get('title', '[No Title]') channel_link = entry.get('channel', {'link': addr})['link'] if entry['title'] not in oldtitles: - log.info('New feed entry found: %' % entry['title']) - self.hub.send_message('feed.%s' % channel_link, + log.info('New feed entry found: %s' % entry['title']) + self.hub.send_message(['feed.%s' % channel_link, 'feeds'], {'title': entry_title, 'link': entry.get('link')})
def get_page(self, data, args):
commit c06cd189c3a077b613a0f5b5b38f4d2388b39329 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:15:32 2009 -0500
SymbolType has been giving people problems
diff --git a/start-moksha b/start-moksha index 269bf9d..fdb7875 100755 --- a/start-moksha +++ b/start-moksha @@ -14,6 +14,7 @@ if [ ! $VIRTUAL_ENV ]; then easy_install -i http://www.turbogears.org/2.0/downloads/current/index tg.devtools pushd moksha/widgetbrowser; python setup.py develop; popd easy_install "tw.jquery==dev" + easy_install SymbolType else source tg2env/bin/activate fi
commit 328b015e5887d55ca4391421c3bff602c38ce6d5 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:15:20 2009 -0500
Run the FeedStreamer every 10 minutes
diff --git a/moksha/streams/feed/feedstream.py b/moksha/streams/feed/feedstream.py index 24e6a99..eb72445 100644 --- a/moksha/streams/feed/feedstream.py +++ b/moksha/streams/feed/feedstream.py @@ -263,7 +263,7 @@ class FeedStream(PollingDataStream): then Moksha will automatically handle polling it. Upon new entries, AMQP messages will be sent to the `feeds.$URL` queue. """ - frequency = timedelta(minutes=5) + frequency = timedelta(minutes=10)
def poll(self): """ Poll all feeds in our feed cache """
commit 9af4dde6ddb10852860ab090a4e55b293756fee8 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:13:09 2009 -0500
Many FeedStreamer improvements, including the sending of notifications for new entries
diff --git a/moksha/streams/feed/feedstream.py b/moksha/streams/feed/feedstream.py index 0225e11..24e6a99 100644 --- a/moksha/streams/feed/feedstream.py +++ b/moksha/streams/feed/feedstream.py @@ -30,13 +30,16 @@ from twisted.web import client from twisted.web.client import HTTPPageGetter, HTTPClientFactory from twisted.internet import reactor, protocol, defer, ssl from twisted.web import error +from twisted.internet import protocol, defer + from datetime import timedelta, datetime from feedcache import Cache from shove import Shove from tg import config
+from moksha.hub import MokshaHub +from moksha.hub.http import conditional_get_page from moksha.api.streams import PollingDataStream -from moksha.lib.helpers import trace
log = logging.getLogger('moksha.hub')
@@ -46,12 +49,12 @@ feed_cache = Cache(feed_storage) class ConditionalHTTPPageGetter(HTTPPageGetter):
def handleStatus_200(self): - """ If we're good, try recording the last-modified header """ + """ Attempt to save the last-modified header """ if self.headers.has_key('last-modified'): self.factory.lastModified(self.headers['last-modified'])
def handleStatus_304(self): - """ Close connection """ + """ Close the connection """ self.factory.notModified() self.transport.loseConnection()
@@ -61,8 +64,7 @@ class ConditionalHTTPClientFactory(HTTPClientFactory): protocol = ConditionalHTTPPageGetter
def __init__(self, url, method='GET', postdata=None, headers=None, - agent="Moksha Feed Streamer", timeout=0, cookies=None, - followRedirect=1): + agent=None, timeout=0, cookies=None, followRedirect=1):
self.url = url
@@ -91,9 +93,9 @@ class ConditionalHTTPClientFactory(HTTPClientFactory):
try: feed_storage[self.url] = (t, parsed_feed) - except: + except Exception, e: log.error("Unable to store parsed_feed: %r" % parsed_feed) - raise + log.exception(e)
def notModified(): if self.waiting: @@ -119,11 +121,10 @@ class FeederProtocol(object):
def __init__(self): self.parsed = 1 - self.with_errors = 0 - self.error_list = [] + self.hub = MokshaHub()
def is_cached(self, site): - already_got = feed_storage.get(site[0]) + already_got = feed_storage.get(site) if already_got: elapsed_time = time.time() - already_got[0] if elapsed_time < self.max_age: @@ -136,8 +137,6 @@ class FeederProtocol(object): def on_error(self, traceback, extra_args): log.error(extra_args) log.exception(traceback) - self.with_errors += 1 - self.error_list.append(extra_args)
def get_feed_from_cache(self, data, key=None): """ Return feed data from the cache based on a given ``key`` """ @@ -162,72 +161,70 @@ class FeederProtocol(object): feed_storage[addr] = (time.time(), feed) return feed
- def process_feed(self, parsed_feed, addr): + def get_feed(self, addr): + return feed_storage[addr][1] + + def process_feed(self, parsed_feed, addr, olddata): + """ Process the parsed feed. + + If `olddata` is provided, this method will look for new feed entries, + and send notifications to the `feed.$FEED_URL` MokshaHub Topic. + + :param parsed_feed: A parsed :mod:`feedcache` feed + :param addr: The URL of the feed + :param olddata: The cached feed data + """ if not parsed_feed: log.error("Cannot process %r feed for %s" % (parsed_feed, addr)) return + chan = parsed_feed.get('channel', None) if chan: log.debug(chan.get('title', '')) - # print chan.get('link', '') - # #print chan.get('tagline', '') - # print chan.get('description','') - #print "-"*20 - #items = parsed_feed.get('items', None) - #if items: - # for item in items: - # print '\tTitle: ', item.get('title','') - # print '\tDate: ', item.get('date', '') - # print '\tLink: ', item.get('link', '') - # print '\tDescription: ', item.get('description', '') - # print '\tSummary: ', item.get('summary','') - # print "-"*20 - #print "got",addr - #print "="*40 - return parsed_feed - - ## @@ post processing - for url in feed_storage: - olddata = feed_storage[url][1].copy() - newdata = feed_cache.fetch(url) #, force_update=True) + + # Previous data provided; look for new entries. + if olddata: + oldtitles = [entry.get('title') for entry in olddata['entries']] + new_entries = parsed_feed.get('entries', [{}]) + if not len(new_entries): + log.warning('Feed contains empty entries: %s' % addr) + return + # If there are no new entries, move on... - if newdata['entries'][0]['title'] == olddata['entries'][0]['title']: - log.debug('No feed changes for %s' % url) - del(olddata) - continue - oldtitles = [entry['title'] for entry in olddata['entries']] - for entry in newdata['entries'][::-1]: - # If we haven't seen this entry yet... + newtitle = new_entries[0].get('title', None) + if newtitle == oldtitles[0]: + return + + # Send notifications for each new entry + for entry in new_entries[::-1]: + entry_title = entry.get('title', '[No Title]') + channel_link = entry.get('channel', {'link': addr})['link'] if entry['title'] not in oldtitles: - # Send a message to the feed's topic - print "New entry:", entry['title'] - self.send_message('feed.%s' % entry.channel.link, - {'title': entry['title'], 'link': entry['link']}) - else: - print "Dupe entry", entry['title'] - + log.info('New feed entry found: %' % entry['title']) + self.hub.send_message('feed.%s' % channel_link, + {'title': entry_title, 'link': entry.get('link')})
def get_page(self, data, args): return conditional_get_page(args, timeout=self.timeout)
- def print_status(self, data=None): - pass - def start(self, data=None): d = defer.succeed(True) for feed in data: + olddata = None if self.is_cached(feed): - d.addCallback(self.get_feed_from_cache, feed[0]) - d.addErrback(self.on_error, (feed[0], 'fetching from cache')) + d.addCallback(self.get_feed_from_cache, feed) + d.addErrback(self.on_error, (feed, 'fetching from cache')) else: - d.addCallback(self.get_page, feed[0]) - d.addErrback(self.on_error, (feed[0], 'fetching')) - d.addCallback(self.parse_feed, feed[0]) - d.addErrback(self.on_error, (feed[0], 'parsing')) - d.addCallback(self.store_feed, feed[0]) - d.addErrback(self.on_error, (feed[0], 'storing')) - d.addCallback(self.process_feed, feed[0]) - d.addErrback(self.on_error, (feed[0], 'processing')) + d.addCallback(self.get_page, feed) + d.addErrback(self.on_error, (feed, 'fetching')) + d.addCallback(self.parse_feed, feed) + d.addErrback(self.on_error, (feed, 'parsing')) + olddata = self.get_feed(feed) + d.addCallback(self.store_feed, feed) + d.addErrback(self.on_error, (feed, 'storing')) + d.addCallback(self.process_feed, feed, olddata) + d.addErrback(self.on_error, (feed, 'processing')) + del(olddata) return d
@@ -266,35 +263,24 @@ class FeedStream(PollingDataStream): then Moksha will automatically handle polling it. Upon new entries, AMQP messages will be sent to the `feeds.$URL` queue. """ - #frequency = timedelta(minutes=15) - frequency = 5 # seconds - running = False + frequency = timedelta(minutes=5)
def poll(self): - """ Poll all known feeds. + """ Poll all feeds in our feed cache """ + log.debug('FeedStream.poll()')
- - Iterate over all feeds in our global moksha feed cache.. - :Warning: the MokshaHub will not use this unless both it and the - Moksha WSGI app are using the same `feed_cache` database. + feeds = set() + for feed in feed_storage.keys(): + feeds.add(str(feed))
- - Keep feed caches fresh. + # Read in all feeds from the `feeds.txt` file for testing... + if os.path.isfile('feeds.txt'): + feed_list = file('feeds.txt') + for feed in feed_list.readlines(): + feeds.add(str(feed.strip()))
- - Send messages to topics for new entries - `feed.$NAME` topic ? - `tag.category - """ - # @@ Only run once... for testing purposes - if self.running: - print "Feed streamer already running" - return - self.running = True - - log.debug('FeedStream.poll()') - feeds = map(lambda x: x.strip(), file('feeds.txt').readlines()) - feeds = [(feed, '') for feed in feeds] f = FeederFactory() f.start(addresses=feeds) - return
def stop(self): feed_storage.close()
commit 452d59484e33527480838d89024c5440f850ea76 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:10:36 2009 -0500
Have our LiveFeedWidget automatically subscribe to the `feed.$URL` topic.
diff --git a/moksha/api/widgets/feed/live.py b/moksha/api/widgets/feed/live.py index 4ceda47..3b869fb 100644 --- a/moksha/api/widgets/feed/live.py +++ b/moksha/api/widgets/feed/live.py @@ -38,3 +38,9 @@ class LiveFeedWidget(LiveWidget): }); """ feed = Feed() + topic = None + + def update_params(self, d): + super(LiveFeedWidget, self).update_params(d) + if not d.topic: + d.topic = 'feed.%s' % d.url
commit ecfe0d30615b955c96e110669b475815359f0b28 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:09:50 2009 -0500
Make our Feed widget handle dicts as feed data
diff --git a/moksha/api/widgets/feed/feed.py b/moksha/api/widgets/feed/feed.py index 3818bc1..104f564 100644 --- a/moksha/api/widgets/feed/feed.py +++ b/moksha/api/widgets/feed/feed.py @@ -74,9 +74,9 @@ class Feed(Widget): feed_storage = Shove('sqlite:///feeds.db', compress=True) feed_cache = Cache(feed_storage) feed = feed_cache.fetch(url) - if not (200 <= feed.status < 400): + if not (200 <= feed.get('status', 200) < 400): log.warning('Got %s status from %s: %s' % ( - feed.status, url, feed.headers.get('status'))) + feed['status'], url, feed.headers.get('status'))) d['title'] = feed.headers.get('status') d['link'] = feed.feed.get('link') return @@ -87,7 +87,7 @@ class Feed(Widget): except AttributeError: d['title'] = 'Unable to parse feed' return - for i, entry in enumerate(feed.entries): + for i, entry in enumerate(feed.get('entries', [])): entry['uid'] = '%s_%d' % (self.id, i) entry['link'] = entry.get('link') yield entry
commit 0f0da20d8bf270d9c9f329b4ed0bbba22de2d109 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 22:09:33 2009 -0500
Make our PollingDataStreamers start up right away
diff --git a/moksha/api/streams/datastream.py b/moksha/api/streams/datastream.py index 5b476f1..b8c4756 100644 --- a/moksha/api/streams/datastream.py +++ b/moksha/api/streams/datastream.py @@ -31,7 +31,7 @@ class PollingDataStream(DataStream): """ frequency = None # Either a timedelta object, or the number of seconds
- def __init__(self): + def __init__(self, now=True): super(PollingDataStream, self).__init__() self.timer = LoopingCall(self.poll) if isinstance(self.frequency, timedelta): @@ -41,7 +41,7 @@ class PollingDataStream(DataStream): else: seconds = self.frequency log.debug("Setting a %s second timers" % seconds) - self.timer.start(seconds, now=False) + self.timer.start(seconds, now=now)
def poll(self): raise NotImplementedError
commit d0fc5a6acc0a81e58b324416d9eab33c4a982005 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 19:59:33 2009 -0500
Remove an unnecessary method
diff --git a/moksha/lib/cache.py b/moksha/lib/cache.py index 64d710d..00b6c70 100644 --- a/moksha/lib/cache.py +++ b/moksha/lib/cache.py @@ -7,9 +7,6 @@ class Cache(object): """ A memcached-specific caching interface """
def __init__(self, url, timeout=None, prefix=None): - self.create_memcached_cache() - - def create_memcached_cache(self): try: import memcache except ImportError:
commit 9f1dda5f09db2453166c0a35615815b03d7e1ef8 Author: Luke Macken lmacken@redhat.com Date: Tue Feb 17 19:10:22 2009 -0500
Gracefully handle not having python-memcached installed
diff --git a/moksha/lib/cache.py b/moksha/lib/cache.py index 20536c1..64d710d 100644 --- a/moksha/lib/cache.py +++ b/moksha/lib/cache.py @@ -1,10 +1,22 @@ -import memcache +import logging +log = logging.getLogger(__name__) + from moksha.exc import CacheBackendException
class Cache(object): """ A memcached-specific caching interface """
def __init__(self, url, timeout=None, prefix=None): + self.create_memcached_cache() + + def create_memcached_cache(self): + try: + import memcache + except ImportError: + log.warning('Cannot import the `memcache` module. Install the ' + 'python-memcached package to enable Mokshas memcached ' + 'integration.') + return self.mc = memcache.Client([url]) self.timeout = timeout or 300 self.prefix = prefix
moksha-commits@lists.fedorahosted.org