rx_intro.swift

import Foundation
import RxSwift
import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true


class TryOut {
	private var timer: Timer!
	private let data = Array<Int>(1...5) // data to broadcast asynchronously
	public var outsideStream: Observable<Int>? // want to stream data through here for the outside world
	private var currentIndex = -1
	public var broadcast:((Observable<Int>) -> ())? = nil // second option since I could assign self.outsideStream = internalStream within execute function
	
	 init() {
		// call execute every 2 second
		timer = Timer.scheduledTimer(timeInterval: 2.0, target: self, selector: #selector(self.execute), userInfo: nil, repeats: true)
	}
	
	@objc private func execute() {
		//print("before currentIndex",self.currentIndex)
		self.currentIndex += 1
		print("after currentIndex",self.currentIndex)
		
		 let internalStream = Observable<Int>.create { event in
			
				if self.currentIndex < self.data.count {
					let valueToEmit = self.data[self.currentIndex]
					//print("emitted value", valueToEmit)
					event.onNext(valueToEmit)
				}
			
				if self.currentIndex + 1 > self.data.count {
					//print("internal completion")
					event.onCompleted()
					if self.timer.isValid {
						self.timer.invalidate()
					}
				}
				return Disposables.create()
		 }
	
		// self.outsideStream = internalStream //<-- why this didn't work
		broadcast?(internalStream)
	}
}


let participant = TryOut()
let disposeBag = DisposeBag()
participant.broadcast =  { stream in
	stream
		.filter({ $0 % 2 == 0 })
		.map({ $0 * 10 })
		.subscribe { (event) in
		    switch event {
	        	case .next(let value): print(value)
	      		case .error(_ ): break
      			case .completed: print("\n stream has completed \n")
	    	}
	}.disposed(by: disposeBag)
	
}

//
//participant.outsideStream?.subscribe { (event) in
//	switch event {
//		case .next(let value): print(value)
//		case .error(_ ): break
//		case .completed: print("stream has completed")
//	}
//}