<commit_before>//
//  The MIT License (MIT)
//
//  Copyright (c) 2016 <NAME> (@srdanrasic)
//
//  Permission is hereby granted, free of charge, to any person obtaining a copy
//  of this software and associated documentation files (the "Software"), to deal
//  in the Software without restriction, including without limitation the rights
//  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//  copies of the Software, and to permit persons to whom the Software is
//  furnished to do so, subject to the following conditions:
//
//  The above copyright notice and this permission notice shall be included in
//  all copies or substantial portions of the Software.
//
//  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//  THE SOFTWARE.
//

// MARK: - _StreamType

/// Represents a stream over generic EventType.
public protocol _StreamType {

  /// The type of events generated by the stream.
  associatedtype Event: EventType

  /// Register an observer that will receive events from a stream.
  ///
  /// In case of pull-driven streams, e.g. `Stream<T>` or `Operation<T, E>`,
  /// this actually triggers event generation.
  @warn_unused_result
  func observe(observer: Event -> Void) -> Disposable
}

extension _StreamType {

  /// Register an observer that will receive elements from `.Next` events of the stream.
  @warn_unused_result
  public func observeNext(observer: Event.Element -> Void) -> Disposable {
    return observe { event in
      if let element = event.element {
        observer(element)
      }
    }
  }
}

// MARK: - RawStreamType

/// Represents an underlying stream generalized over EventType and used by
/// higher-level implementations like `Stream<T>` or `Operation<T, E>`.
public protocol RawStreamType: _StreamType {
}

// MARK: - RawStream

/// An underlying stream generalized over EventType and used by
/// higher-level implementations like `Stream<T>` or `Operation<T, E>`.
public struct RawStream<Event: EventType>: RawStreamType {

  private let producer: Observer<Event> -> Disposable

  /// Create new `RawStream` given a producer closure.
  public init(producer: Observer<Event> -> Disposable) {
    self.producer = producer
  }

  /// Register an observer that will receive events from a stream.
  @warn_unused_result
  public func observe(observer: Event -> Void) -> Disposable {
    var terminated = false
    let observer = Observer<Event> { event in
      guard !terminated else { return }
      terminated = event.isTermination
      observer(event)
    }
    return producer(observer)
  }
}

// MARK: - Extensions

public extension _StreamType {

  /// Transform a stream type into concrete `RawStream`.
  @warn_unused_result
  public func toRawStream() -> RawStream<Event> {
    return RawStream { observer in
      return self.observe(observer.observer)
    }
  }
}

// MARK: Creation

public extension RawStream where Event.Element == Int {

  /// Create a stream that emits an integer every `interval` time on a given queue.
  @warn_unused_result
  public static func interval(interval: TimeValue, queue: Queue) -> RawStream<Event> {
    return RawStream { observer in
      var number = 0
      var dispatch: (() -> Void)!
      let disposable = SimpleDisposable()
      dispatch = {
        queue.after(interval) {
          if !disposable.isDisposed {
            observer.next(number)
            number += 1
            dispatch()
          } else {
            dispatch = nil
          }
        }
      }
      dispatch()
      return disposable
    }
  }
}

public extension RawStream {

  /// Create a stream that emits given element after `time` time on a given queue.
  @warn_unused_result
  public static func timer(element: Event.Element, time: TimeValue, queue: Queue) -> RawStream<Event> {
    return RawStream { observer in
      let disposable = SimpleDisposable()
      queue.after(time) {<commit_msg>Make RawStream.interval to work with all IntegerType, not just Int<commit_after>//
//  The MIT License (MIT)
//
//  Copyright (c) 2016 <NAME> (@srdanrasic)
//
//  Permission is hereby granted, free of charge, to any person obtaining a copy
//  of this software and associated documentation files (the "Software"), to deal
//  in the Software without restriction, including without limitation the rights
//  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//  copies of the Software, and to permit persons to whom the Software is
//  furnished to do so, subject to the following conditions:
//
//  The above copyright notice and this permission notice shall be included in
//  all copies or substantial portions of the Software.
//
//  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//  THE SOFTWARE.
//

// MARK: - _StreamType

/// Represents a stream over generic EventType.
public protocol _StreamType {

  /// The type of events generated by the stream.
  associatedtype Event: EventType

  /// Register an observer that will receive events from a stream.
  ///
  /// In case of pull-driven streams, e.g. `Stream<T>` or `Operation<T, E>`,
  /// this actually triggers event generation.
  @warn_unused_result
  func observe(observer: Event -> Void) -> Disposable
}

extension _StreamType {

  /// Register an observer that will receive elements from `.Next` events of the stream.
  @warn_unused_result
  public func observeNext(observer: Event.Element -> Void) -> Disposable {
    return observe { event in
      if let element = event.element {
        observer(element)
      }
    }
  }
}

// MARK: - RawStreamType

/// Represents an underlying stream generalized over EventType and used by
/// higher-level implementations like `Stream<T>` or `Operation<T, E>`.
public protocol RawStreamType: _StreamType {
}

// MARK: - RawStream

/// An underlying stream generalized over EventType and used by
/// higher-level implementations like `Stream<T>` or `Operation<T, E>`.
public struct RawStream<Event: EventType>: RawStreamType {

  private let producer: Observer<Event> -> Disposable

  /// Create new `RawStream` given a producer closure.
  public init(producer: Observer<Event> -> Disposable) {
    self.producer = producer
  }

  /// Register an observer that will receive events from a stream.
  @warn_unused_result
  public func observe(observer: Event -> Void) -> Disposable {
    var terminated = false
    let observer = Observer<Event> { event in
      guard !terminated else { return }
      terminated = event.isTermination
      observer(event)
    }
    return producer(observer)
  }
}

// MARK: - Extensions

public extension _StreamType {

  /// Transform a stream type into concrete `RawStream`.
  @warn_unused_result
  public func toRawStream() -> RawStream<Event> {
    return RawStream { observer in
      return self.observe(observer.observer)
    }
  }
}

// MARK: Creation

public extension RawStream where Event.Element: IntegerType {

  /// Create a stream that emits an integer every `interval` time on a given queue.
  @warn_unused_result
  public static func interval(interval: TimeValue, queue: Queue) -> RawStream<Event> {
    return RawStream { observer in
      var number: Event.Element = 0
      var dispatch: (() -> Void)!
      let disposable = SimpleDisposable()
      dispatch = {
        queue.after(interval) {
          guard !disposable.isDisposed else { dispatch = nil; return }
          observer.next(number)
          number = number + 1
          dispatch()
        }
      }
      dispatch()
      return disposable
    }
  }
}

public extension RawStream {

  /// Create a stream that emits given element after `time` time on a given queue.
  @warn_unused_result
  public static func timer(element: Event.Element, time: TimeValue, queue: Queue) -> RawStream<Event> {
    return RawStream { observer in
      let disposable = SimpleDisposable()
      queue.after(time) {
{'fraction_non_alphanumeric': 0.09052170908672118, 'fraction_numerical': 0.0013820108257514684, 'mean_word_length': 3.059841047218326, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 55, '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': '2846671', 'n_tokens_mistral': 2416, 'n_tokens_neox': 2297, 'n_words': 954}