<commit_before>"""
RxPY example running a Tornado server doing search queries against Wikipedia to
populate the autocomplete dropdown in the web UI. Start using 
`python autocomplete.py` and navigate your web browser to http://localhost:8080

Uses the RxPY IOLoopScheduler (works on both Python 2.7 and 3.4)
"""

import os

from tornado.websocket import WebSocketHandler
from tornado.web import RequestHandler, StaticFileHandler, Application, url
from tornado.httpclient import AsyncHTTPClient
from tornado.httputil import url_concat
from tornado.escape import json_encode, json_decode
from tornado import ioloop

from rx.subjects import Subject
from rx.concurrency import IOLoopScheduler

scheduler = IOLoopScheduler()

def search_wikipedia(term):
    """Search Wikipedia for a given term"""
    url = 'http://en.wikipedia.org/w/api.php'
    
    params = {
        "action": 'opensearch',
        "search": term,
        "format": 'json'
    }
    # Must set a user agent for non-browser requests to Wikipedia
    user_agent = "RxPY/1.0 (https://github.com/dbrattli/RxPY; <EMAIL>) Tornado/4.0.1"
    
    url = url_concat(url, params)
    
    http_client = AsyncHTTPClient()
    return http_client.fetch(url, method='GET', user_agent=user_agent)

class WSHandler(WebSocketHandler):
    def open(self):
        print("WebSocket opened")

        # A Subject is both an observable and observer, so we can both subscribe
        # to it and also feed (on_next) it with new values
        self.subject = Subject()

        # Get all distinct key up events from the input and only fire if long enough and distinct
        query = self.subject.map(
            lambda x: x["term"]
        ).filter(
            lambda text: len(text) > 2 # Only if the text is longer than 2 characters
        ).debounce(
            0.750, # Pause for 750ms
            scheduler=scheduler
        ).distinct_until_changed() # Only if the value has changed

        searcher = query.flat_map_latest(search_wikipedia)

        def send_response(x):
            self.write_message(x.body)

        def on_error(ex):
            print(ex)

        searcher.subscribe(send_response, on_error)

    def on_message(self, message):
        obj = json_decode(message)
        self.subject.on_next(obj)

    def on_close(self):
        print("WebSocket closed")

class MainHandler(RequestHandler):
    def get(self):
        self.render("index.html")

def main():
    port = os.environ.get("PORT", 8080)
    app = Application([
        url(r"/", MainHandler),
        (r'/ws', WSHandler),
        (r'/static/(.*)', StaticFileHandler, {'path': "."})
    ])
    print("Starting server at port: %s" % port)
    app.listen(port)
    ioloop.IOLoop.current().start()

if __name__ == '__main__':
    main()<commit_msg>Call instances of a subject for a stream<commit_after>"""
RxPY example running a Tornado server doing search queries against Wikipedia to
populate the autocomplete dropdown in the web UI. Start using 
`python autocomplete.py` and navigate your web browser to http://localhost:8080

Uses the RxPY IOLoopScheduler (works on both Python 2.7 and 3.4)
"""

import os

from tornado.websocket import WebSocketHandler
from tornado.web import RequestHandler, StaticFileHandler, Application, url
from tornado.httpclient import AsyncHTTPClient
from tornado.httputil import url_concat
from tornado.escape import json_decode
from tornado import ioloop

from rx.subjects import Subject
from rx.concurrency import IOLoopScheduler

scheduler = IOLoopScheduler()


def search_wikipedia(term):
    """Search Wikipedia for a given term"""
    url = 'http://en.wikipedia.org/w/api.php'

    params = {
        "action": 'opensearch',
        "search": term,
        "format": 'json'
    }
    # Must set a user agent for non-browser requests to Wikipedia
    user_agent = "RxPY/1.0 (https://github.com/dbrattli/RxPY; <EMAIL>) Tornado/4.0.1"

    url = url_concat(url, params)

    http_client = AsyncHTTPClient()
    return http_client.fetch(url, method='GET', user_agent=user_agent)


class WSHandler(WebSocketHandler):
    def open(self):
        print("WebSocket opened")

        # A Subject is both an observable and observer, so we can both subscribe
        # to it and also feed (on_next) it with new values
        self.stream = Subject()

        # Get all distinct key up events from the input and only fire if long enough and distinct
        query = self.stream.map(
            lambda x: x["term"]
        ).filter(
            lambda text: len(text) > 2  # Only if the text is longer than 2 characters
        ).debounce(
            0.750,  # Pause for 750ms
            scheduler=scheduler
        ).distinct_until_changed()  # Only if the value has changed

        searcher = query.flat_map_latest(search_wikipedia)

        def send_response(x):
            self.write_message(x.body)

        def on_error(ex):
            print(ex)

        searcher.subscribe(send_response, on_error)

    def on_message(self, message):
        obj = json_decode(message)
        self.stream.on_next(obj)

    def on_close(self):
        print("WebSocket closed")


class MainHandler(RequestHandler):
    def get(self):
        self.render("index.html")


def main():
    port = os.environ.get("PORT", 8080)
    app = Application([
        url(r"/", MainHandler),
        (r'/ws', WSHandler),
        (r'/static/(.*)', StaticFileHandler, {'path': "."})
    ])
    print("Starting server at port: %s" % port)
    app.listen(port)
    ioloop.IOLoop.current().start()

if __name__ == '__main__':
    main()
{'fraction_non_alphanumeric': 0.10139985642498206, 'fraction_numerical': 0.00933237616654702, 'mean_word_length': 2.919127988748242, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 2, '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': '7470774', 'n_tokens_mistral': 1716, 'n_tokens_neox': 1628, 'n_words': 515}