<commit_before>//
//  Sample.swift
//  RxSwift
//
//  Created by <NAME> on 5/1/15.
//  Copyright (c) 2015 <NAME>. All rights reserved.
//

import Foundation

class SampleImpl_<O: ObserverType, ElementType, SampleType where O.Element == ElementType> : Observer<SampleType> {
    typealias Parent = Sample_<O, SampleType>
    
    let parent: Parent
    
    init(parent: Parent) {
        self.parent = parent
    }
    
    override func on(event: Event<Element>) {
        parent.lock.performLocked {
            switch event {
            case .Next:
                if let element = parent.sampleState.element {
                    if self.parent.parent.onlyNew {
                        parent.sampleState.element = nil
                    }
                    
                    trySend(parent.observer, element)
                }

                if parent.sampleState.atEnd {
                    trySendCompleted(parent.observer)
                    parent.dispose()
                }
            case .Error(let e):
                trySendError(parent.observer, e)
                parent.dispose()
            case .Completed:
                if let element = parent.sampleState.element {
                    parent.sampleState.element = nil
                    trySend(parent.observer, element)
                }
                if parent.sampleState.atEnd {
                    trySendCompleted(parent.observer)
                    parent.dispose()
                }
            }
        }
    }
}

class Sample_<O: ObserverType, SampleType> : Sink<O>, ObserverType {
    typealias Element = O.Element
    typealias Parent = Sample<Element, SampleType>
    typealias SampleState = (
        element: Event<Element>?,
        atEnd: Bool,
        sourceSubscription: SingleAssignmentDisposable
    )
    
    let parent: Parent

    var lock = NSRecursiveLock()
    
    var sampleState: SampleState = (
        element: nil,
        atEnd: false,
        sourceSubscription: SingleAssignmentDisposable()
    )
    
    init(parent: Parent, observer: O, cancel: Disposable) {
        self.parent = parent
        super.init(observer: observer, cancel: cancel)
    }
    
    func run() -> Disposable {
        sampleState.sourceSubscription.disposable = parent.source.subscribeSafe(self)
        let samplerSubscription = parent.sampler.subscribeSafe(SampleImpl_(parent: self))
        
        return CompositeDisposable(sampleState.sourceSubscription, samplerSubscription)
    }
    
    func on(event: Event<Element>) {
        self.lock.performLocked {
            switch event {
            case .Next:
                self.sampleState.element = event
            case .Error:
                trySend(observer, event)
                self.dispose()
            case .Completed:
                self.sampleState.atEnd = true
                self.sampleState.sourceSubscription.dispose()
            }
        }
    }
    
}

class Sample<Element, SampleType> : Producer<Element> {
    let source: Observable<Element>
    let sampler: Observable<SampleType>
    let onlyNew: Bool

    init(source: Observable<Element>, sampler: Observable<SampleType>, onlyNew: Bool) {
        self.source = source
        self.sampler = sampler
        self.onlyNew = onlyNew
    }
    
    override func run<O: ObserverType where O.Element == Element>(observer: O, cancel: Disposable, setSink: (Disposable) -> Void) -> Disposable {
        let sink = Sample_(parent: self, observer: observer, cancel: cancel)
        setSink(sink)
        return sink.run()
    }
}<commit_msg>`trySend` to `?.on` and cleanup for `Sample`<commit_after>//
//  Sample.swift
//  RxSwift
//
//  Created by <NAME> on 5/1/15.
//  Copyright (c) 2015 <NAME>. All rights reserved.
//

import Foundation

class SamplerSink<O: ObserverType, ElementType, SampleType where O.Element == ElementType> : Observer<SampleType> {
    typealias Parent = SampleSequenceSink<O, SampleType>
    
    let parent: Parent
    
    init(parent: Parent) {
        self.parent = parent
    }
    
    override func on(event: Event<Element>) {
        parent.lock.performLocked {
            switch event {
            case .Next:
                if let element = parent.element {
                    if self.parent.parent.onlyNew {
                        parent.element = nil
                    }
                    
                    parent.observer?.on(.Next(element))
                }

                if parent.atEnd {
                    parent.observer?.on(.Completed)
                    parent.dispose()
                }
            case .Error(let e):
                parent.observer?.on(.Error(e))
                parent.dispose()
            case .Completed:
                if let element = parent.element {
                    parent.element = nil
                    parent.observer?.on(.Next(element))
                }
                if parent.atEnd {
                    parent.observer?.on(.Completed)
                    parent.dispose()
                }
            }
        }
    }
}

class SampleSequenceSink<O: ObserverType, SampleType> : Sink<O>, ObserverType {
    typealias Element = O.Element
    typealias Parent = Sample<Element, SampleType>
    
    let parent: Parent

    var lock = NSRecursiveLock()
    // state
    var element = nil as Element?
    var atEnd = false
    
    let sourceSubscription = SingleAssignmentDisposable()
    
    init(parent: Parent, observer: O, cancel: Disposable) {
        self.parent = parent
        super.init(observer: observer, cancel: cancel)
    }
    
    func run() -> Disposable {
        sourceSubscription.disposable = parent.source.subscribeSafe(self)
        let samplerSubscription = parent.sampler.subscribeSafe(SamplerSink(parent: self))
        
        return CompositeDisposable(sourceSubscription, samplerSubscription)
    }
    
    func on(event: Event<Element>) {
        self.lock.performLocked {
            switch event {
            case .Next(let element):
                self.element = element
            case .Error:
                observer?.on(event)
                self.dispose()
            case .Completed:
                atEnd = true
                sourceSubscription.dispose()
            }
        }
    }
    
}

class Sample<Element, SampleType> : Producer<Element> {
    let source: Observable<Element>
    let sampler: Observable<SampleType>
    let onlyNew: Bool

    init(source: Observable<Element>, sampler: Observable<SampleType>, onlyNew: Bool) {
        self.source = source
        self.sampler = sampler
        self.onlyNew = onlyNew
    }
    
    override func run<O: ObserverType where O.Element == Element>(observer: O, cancel: Disposable, setSink: (Disposable) -> Void) -> Disposable {
        let sink = SampleSequenceSink(parent: self, observer: observer, cancel: cancel)
        setSink(sink)
        return sink.run()
    }
}
{'fraction_non_alphanumeric': 0.09247218610027452, 'fraction_numerical': 0.002311804652506863, 'mean_word_length': 1.70390625, 'pattern_counts': {'":': 0, '<': 38, '<?xml version=': 0, '>': 44, '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': '6740082', 'n_tokens_mistral': 1896, 'n_tokens_neox': 1792, 'n_words': 418}