<commit_before>//
// Created by <NAME> on 15/09/15.
// Copyright (c) 2015 <NAME>. All rights reserved.
//

let noValueError = NSError(domain: "no value", code: -1, userInfo: nil)

let q = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_BACKGROUND, 0)

public class Source<T>: Hub<T> {
    
    let state: AtomicReference<Try<T>>

    public init(initValue: T) {
        state = AtomicReference(Try(initValue))
        super.init()
    }
    
    public override init() {
        state = AtomicReference(Try(noValueError))
        super.init()
    }
    
    public func update(newValue: T) {
        self.updateSilent(newValue)
        dispatch_async(q) {
            self.propagate()
        }
    }
    
    public func error(error: NSError) {
        self.state.value = Try(error)
        propagate()
    }
    
    public func updateSilent(newValue:T) {
        self.state.value = Try(newValue)
    }
    
    override func isSuccess() -> Bool {
        return toTry().isSuccess()
    }
    
    override public func toTry() -> Try<T> {
        return state.value
    }
    
    override func ping(incoming: Set<Node>) -> Set<Node> {<commit_msg>Propagate errors in async style<commit_after>//
// Created by <NAME> on 15/09/15.
// Copyright (c) 2015 <NAME>. All rights reserved.
//

let noValueError = NSError(domain: "no value", code: -1, userInfo: nil)

let q = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_BACKGROUND, 0)

public class Source<T>: Hub<T> {
    
    let state: AtomicReference<Try<T>>

    public init(initValue: T) {
        state = AtomicReference(Try(initValue))
        super.init()
    }
    
    public override init() {
        state = AtomicReference(Try(noValueError))
        super.init()
    }
    
    public func update(newValue: T) {
        self.updateSilent(newValue)
        dispatch_async(q) {
            self.propagate()
        }
    }
    
    public func error(error: NSError) {
        self.state.value = Try(error)
        dispatch_async(q) {
            self.propagate()
        }
    }
    
    public func updateSilent(newValue:T) {
        self.state.value = Try(newValue)
    }
    
    override func isSuccess() -> Bool {
        return toTry().isSuccess()
    }
    
    override public func toTry() -> Try<T> {
        return state.value
    }
    
    override func ping(incoming: Set<Node>) -> Set<Node> {
{'fraction_non_alphanumeric': 0.12040557667934093, 'fraction_numerical': 0.010139416983523447, 'mean_word_length': 2.2130257801899593, 'pattern_counts': {'":': 0, '<': 21, '<?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': '3375178', 'n_tokens_mistral': 798, 'n_tokens_neox': 758, 'n_words': 169}