<commit_before>import datetime
import redis
import time
import urlparse

import beaver.transport


class RedisTransport(beaver.transport.Transport):

    def __init__(self, beaver_config, file_config, logger=None):
        super(RedisTransport, self).__init__(beaver_config, file_config, logger=logger)

        redis_url = beaver_config.get('redis_url')
        redis_password = beaver_config.get('redis_password')
        _url = urlparse.urlparse(redis_url, scheme="redis")
        _, _, _db = _url.path.rpartition("/")

        self._redis = redis.StrictRedis(host=_url.hostname, port=_url.port, password=<PASSWORD>, db=int(_db), socket_timeout=10)
        self._redis_namespace = beaver_config.get('redis_namespace')

        wait = 0
        while 1:
            if wait == 20:
                break

            time.sleep(0.1)
            wait += 1
            try:
                self._redis.ping()
                break
            except redis.exceptions.ConnectionError:
                pass

    def callback(self, filename, lines):
        timestamp = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")

        for line in lines:
            self._redis.rpush(
                self._redis_namespace,
                self.format(filename, timestamp, line)
            )<commit_msg>Use redis pipelining when sending events<commit_after>import datetime
import redis
import time
import urlparse

import beaver.transport


class RedisTransport(beaver.transport.Transport):

    def __init__(self, beaver_config, file_config, logger=None):
        super(RedisTransport, self).__init__(beaver_config, file_config, logger=logger)

        redis_url = beaver_config.get('redis_url')
        redis_password = beaver_config.get('redis_password')
        _url = urlparse.urlparse(redis_url, scheme="redis")
        _, _, _db = _url.path.rpartition("/")

        self._redis = redis.StrictRedis(host=_url.hostname, port=_url.port, password=<PASSWORD>, db=int(_db), socket_timeout=10)
        self._redis_namespace = beaver_config.get('redis_namespace')

        wait = 0
        while 1:
            if wait == 20:
                break

            time.sleep(0.1)
            wait += 1
            try:
                self._redis.ping()
                break
            except redis.exceptions.ConnectionError:
                pass

        self._pipeline = self._redis.pipeline(transaction=False)

    def callback(self, filename, lines):
        timestamp = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")

        for line in lines:
            self._pipeline.rpush(
                self._redis_namespace,
                self.format(filename, timestamp, line)
            )

        self._pipeline.execute()
{'fraction_non_alphanumeric': 0.1062043795620438, 'fraction_numerical': 0.006569343065693431, 'mean_word_length': 2.290516206482593, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '6057068', 'n_tokens_mistral': 837, 'n_tokens_neox': 799, 'n_words': 148}