<commit_before># -*- coding: utf-8 -
#
# This file is part of couchdbkit released under the MIT license. 
# See the NOTICE for more information.

import anyjson
import asyncore
import asynchat
import socket
import sys

class Consumer(object):
    """ Database change consumer
    
    Example Usage:
    
        >>> from couchdbkit import Server, Consumer
        >>> s = Server()
        >>> db = s['testdb']
        >>> c = Consumer(db)
        >>> def print_line(line):
        ...     print "got %s" % line
        ... 
        >>> c.register_callback(print_line)
        >>> c.wait() # Go into receive loop
         
    """
    
    def __init__(self, db):
        self.db = db
        self.callbacks = []
        self._resp = None

    def register_callback(self, callback):
        """ Register a callback on which changes will
        be sent with `wait_once` and `wait`methods. 
        """
        
        if not callable(callback):
            raise TypeError("callback isn't a callable")
        self.callbacks.append(callback)

    def _make_params(self, feed, since, heartbeat, timeout, filter_name):
        params = {"feed": feed}
        if since: params["since"] = since
        if heartbeat:
            params["heartbeat"] = heartbeat
        elif timeout:
            params["timeout"] = timeout
        if filter_name: params["filter"] = filter_name
        return params
        
    def fetch(self, since=None, filter_name=None):
        """ Fetch all changes and return. If since is specified, fetch all changes
        since this doc sequence
        
        Args:
        @param since: str or int, sequence from which you want to get changes
        @param filter_name: filter_name to use
        
        @return: dict, change result
        
        """
        params = {}
        if since: params['since'] = since
        if filter_name: params["filter"] = filter_name
        resp = self.db.res.get("_changes", **params)
        return resp.json_body
        
    def wait_once(self, since=None, heartbeat=False, timeout=60000, filter_name=None):
        """Wait for one change and return (longpoll feed) 
        
        Args:
        @param since: str or int, sequence from which you want to get changes
        @param heartbeat: boolean, try to maintain connection by sending '\n'
        @param timeout: int, timeout in ms
        @param filter_name: filter_name to use
        
        @return: dict, change result
        """
        params = self._make_params("longpoll", since, heartbeat, 
                            timeout, filter_name)
        resp = self.db.res.get("_changes", **params)
        buf = ""
        while True:
            data = resp.body_file.read()
            if not data: break
            buf += data
            
        ret = anyjson.deserialize(buf)
        for callback in self.callbacks:
            callback(ret)
        return ret
        
    def wait(self, since=None, heartbeat=True, timeout=None, filter_name=None):
        """ Wait for changes until the connection close (continuous feed)
        
        Args:
        @param since: str or int, sequence from which you want to get changes
        @param heartbeat: boolean, try to maintain connection by sending '\n'
        @param timeout: int, timeout in ms
        @param filter_name: str, filter_name to use
        
        @return: dict, line of change
        """
        params = self._make_params("continuous", since, heartbeat, 
                            timeout, filter_name)
        self.resp = resp = self.db.res.get("_changes", **params)
        <commit_msg>allows variable arguments in consumer. It break compatibility with olde version , yiou now have to use filter parameter instead of filter_name.<commit_after># -*- coding: utf-8 -
#
# This file is part of couchdbkit released under the MIT license. 
# See the NOTICE for more information.

import anyjson
import asyncore
import asynchat
import socket
import sys

class Consumer(object):
    """ Database change consumer
    
    Example Usage:
    
        >>> from couchdbkit import Server, Consumer
        >>> s = Server()
        >>> db = s['testdb']
        >>> c = Consumer(db)
        >>> def print_line(line):
        ...     print "got %s" % line
        ... 
        >>> c.register_callback(print_line)
        >>> c.wait() # Go into receive loop
         
    """
    
    def __init__(self, db):
        self.db = db
        self.callbacks = []
        self._resp = None

    def register_callback(self, callback):
        """ Register a callback on which changes will
        be sent with `wait_once` and `wait`methods. 
        """
        
        if not callable(callback):
            raise TypeError("callback isn't a callable")
        self.callbacks.append(callback)

    def fetch(self, **params):
        """ Fetch all changes and return. If since is specified, fetch all changes
        since this doc sequence
        
        Args:
        @param params: kwargs
        See Changes API (http://wiki.apache.org/couchdb/HTTP_database_API#Changes)
        
        @return: dict, change result
        
        """
        resp = self.db.res.get("_changes", **params)
        return resp.json_body
        
    def wait_once(self, **params):
        """Wait for one change and return (longpoll feed) 
        
        Args:
        @param params: kwargs
        See Changes API (http://wiki.apache.org/couchdb/HTTP_database_API#Changes)
        
        @return: dict, change result
        """
        params.update({"feed": "longpoll"})
        resp = self.db.res.get("_changes", **params)
        buf = ""
        while True:
            data = resp.body_file.read()
            if not data: break
            buf += data
            
        ret = anyjson.deserialize(buf)
        for callback in self.callbacks:
            callback(ret)
        return ret
        
    def wait(self, **params):
        """ Wait for changes until the connection close (continuous feed)
        
        Args:
        @param params: kwargs
        See Changes API (http://wiki.apache.org/couchdb/HTTP_database_API#Changes)
        
        @return: dict, line of change
        """
        params.update({"feed": "continuous"})
        self.resp = resp = self.db.res.get("_changes", **params)
        
{'fraction_non_alphanumeric': 0.09990500316656112, 'fraction_numerical': 0.0011082963901203294, 'mean_word_length': 1.9003673094582185, 'pattern_counts': {'":': 3, '<': 3, '<?xml version=': 0, '>': 45, 'https://': 0, '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': '3476759', 'n_tokens_mistral': 1700, 'n_tokens_neox': 1665, 'n_words': 554}