<commit_before>import IDisposable
import Scope

public class Stream<T> : IStream
{
	public typealias ChunkType = T

	public init() {
	}

	public func dispose() {
		let disposables = self.disposables
			+ (ownershipSemantic == .Chained || ownershipSemantic == .Source ? self.downstreamDisposables : [])
			+ (ownershipSemantic == .Chained || ownershipSemantic == .Sink ? self.upstreamDisposables : [])

		self.subscribers = [Subscriber<ChunkType>]()
		self.downstreamDisposables = [IDisposable]()
		self.upstreamDisposables = [IDisposable]()
		self.disposables = [IDisposable]()

		for disposable in disposables {
			disposable.dispose()
		}
	}

	public func publish(_ chunk: ChunkType) -> Void {
		for subscriber in self.subscribers {
			subscriber.publish(chunk)
		}
	}

	public func subscribe(_ onChunk: @escaping (_ chunk: ChunkType) -> Void) -> Scope
	{
		let subscriber = Subscriber(callback: onChunk)
		self.subscribers.append(subscriber)
		return Scope(dispose: { self.removeSubscriber(subscriber: subscriber) })
	}

	func removeSubscriber(subscriber: Subscriber<ChunkType>) -> Void
	{
		let i = self.subscribers.firstIndex(where: { (x) -> Bool in
			return x === subscriber
		})
		self.subscribers.remove(at: i!)
	}

	public func addDownstreamDisposable(_ disposable: IDisposable) {
		downstreamDisposables.append(disposable)
	}

	public func addUpstreamDisposable(_ disposable: IDisposable) {
		upstreamDisposables.append(disposable)
	}

	public func addDisposable(_ disposable: IDisposable) {
		disposables.append(disposable)
	}

	public func configureOwnershipSemantic(_ semantic: OwnershipSemantic) {
		ownershipSemantic = semantic
	}<commit_msg>Make removeSubscriber safe to call when subscriber doesn't exist.<commit_after>import IDisposable
import Scope

public class Stream<T> : IStream
{
	public typealias ChunkType = T

	public init() {
	}

	public func dispose() {
		let disposables = self.disposables
			+ (ownershipSemantic == .Chained || ownershipSemantic == .Source ? self.downstreamDisposables : [])
			+ (ownershipSemantic == .Chained || ownershipSemantic == .Sink ? self.upstreamDisposables : [])

		self.subscribers = [Subscriber<ChunkType>]()
		self.downstreamDisposables = [IDisposable]()
		self.upstreamDisposables = [IDisposable]()
		self.disposables = [IDisposable]()

		for disposable in disposables {
			disposable.dispose()
		}
	}

	public func publish(_ chunk: ChunkType) -> Void {
		for subscriber in self.subscribers {
			subscriber.publish(chunk)
		}
	}

	public func subscribe(_ onChunk: @escaping (_ chunk: ChunkType) -> Void) -> Scope
	{
		let subscriber = Subscriber(callback: onChunk)
		self.subscribers.append(subscriber)
		return Scope(dispose: { self.removeSubscriber(subscriber: subscriber) })
	}

	func removeSubscriber(subscriber: Subscriber<ChunkType>) -> Void
	{
		if let i = self.subscribers.firstIndex(where: { (x) -> Bool in
			return x === subscriber
		}) {
			self.subscribers.remove(at: i)
		}
	}

	public func addDownstreamDisposable(_ disposable: IDisposable) {
		downstreamDisposables.append(disposable)
	}

	public func addUpstreamDisposable(_ disposable: IDisposable) {
		upstreamDisposables.append(disposable)
	}

	public func addDisposable(_ disposable: IDisposable) {
		disposables.append(disposable)
	}

	public func configureOwnershipSemantic(_ semantic: OwnershipSemantic) {
		ownershipSemantic = semantic
	}
{'fraction_non_alphanumeric': 0.10986082321587208, 'fraction_numerical': 0.0, 'mean_word_length': 4.784246575342466, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 19, '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': '4910430', 'n_tokens_mistral': 1213, 'n_tokens_neox': 1067, 'n_words': 233}