<commit_before>//
//  Switch.swift
//  Rx
//
//  Created by <NAME> on 3/12/15.
//  Copyright (c) 2015 <NAME>. All rights reserved.
//

import Foundation

class SwitchSink<S: ObservableType, O: ObserverType where S.E == O.Element> : Sink<O>, ObserverType {
    typealias Element = S
    typealias Parent = Switch<S>
    
    typealias SwitchState = (
        subscription: SingleAssignmentDisposable,
        innerSubscription: SerialDisposable,
        stopped: Bool,
        latest: Int,
        hasLatest: Bool
    )
    
    let parent: Parent
    
    var lock = NSRecursiveLock()
    var switchState: SwitchState
    
    init(parent: Parent, observer: O, cancel: Disposable) {
        self.parent = parent
        self.switchState = (
                subscription: SingleAssignmentDisposable(),
                innerSubscription: SerialDisposable(),
                stopped: false,
                latest: 0,
                hasLatest: false
        )
        super.init(observer: observer, cancel: cancel)
    }
    
    func run() -> Disposable {
        let subscription = self.parent.sources.subscribeSafe(self)
        let switchState = self.switchState
        switchState.subscription.disposable = subscription
        return CompositeDisposable(switchState.subscription, switchState.innerSubscription)
    }
    
    func on(event: Event<Element>) {
        switch event {
        case .Next(let observable):
            let latest: Int = self.lock.calculateLocked {
                self.switchState.hasLatest = true
                self.switchState.latest = self.switchState.latest &+ 1
                return self.switchState.latest
            }
            
            let d = SingleAssignmentDisposable()
            self.switchState.innerSubscription.disposable = d
               
            let observer = SwitchSinkIter(parent: self, id: latest, _self: d)
            let disposable = observable.subscribeSafe(observer)
            d.disposable = disposable
        case .Error(let error):
            self.lock.performLocked {
                trySendError(observer, error)
                self.dispose()
            }
        case .Completed:
            self.lock.performLocked {
                self.switchState.stopped = true
                
                self.switchState.subscription.dispose()
                
                if !self.switchState.hasLatest {
                    trySendCompleted(observer)
                    self.dispose()
                }
            }
        }
    }
}

class SwitchSinkIter<S: ObservableType, O: ObserverType where S.E == O.Element> : ObserverType {
    typealias Element = O.Element
    typealias Parent = SwitchSink<S, O>
    
    let parent: Parent
    let id: Int
    let _self: Disposable
    
    init(parent: Parent, id: Int, _self: Disposable) {
        self.parent = parent
        self.id = id
        self._self = _self
    }
    
    func on(event: Event<Element>) {
        return parent.lock.calculateLocked {
            let switchState = self.parent.switchState
            
            switch event {
            case .Next: break
            case .Error: fallthrough
            case .Completed: self._self.dispose()
            }
            
            if switchState.latest != self.id {
                return
            }
           
            let observer = self.parent.observer
            
            switch event {
            case .Next:
                trySend(observer, event)
            case .Error:
                trySend(observer, event)
                self.parent.dispose()
            case .Completed:
                parent.switchState.hasLatest = false
                if switchState.stopped {
                    trySend(observer, event)
                    self.parent.dispose()
                }
            }
        }
    }
}

class Switch<S: ObservableType> : Producer<S.E> {<commit_msg>`trySend` to `?.on` and cleanup for `switchLatest`<commit_after>//
//  Switch.swift
//  Rx
//
//  Created by <NAME> on 3/12/15.
//  Copyright (c) 2015 <NAME>. All rights reserved.
//

import Foundation

class SwitchSink<S: ObservableType, O: ObserverType where S.E == O.Element> : Sink<O>, ObserverType {
    typealias Element = S
    typealias Parent = Switch<S>

    let subscriptions: SingleAssignmentDisposable = SingleAssignmentDisposable()
    let innerSubscription: SerialDisposable = SerialDisposable()
    let parent: Parent
    
    var lock = NSRecursiveLock()
    
    // state
    var stopped = false
    var latest = 0
    var hasLatest = false
    
    init(parent: Parent, observer: O, cancel: Disposable) {
        self.parent = parent
        
        super.init(observer: observer, cancel: cancel)
    }
    
    func run() -> Disposable {
        let subscription = self.parent.sources.subscribeSafe(self)
        subscriptions.disposable = subscription
        return CompositeDisposable(subscriptions, innerSubscription)
    }
    
    func on(event: Event<Element>) {
        switch event {
        case .Next(let observable):
            let latest: Int = self.lock.calculateLocked {
                hasLatest = true
                self.latest = self.latest &+ 1
                return self.latest
            }
            
            let d = SingleAssignmentDisposable()
            innerSubscription.disposable = d
               
            let observer = SwitchSinkIter(parent: self, id: latest, _self: d)
            let disposable = observable.subscribeSafe(observer)
            d.disposable = disposable
        case .Error(let error):
            self.lock.performLocked {
                observer?.on(.Error(error))
                self.dispose()
            }
        case .Completed:
            self.lock.performLocked {
                self.stopped = true
                
                self.subscriptions.dispose()
                
                if !self.hasLatest {
                    observer?.on(.Completed)
                    self.dispose()
                }
            }
        }
    }
}

class SwitchSinkIter<S: ObservableType, O: ObserverType where S.E == O.Element> : ObserverType {
    typealias Element = O.Element
    typealias Parent = SwitchSink<S, O>
    
    let parent: Parent
    let id: Int
    let _self: Disposable
    
    init(parent: Parent, id: Int, _self: Disposable) {
        self.parent = parent
        self.id = id
        self._self = _self
    }
    
    func on(event: Event<Element>) {
        return parent.lock.calculateLocked {
            
            switch event {
            case .Next: break
            case .Error: fallthrough
            case .Completed: self._self.dispose()
            }
            
            if parent.latest != self.id {
                return
            }
           
            let observer = self.parent.observer
            
            switch event {
            case .Next:
                observer?.on(event)
            case .Error:
                observer?.on(event)
                self.parent.dispose()
            case .Completed:
                parent.hasLatest = false
                if parent.stopped {
                    observer?.on(event)
                    self.parent.dispose()
                }
            }
        }
    }
}

class Switch<S: ObservableType> : Producer<S.E> {
{'fraction_non_alphanumeric': 0.08504838489845987, 'fraction_numerical': 0.0029985007496251873, 'mean_word_length': 1.479054054054054, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 27, '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': '4889122', 'n_tokens_mistral': 1980, 'n_tokens_neox': 1887, 'n_words': 452}