<commit_before>from array import array
from contextlib import closing
import sys
import time

from javax.net.ssl import SSLContext, TrustManager, X509TrustManager
from java.net import URL

from org.apache.abdera import Abdera


# Modified from http://tech.pedersen-live.com/2010/10/trusting-all-certificates-in-jython/
# would be best to manage the trust store for something real

class TrustAllX509TrustManager(X509TrustManager):
    """Define a custom TrustManager which will blindly accept all certificates"""
    def checkClientTrusted(self, chain, auth):
        pass

    def checkServerTrusted(self, chain, auth):
        pass

    def getAcceptedIssuers(self):
        return None


def trust_all():
    trust_managers = array(TrustManager, [TrustAllX509TrustManager()])
    trust_all_context = SSLContext.getInstance("SSL")
    trust_all_context.init(None, trust_managers, None)
    SSLContext.setDefault(trust_all_context)


def read_events(feed_url, mark_id=None, read_back=200, poll_time=5.):
    # a given read of the feed is going to be reversed; use the urn as the mark
    count = 0
    abdera = Abdera()
    parser = abdera.getParser()

    def read_chunks(last_id):
        chunk_count = 0
        url = feed_url
        while chunk_count < read_back:
            with closing(URL(url).openStream()) as f:
                print >> sys.stderr, "Reading feed", url
                doc = parser.parse(f)
                feed = doc.getRoot()
                url = str(feed.getLinks("next")[0].href)
                for entry in feed.entries:
                    if last_id == entry.id:
                        return
                    yield entry
                    chunk_count += 1

    # combine loops together FIXME!
    last_id = None
    for entry in reversed(list(read_chunks(mark_id))):
        last_id = entry.id
        count += 1
        yield entry
    print >> sys.stderr, "Read", count, "events so far"
    while True:
        print >> sys.stderr, "Sleeping", poll_time, "then read to", last_id
        time.sleep(poll_time)
        for entry in reversed(list(read_chunks(last_id))):
            last_id = entry.id
            count += 1
            yield entry
        print >> sys.stderr, "Read", count, "events so far"


def main():
    trust_all()
    url = "https://atom.prod.dfw1.us.ci.rackspace.net/nova/events"
    last_updated = None
    seen = set()
    count = 0
    for event in read_events(url):
        if last_updated and event.updated < last_updated:
            print >> sys.stderr, "Out of order!!!!"
        if event.id in seen:
            print >> sys.stderr, "Duplicated", event.id
            continue
        print count, event.id, event.updated, [cat.term for cat in event.categories]

        last_updated = event.updated
        seen.add(event.id)
        
        if count >= 1000:
            break

        count += 1

main()<commit_msg>Initial version of nontransactional spout<commit_after>from collections import deque
from contextlib import closing
import time

from backtype.storm.topology.base import BaseRichSpout
from backtype.storm.tuple import Fields, Values
from java.net import URL
from org.apache.abdera import Abdera
from org.slf4j import LoggerFactory

from clamp import PackageProxy
from romper.trust import trust_all_certificates


# still need to implement consistent hashing if this will be used at large scale
# on the other hand, currently we see 1 event/s/data center for status updates, which can be readily handled
# with just one instance of the spout

# Make transactional; see https://github.com/nathanmarz/storm/wiki/Trident-state
# and https://github.com/nathanmarz/storm/wiki/Trident-spouts
# specifically opaque transactional support, such as seen in 
# https://github.com/nathanmarz/storm-contrib/blob/master/storm-kafka/src/jvm/storm/kafka/trident/OpaqueTridentKafkaSpout.java


class AtomSpout(BaseRichSpout):
    # FIXME doc the conf requirements

    __proxymaker__ = PackageProxy("otter")

    def open(self, conf, context, collector):
        if conf.get("trust_all_certificates"):
            trust_all_certificates()
        self.collector = collector
        self.last_time = time.time()

        # FIXME get last_id from ZK; also consider how to partition
        # with consistent hashing
        self.readers = [AtomReader(feed_url) for feed_url in conf["atom_feeds"]]

    def nextTuple(self):
        for reader in self.readers:
            for event in reader.read_events():
                self.collector.emit(Values([event.updated, event]))
        # FIXME update last_id per feed in ZK

        # FIXME low pri, but should make configurable
        # Also, in the event of lengthy disconnect we may also see rate limiting
        time.sleep(1.)

    def declareOutputFields(self, declarer):
        declarer.declare(Fields(["ts", "event"]))


# NOTE keep this as a distinct, testable piece of code


class SeenWindow(object):
    """Keeps track of what has been seen for `window` number of items"""

    def __init__(self, window=500):
        self.seen = set()
        self.purgeq = deque()
        self.window = window

    def add(self, item):
        self.seen.add(item)
        self.purgeq.append(item)
        if len(self.purgeq) > self.window:
            old = self.purgeq.popLeft()
            self.seen.remove(old)

    def __contains__(self, item):
        return item in self.seen


class AtomReader(object):

    def __init__(self, feed_url, read_back_pages=200, last_id=None):
        self.feed_url = feed_url
        self.read_back_pages = read_back_pages
        self.last_id = last_id
        self.parser = Abdera().getParser()
        self.seen = SeenWindow()
        self.last_updated = None
        self.total_events = 0
        self.log = LoggerFactory.getLogger(AtomReader)

    def _read_pages(self):
        """Reads pages from the feed, from newest to oldest (reverse chronology)"""
        page_count = 0
        url = self.feed_url
        while page_count < self.read_back_pages:  # FIXME what if we exhaust our pages?
            with closing(URL(url).openStream()) as f:
                self.log.debug("Reading feed: {}", url)
                doc = self.parser.parse(f)
                feed = doc.getRoot()
                url = str(feed.getLinks("next")[0].href)
                for entry in feed.entries:
                    if self.last_id == entry.id:
                        return  # done given read back to last_id
                    yield entry
                    page_count += 1

    def read_events(self):
        """Deliver events from oldest to newest, with some minimal sanity checking"""
        count = 0
        for event in reversed(list(self._read_pages())):
            if self.last_updated and event.updated < self.last_updated:
                self.log.warn("Ignoring out of order event in feed {}: {} ({}) is older than previous event {}",
                         self.feed_url, event.id, event.updated, self.last_updated)
                continue
            if event.id in self.seen:
                self.log.warn("Ignoring duplicated event in feed {}: {}", self.feed_url, event.id)
                continue
            self.last_updated = event.updated
            self.seen.add(event.id)
            self.last_id = event.id
            count += 1
            yield event

        # FIXME storm supports metrics. use that functionality.
        self.total_events += count
        self.log.debug("Read {} of {} events from feed {}", count, self.total_events, self.feed_url)
{'fraction_non_alphanumeric': 0.09177718832891246, 'fraction_numerical': 0.006631299734748011, 'mean_word_length': 2.3605169340463457, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6025811', 'n_tokens_mistral': 2233, 'n_tokens_neox': 2125, 'n_words': 637}