<commit_before># Copyright 2016 Google Inc.
#
# 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 concurrent.futures import ThreadPoolExecutor
import queue
import re
import threading
import time
import traceback


class EventDispatcherError(Exception):
  pass


class IllegalStateError(EventDispatcherError):
  """Raise when user tries to put event_dispatcher into an illegal state.
  """


class DuplicateError(EventDispatcherError):
  """Raise when a duplicate is being created and it shouldn't.
  """


class EventDispatcher:
  """Class managing events for an sl4a connection.
  """

  DEFAULT_TIMEOUT = 60

  def __init__(self, sl4a):
    self._sl4a = sl4a
    self.started = False
    self.executor = None
    self.poller = None
    self.event_dict = {}
    self.handlers = {}
    self.lock = threading.RLock()

  def poll_events(self):
    """Continuously polls all types of events from sl4a.

    Events are sorted by name and store in separate queues.
    If there are registered handlers, the handlers will be called with
    corresponding event immediately upon event discovery, and the event
    won't be stored. If exceptions occur, stop the dispatcher and return
    """
    while self.started:
      event_obj = None
      event_name = None
      try:
        event_obj = self._sl4a.eventWait(50000)
      except:
        if self.started:
          print("Exception happened during polling.")
          print(traceback.format_exc())
          raise
      if not event_obj:
        continue
      elif 'name' not in event_obj:
        print("Received Malformed event {}".format(event_obj))
        continue
      else:
        event_name = event_obj['name']
      # if handler registered, process event
      if event_name in self.handlers:
        self.handle_subscribed_event(event_obj, event_name)
      if event_name == "EventDispatcherShutdown":
        self._sl4a.closeSl4aSession()
        break
      else:
        self.lock.acquire()
        if event_name in self.event_dict:  # otherwise, cache event
          self.event_dict[event_name].put(event_obj)
        else:
          q = queue.Queue()
          q.put(event_obj)
          self.event_dict[event_name] = q
        self.lock.release()

  def register_handler(self, handler, event_name, args):
    """Registers an event handler.

    One type of event can only have one event handler associated with it.

    Args:
      handler: The event handler function to be registered.
      event_name: Name of the event the handler is for.
      args: User arguments to be passed to the handler when it's called.

    Raises:
      IllegalStateError: Raised if attempts to register a handler after
        the dispatcher starts running.
      DuplicateError: Raised if attempts to register more than one
        handler for one type of event.
    """
    if self.started:
      raise IllegalStateError(("Can't register service after polling is"
                               " started"))
    self.lock.acquire()
    try:
      if event_name in self.handlers:
        raise DuplicateError(
            'A handler for {} already exists'.format(event_name))
      self.handlers[event_name] = (handler, args)
    finally:
      self.lock.release()

  def start(self):
    """Starts the event dispatcher.

    Initiates executor and start polling events.

    Raises:
      IllegalStateError: Can't start a dispatcher again when it's already
        running.
    """
    if not self.started:
      self.started = True
      self.executor = ThreadPoolExecutor(max_workers=32)
      self.poller = self.executor.submit(self.poll_events)
    else:
      raise IllegalStateError("Dispatcher is already started.")

  def clean_up(self):
    """Clean up and release resources after the event dispatcher polling
    loop has been broken.

    The following things happen:
    1. Clear all events and flags.
    2. Close the sl4a client the event_dispatcher object holds.
    3. Shut down executor without waiting.
    """
    if not self.started:<commit_msg>Remove importing class from modules. (#721)<commit_after># Copyright 2016 Google Inc.
#
# 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 concurrent import futures
import queue
import re
import threading
import time
import traceback


class EventDispatcherError(Exception):
  pass


class IllegalStateError(EventDispatcherError):
  """Raise when user tries to put event_dispatcher into an illegal state.
  """


class DuplicateError(EventDispatcherError):
  """Raise when a duplicate is being created and it shouldn't.
  """


class EventDispatcher:
  """Class managing events for an sl4a connection.
  """

  DEFAULT_TIMEOUT = 60

  def __init__(self, sl4a):
    self._sl4a = sl4a
    self.started = False
    self.executor = None
    self.poller = None
    self.event_dict = {}
    self.handlers = {}
    self.lock = threading.RLock()

  def poll_events(self):
    """Continuously polls all types of events from sl4a.

    Events are sorted by name and store in separate queues.
    If there are registered handlers, the handlers will be called with
    corresponding event immediately upon event discovery, and the event
    won't be stored. If exceptions occur, stop the dispatcher and return
    """
    while self.started:
      event_obj = None
      event_name = None
      try:
        event_obj = self._sl4a.eventWait(50000)
      except:
        if self.started:
          print("Exception happened during polling.")
          print(traceback.format_exc())
          raise
      if not event_obj:
        continue
      elif 'name' not in event_obj:
        print("Received Malformed event {}".format(event_obj))
        continue
      else:
        event_name = event_obj['name']
      # if handler registered, process event
      if event_name in self.handlers:
        self.handle_subscribed_event(event_obj, event_name)
      if event_name == "EventDispatcherShutdown":
        self._sl4a.closeSl4aSession()
        break
      else:
        self.lock.acquire()
        if event_name in self.event_dict:  # otherwise, cache event
          self.event_dict[event_name].put(event_obj)
        else:
          q = queue.Queue()
          q.put(event_obj)
          self.event_dict[event_name] = q
        self.lock.release()

  def register_handler(self, handler, event_name, args):
    """Registers an event handler.

    One type of event can only have one event handler associated with it.

    Args:
      handler: The event handler function to be registered.
      event_name: Name of the event the handler is for.
      args: User arguments to be passed to the handler when it's called.

    Raises:
      IllegalStateError: Raised if attempts to register a handler after
        the dispatcher starts running.
      DuplicateError: Raised if attempts to register more than one
        handler for one type of event.
    """
    if self.started:
      raise IllegalStateError(("Can't register service after polling is"
                               " started"))
    self.lock.acquire()
    try:
      if event_name in self.handlers:
        raise DuplicateError(
            'A handler for {} already exists'.format(event_name))
      self.handlers[event_name] = (handler, args)
    finally:
      self.lock.release()

  def start(self):
    """Starts the event dispatcher.

    Initiates executor and start polling events.

    Raises:
      IllegalStateError: Can't start a dispatcher again when it's already
        running.
    """
    if not self.started:
      self.started = True
      self.executor = futures.ThreadPoolExecutor(max_workers=32)
      self.poller = self.executor.submit(self.poll_events)
    else:
      raise IllegalStateError("Dispatcher is already started.")

  def clean_up(self):
    """Clean up and release resources after the event dispatcher polling
    loop has been broken.

    The following things happen:
    1. Clear all events and flags.
    2. Close the sl4a client the event_dispatcher object holds.
    3. Shut down executor without waiting.
    """
    if not self.started:
{'fraction_non_alphanumeric': 0.07315725494529782, 'fraction_numerical': 0.006741076362028953, 'mean_word_length': 2.8009239815203695, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4150334', 'n_tokens_mistral': 2529, 'n_tokens_neox': 2410, 'n_words': 926}