<commit_before># -*- coding: utf-8 -*-

# vim: tabstop=4 shiftwidth=4 softtabstop=4

#    Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import logging

from taskflow import exceptions as exc
from taskflow import job
from taskflow import jobboard
from taskflow import logbook

LOG = logging.getLogger(__name__)


class MemoryJob(job.Job):
    pass


class MemoryLogBook(logbook.LogBook):
    def __init__(self, resource_uri):
        super(MemoryLogBook, self).__init__(resource_uri)
        self._entries = {}

    def add_record(self, name, metadata):
        if name in self._entries:
            raise exc.RecordAlreadyExists()
        self._entries[name] = metadata

    def fetch_record(self, name):
        if name not in self._entries:
            raise exc.RecordNotFound()
        return self._entries[name]


class MemoryJobBoard(jobboard.JobBoard):
    pass<commit_msg>Fill in the majority of the memory job.<commit_after># -*- coding: utf-8 -*-

# vim: tabstop=4 shiftwidth=4 softtabstop=4

#    Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from datetime import datetime

import functools
import logging
import threading

from taskflow import catalog
from taskflow import exceptions as exc
from taskflow import jobboard
from taskflow import logbook
from taskflow import utils

LOG = logging.getLogger(__name__)


def check_not_closed(meth):

    @functools.wraps(meth)
    def check(self, *args, **kwargs):
        if self._closed:
            raise exc.ClosedException("Unable to call %s on closed object" %
                                      (meth.__name__))
        return meth(self, *args, **kwargs)

    return check


class MemoryCatalog(catalog.Catalog):
    def __init__(self):
        super(MemoryCatalog, self).__init__()
        self._catalogs = []
        self._closed = False
        self._lock = threading.RLock()

    def close(self):
        self._closed = True

    @check_not_closed
    def create_or_fetch(self, job):
        with self._lock:
            for (j, b) in self._catalogs:
                if j == job:
                    return b
            b = MemoryLogBook()
            self._catalogs.append((j, b))
            return b

    @check_not_closed
    def erase(self, job):
        with self._lock:
            self._catalogs = [(j, b) for (j, b) in self._catalogs if j != job]


class MemoryLogBook(logbook.LogBook):
    def __init__(self):
        super(MemoryLogBook, self).__init__()
        self._entries = []
        self._closed = False

    @check_not_closed
    def add(self, entry):
        self._entries.append(entry)

    @check_not_closed
    def __iter__(self):
        for e in self._entries:
            yield e

    def close(self):
        self._closed = True

    @check_not_closed
    def __contains__(self, name):
        for e in self:
            if e.name == name:
                return True
        return False

    @check_not_closed
    def erase(self, name):
        self._entries = [e for e in self._entries if e.name == name]


class MemoryJobBoard(jobboard.JobBoard):
    def __init__(self):
        super(MemoryJobBoard, self).__init__()
        self._event = threading.Event()
        # Locking to ensure that if there are multiple
        # users posting to the backing board that we only
        # have 1 writer modifying it at a time, but we can
        # have X readers.
        self._lock = utils.ReaderWriterLock()
        self._board = []
        self._closed = False

    def close(self):
        self._closed = True

    def _select_posts(self, date_functor):
        for (d, j) in self._board:
            if date_functor(d):
                yield j

    @check_not_closed
    def post(self, job):
        with self._lock.acquire(read=False):
            self._active_jobs.append((datetime.utcnow(), job))
        # Let people know a job is here
        self._notify_posted(job)
        self._event.set()
        # And now that they are notified, wait for
        # another posting.
        self._event.clear()

    @check_not_closed
    def posted_before(self, date_posted=None):
        date_functor = lambda d: True
        if date_posted is not None:
            date_functor = lambda d: d < date_posted

        with self._lock.acquire(read=True):
            return [j for j in self._select_posts(date_functor)]

    @check_not_closed
    def erase(self, job):
        with self._lock.acquire(read=False):
            # Ensure that we even have said job in the first place.
            exists = False
            for (d, j) in self._board:
                if j == job:
                    exists = True
                    break
           if not exists:
               raise exc.JobNotFound()
           self._board = [(d, j) for (d, j) in self._board if j != job]

    @check_not_closed
    def posted_after(self, date_posted=None):
        date_functor = lambda d: True
        if date_posted is not None:
            date_functor = lambda d: d >= date_posted

        with self._lock.acquire(read=True):
            return [j for j in self._select_posts(date_functor)]

    @check_not_closed
    def await(self, timeout=None):
        with not self._event.is_set():
            self._event.wait(timeout)
{'fraction_non_alphanumeric': 0.08481231349144024, 'fraction_numerical': 0.003926495994974085, 'mean_word_length': 2.3235908141962422, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2945969', 'n_tokens_mistral': 1834, 'n_tokens_neox': 1777, 'n_words': 613}