<commit_before>//
//  AsyncSequence.swift
//  RxSwift
//
//  Created by <NAME> on 2014-06-12.
//  Copyright (c) 2014 GitHub. All rights reserved.
//

import Foundation

// TODO: This lives outside of the definition below because of a crash in the
// compiler. Move it back within AsyncSequence once that's fixed.
struct _FlattenScanGenerator<S, T, U>: Generator {
	let disposable: Disposable
	let scanFunc: (S, T) -> (S?, Stream<U>)
	
	// TODO: Thread safety
	var valueGenerators: GeneratorOf<Promise<Event<U>>>[]
	var state: S
	var selfGenerator: GeneratorOf<Promise<Event<T>>>

	mutating func generateStreamFromValue(x: T) {
		let (newState, stream) = scanFunc(state, x)

		if let s = newState {
			state = s
		} else {
			disposable.dispose()
		}

		let seq = stream as AsyncSequence<U>
		valueGenerators.append(seq.generate())
	}

	mutating func next() -> Promise<Event<U>>? {
		if disposable.disposed {
			return nil
		}

		var next = valueGenerators[0].next()
		while next == nil {
			valueGenerators.removeAtIndex(0)
			if valueGenerators.isEmpty {
				if let promise = selfGenerator.next() {
					return promise.then { event in
						switch event {
						case let .Next(x):
							self.generateStreamFromValue(x)
							return Promise { .Completed }

						case let .Completed:
							return Promise { .Completed }

						case let .Error(error):
							self.disposable.dispose()
							return Promise { .Error(error) }
						}
					}
				} else {
					return nil
				}
			}

			next = valueGenerators[0].next()
		}

		return next
	}
}

/// A consumer-driven (pull-based) stream of values.
class AsyncSequence<T>: Stream<T>, Sequence {
	typealias GeneratorType = GeneratorOf<Promise<Event<T>>>

	let _generate: () -> GeneratorType
	init(_ generate: () -> GeneratorType) {
		self._generate = generate
	}

	/// Instantiates a generator that will instantly return Promise<Event<T>>
	/// objects, representing future events in the stream.
	///
	/// Work will only begin when the generated promises are actually resolved.
	/// Each promise may be evaluated in any order, or even skipped entirely.
	func generate() -> GeneratorType {
		return self._generate()
	}
	
	override class func empty() -> AsyncSequence<T> {
		return AsyncSequence {
			return GeneratorOf { nil }
		}
	}

	override class func single(x: T) -> AsyncSequence<T> {
		return AsyncSequence {
			return SequenceOf([
				Promise { Event.Next(Box(x)) },
				Promise { Event.Completed }
			]).generate()
		}
	}

	override func flattenScan<S, U>(initial: S, _ f: (S, T) -> (S?, Stream<U>)) -> AsyncSequence<U> {
		return AsyncSequence<U> {
			let g = _FlattenScanGenerator(disposable: SimpleDisposable(), scanFunc: f, valueGenerators: [], state: initial, selfGenerator: self.generate())
			return GeneratorOf(g)
		}
	}
}<commit_msg>Fix AsyncSequence.empty definition<commit_after>//
//  AsyncSequence.swift
//  RxSwift
//
//  Created by <NAME> on 2014-06-12.
//  Copyright (c) 2014 GitHub. All rights reserved.
//

import Foundation

// TODO: This lives outside of the definition below because of a crash in the
// compiler. Move it back within AsyncSequence once that's fixed.
struct _FlattenScanGenerator<S, T, U>: Generator {
	let disposable: Disposable
	let scanFunc: (S, T) -> (S?, Stream<U>)
	
	// TODO: Thread safety
	var valueGenerators: GeneratorOf<Promise<Event<U>>>[]
	var state: S
	var selfGenerator: GeneratorOf<Promise<Event<T>>>

	mutating func generateStreamFromValue(x: T) {
		let (newState, stream) = scanFunc(state, x)

		if let s = newState {
			state = s
		} else {
			disposable.dispose()
		}

		let seq = stream as AsyncSequence<U>
		valueGenerators.append(seq.generate())
	}

	mutating func next() -> Promise<Event<U>>? {
		if disposable.disposed {
			return nil
		}

		var next = valueGenerators[0].next()
		while next == nil {
			valueGenerators.removeAtIndex(0)
			if valueGenerators.isEmpty {
				if let promise = selfGenerator.next() {
					return promise.then { event in
						switch event {
						case let .Next(x):
							self.generateStreamFromValue(x)
							return Promise { .Completed }

						case let .Completed:
							return Promise { .Completed }

						case let .Error(error):
							self.disposable.dispose()
							return Promise { .Error(error) }
						}
					}
				} else {
					return nil
				}
			}

			next = valueGenerators[0].next()
		}

		return next
	}
}

/// A consumer-driven (pull-based) stream of values.
class AsyncSequence<T>: Stream<T>, Sequence {
	typealias GeneratorType = GeneratorOf<Promise<Event<T>>>

	let _generate: () -> GeneratorType
	init(_ generate: () -> GeneratorType) {
		self._generate = generate
	}

	/// Instantiates a generator that will instantly return Promise<Event<T>>
	/// objects, representing future events in the stream.
	///
	/// Work will only begin when the generated promises are actually resolved.
	/// Each promise may be evaluated in any order, or even skipped entirely.
	func generate() -> GeneratorType {
		return self._generate()
	}
	
	override class func empty() -> AsyncSequence<T> {
		return AsyncSequence {
			return GeneratorOf {
				Promise { Event.Completed }
			}
		}
	}

	override class func single(x: T) -> AsyncSequence<T> {
		return AsyncSequence {
			return SequenceOf([
				Promise { Event.Next(Box(x)) },
				Promise { Event.Completed }
			]).generate()
		}
	}

	override func flattenScan<S, U>(initial: S, _ f: (S, T) -> (S?, Stream<U>)) -> AsyncSequence<U> {
		return AsyncSequence<U> {
			let g = _FlattenScanGenerator(disposable: SimpleDisposable(), scanFunc: f, valueGenerators: [], state: initial, selfGenerator: self.generate())
			return GeneratorOf(g)
		}
	}
}
{'fraction_non_alphanumeric': 0.12296243798724309, 'fraction_numerical': 0.005315379163713678, 'mean_word_length': 3.2095451155853842, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 71, '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': '7028768', 'n_tokens_mistral': 2064, 'n_tokens_neox': 1721, 'n_words': 519}