<commit_before>from observable import Producer
from .sink import Sink


class DistinctUntilChanged(Producer):
  def __init__(self, source, keySelector=id):
    self.source = source
    self.keySelector = keySelector

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(Sink):
    def __init__(self, parent, observer, cancel):
      super(DistinctUntilChanged.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.currentKey = None
      self.hasCurrentKey = False

    def onNext(self, value):
      try:
        key = self.parent.keySelector(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        equal = False

        if self.hasCurrentKey:
          try:
            equal = self.currentKey == key
          except Exception as e:
            self.observer.onError(e)
            self.dispose()

        if not self.hasCurrentKey or not equal:
          self.hasCurrentKey = True
          self.currentKey = key
          self.observer.onNext(value)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose()<commit_msg>Removed default parameter from DistinctUntilChanged<commit_after>from observable import Producer
from .sink import Sink


class DistinctUntilChanged(Producer):
  def __init__(self, source, keySelector):
    self.source = source
    self.keySelector = keySelector

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(Sink):
    def __init__(self, parent, observer, cancel):
      super(DistinctUntilChanged.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.currentKey = None
      self.hasCurrentKey = False

    def onNext(self, value):
      try:
        key = self.parent.keySelector(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        equal = False

        if self.hasCurrentKey:
          try:
            equal = self.currentKey == key
          except Exception as e:
            self.observer.onError(e)
            self.dispose()

        if not self.hasCurrentKey or not equal:
          self.hasCurrentKey = True
          self.currentKey = key
          self.observer.onNext(value)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose()
{'fraction_non_alphanumeric': 0.09131238447319778, 'fraction_numerical': 0.0, 'mean_word_length': 2.473684210526316, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '215393', 'n_tokens_mistral': 797, 'n_tokens_neox': 778, 'n_words': 189}