ReactiveX/RxSwift
1
1
Fork 0
mirror of https://github.com/ReactiveX/RxSwift.git synced 2024-10-04 05:57:23 +03:00
Code Issues Projects Releases Wiki Activity

Adds PrimitiveSequence operators: observeOn. ObservableType operators: asSingle, asMaybe, asCompleteable.

Browse Source
This commit is contained in:
Krunoslav Zaher 2017-03-12 19:07:24 +01:00
parent 2f0a21546b
commit f3a7e43922
11 changed files with 554 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.jazzy.yml
View File

@ -53,6 +53,8 @@ custom_categories:
- AddRef
- Amb
- AnonymousObservable
- AsMaybe
- AsSingle
- Buffer
- Catch
- CombineLatest+arity

20
Rx.xcodeproj/project.pbxproj
View File

@ -698,6 +698,14 @@
C898147E1E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift in Sources */ = {isa = PBXBuildFile; fileRef = C898147D1E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift */; };
C898147F1E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift in Sources */ = {isa = PBXBuildFile; fileRef = C898147D1E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift */; };
C89814801E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift in Sources */ = {isa = PBXBuildFile; fileRef = C898147D1E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift */; };
C89814821E75B77B0035949C /* AsMaybe.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814811E75B77B0035949C /* AsMaybe.swift */; };
C89814831E75B77B0035949C /* AsMaybe.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814811E75B77B0035949C /* AsMaybe.swift */; };
C89814841E75B77B0035949C /* AsMaybe.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814811E75B77B0035949C /* AsMaybe.swift */; };
C89814851E75B77B0035949C /* AsMaybe.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814811E75B77B0035949C /* AsMaybe.swift */; };
C89814871E75BE590035949C /* AsSingle.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814861E75BE590035949C /* AsSingle.swift */; };
C89814881E75BE590035949C /* AsSingle.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814861E75BE590035949C /* AsSingle.swift */; };
C89814891E75BE590035949C /* AsSingle.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814861E75BE590035949C /* AsSingle.swift */; };
C898148A1E75BE590035949C /* AsSingle.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89814861E75BE590035949C /* AsSingle.swift */; };
C89AB1731DAAC1680065FBE6 /* ControlTarget.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89AB1711DAAC1680065FBE6 /* ControlTarget.swift */; };
C89AB1741DAAC1680065FBE6 /* ControlTarget.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89AB1711DAAC1680065FBE6 /* ControlTarget.swift */; };
C89AB1751DAAC1680065FBE6 /* ControlTarget.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89AB1711DAAC1680065FBE6 /* ControlTarget.swift */; };
@ -1877,6 +1885,8 @@
C89814771E75A7D70035949C /* PrimitiveSequence+Zip+arity.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "PrimitiveSequence+Zip+arity.swift"; sourceTree = "<group>"; };
C898147C1E75A98A0035949C /* PrimitiveSequenceTest+zip+arity.tt */ = {isa = PBXFileReference; lastKnownFileType = text; path = "PrimitiveSequenceTest+zip+arity.tt"; sourceTree = "<group>"; };
C898147D1E75AD380035949C /* PrimitiveSequenceTest+zip+arity.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "PrimitiveSequenceTest+zip+arity.swift"; sourceTree = "<group>"; };
C89814811E75B77B0035949C /* AsMaybe.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AsMaybe.swift; sourceTree = "<group>"; };
C89814861E75BE590035949C /* AsSingle.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AsSingle.swift; sourceTree = "<group>"; };
C89AB1711DAAC1680065FBE6 /* ControlTarget.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ControlTarget.swift; sourceTree = "<group>"; };
C89AB1A51DAAC25A0065FBE6 /* RxCocoaObjCRuntimeError+Extensions.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RxCocoaObjCRuntimeError+Extensions.swift"; sourceTree = "<group>"; };
C89AB1AB1DAAC3350065FBE6 /* ControlEvent.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ControlEvent.swift; sourceTree = "<group>"; };
@ -2326,6 +2336,8 @@
C8093C931B8A72BE0088E94D /* Zip+arity.tt */,
C8C3D9FD1B935EDF004D233E /* Zip+Collection.swift */,
252FC1DE1E0DC64C00D28877 /* SwitchIfEmpty.swift */,
C89814811E75B77B0035949C /* AsMaybe.swift */,
C89814861E75BE590035949C /* AsSingle.swift */,
);
path = Implementations;
sourceTree = "<group>";
@ -4337,6 +4349,7 @@
C8093CD41B8A72BE0088E94D /* Disposable.swift in Sources */,
C8093CEE1B8A72BE0088E94D /* SingleAssignmentDisposable.swift in Sources */,
C849BE2C1BAB5D070019AD27 /* ObservableConvertibleType.swift in Sources */,
C89814881E75BE590035949C /* AsSingle.swift in Sources */,
C8C3DA0A1B93941E004D233E /* Error.swift in Sources */,
C8093D9C1B8A72BE0088E94D /* SchedulerServices+Emulation.swift in Sources */,
C80EEC351D42D06E00131C39 /* DispatchQueueConfiguration.swift in Sources */,
@ -4362,6 +4375,7 @@
CB883B461BE256D4000AC2EE /* BooleanDisposable.swift in Sources */,
C84B38EF1BA433CD001B7D88 /* Generate.swift in Sources */,
C8093D161B8A72BE0088E94D /* Deferred.swift in Sources */,
C89814831E75B77B0035949C /* AsMaybe.swift in Sources */,
C8093DA41B8A72BE0088E94D /* ReplaySubject.swift in Sources */,
C8C3D9FF1B935EDF004D233E /* Zip+Collection.swift in Sources */,
C8093D641B8A72BE0088E94D /* Observable+Time.swift in Sources */,
@ -4577,6 +4591,7 @@
C8093CED1B8A72BE0088E94D /* SingleAssignmentDisposable.swift in Sources */,
C89814781E75A7D70035949C /* PrimitiveSequence+Zip+arity.swift in Sources */,
C849BE2B1BAB5D070019AD27 /* ObservableConvertibleType.swift in Sources */,
C89814871E75BE590035949C /* AsSingle.swift in Sources */,
C8C3DA091B93941E004D233E /* Error.swift in Sources */,
C8093D9B1B8A72BE0088E94D /* SchedulerServices+Emulation.swift in Sources */,
C80EEC341D42D06E00131C39 /* DispatchQueueConfiguration.swift in Sources */,
@ -4602,6 +4617,7 @@
C8093D151B8A72BE0088E94D /* Deferred.swift in Sources */,
C8093DA31B8A72BE0088E94D /* ReplaySubject.swift in Sources */,
C8C3D9FE1B935EDF004D233E /* Zip+Collection.swift in Sources */,
C89814821E75B77B0035949C /* AsMaybe.swift in Sources */,
C8093D631B8A72BE0088E94D /* Observable+Time.swift in Sources */,
C8093CFB1B8A72BE0088E94D /* ObservableType+Extensions.swift in Sources */,
C8093D491B8A72BE0088E94D /* Throttle.swift in Sources */,
@ -4745,6 +4761,7 @@
C8F0BF961BBBFB8B001B112F /* Disposable.swift in Sources */,
C8F0BF971BBBFB8B001B112F /* SingleAssignmentDisposable.swift in Sources */,
C89461751BC6C1210055219D /* ObservableConvertibleType.swift in Sources */,
C898148A1E75BE590035949C /* AsSingle.swift in Sources */,
C8F0BF981BBBFB8B001B112F /* Error.swift in Sources */,
C8F0BF991BBBFB8B001B112F /* SchedulerServices+Emulation.swift in Sources */,
C80EEC371D42D06E00131C39 /* DispatchQueueConfiguration.swift in Sources */,
@ -4770,6 +4787,7 @@
CB883B481BE256D4000AC2EE /* BooleanDisposable.swift in Sources */,
C8F0BFA41BBBFB8B001B112F /* Generate.swift in Sources */,
C8F0BFA51BBBFB8B001B112F /* Deferred.swift in Sources */,
C89814851E75B77B0035949C /* AsMaybe.swift in Sources */,
C8F0BFA61BBBFB8B001B112F /* ReplaySubject.swift in Sources */,
C8F0BFA71BBBFB8B001B112F /* Zip+Collection.swift in Sources */,
C8F0BFA81BBBFB8B001B112F /* Observable+Time.swift in Sources */,
@ -5098,6 +5116,7 @@
D2EBEB281BB9B6C1003A27DC /* Zip.swift in Sources */,
D2EBEB3E1BB9B6D8003A27DC /* SerialDispatchQueueScheduler.swift in Sources */,
C89461761BC6C1220055219D /* ObservableConvertibleType.swift in Sources */,
C89814891E75BE590035949C /* AsSingle.swift in Sources */,
D2EBEAF71BB9B6B2003A27DC /* ScheduledDisposable.swift in Sources */,
D2EBEAE11BB9B697003A27DC /* ImmediateSchedulerType.swift in Sources */,
C80EEC361D42D06E00131C39 /* DispatchQueueConfiguration.swift in Sources */,
@ -5123,6 +5142,7 @@
CB883B471BE256D4000AC2EE /* BooleanDisposable.swift in Sources */,
D2EBEB001BB9B6BA003A27DC /* Catch.swift in Sources */,
D2EBEB161BB9B6C1003A27DC /* ObserveOnSerialDispatchQueue.swift in Sources */,
C89814841E75B77B0035949C /* AsMaybe.swift in Sources */,
D2EBEB061BB9B6C1003A27DC /* Debug.swift in Sources */,
D2EBEB341BB9B6D2003A27DC /* AnonymousObserver.swift in Sources */,
D2EBEB421BB9B6DE003A27DC /* ReplaySubject.swift in Sources */,

49
RxSwift/Observables/Implementations/AsMaybe.swift Normal file
View File

@ -0,0 +1,49 @@
//
// AsMaybe.swift
// RxSwift
//
// Created by Krunoslav Zaher on 3/12/17.
// Copyright © 2017 Krunoslav Zaher. All rights reserved.
//
fileprivate final class AsMaybeSink<O: ObserverType> : Sink<O>, ObserverType {
typealias ElementType = O.E
typealias E = ElementType
private var _element: Event<E>? = nil
func on(_ event: Event<E>) {
switch event {
case .next:
if _element != nil {
forwardOn(.error(RxError.moreThanOneElement))
dispose()
}
_element = event
case .error:
forwardOn(event)
dispose()
case .completed:
if let element = _element {
forwardOn(element)
}
forwardOn(.completed)
dispose()
}
}
}
final class AsMaybe<Element>: Producer<Element> {
fileprivate let _source: Observable<Element>
init(source: Observable<Element>) {
_source = source
}
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AsMaybeSink(observer: observer, cancel: cancel)
let subscription = _source.subscribe(sink)
return (sink: sink, subscription: subscription)
}
}

52
RxSwift/Observables/Implementations/AsSingle.swift Normal file
View File

@ -0,0 +1,52 @@
//
// AsSingle.swift
// RxSwift
//
// Created by Krunoslav Zaher on 3/12/17.
// Copyright © 2017 Krunoslav Zaher. All rights reserved.
//
fileprivate final class AsSingleSink<O: ObserverType> : Sink<O>, ObserverType {
typealias ElementType = O.E
typealias E = ElementType
private var _element: Event<E>? = nil
func on(_ event: Event<E>) {
switch event {
case .next:
if _element != nil {
forwardOn(.error(RxError.moreThanOneElement))
dispose()
}
_element = event
case .error:
forwardOn(event)
dispose()
case .completed:
if let element = _element {
forwardOn(element)
forwardOn(.completed)
}
else {
forwardOn(.error(RxError.noElements))
}
dispose()
}
}
}
final class AsSingle<Element>: Producer<Element> {
fileprivate let _source: Observable<Element>
init(source: Observable<Element>) {
_source = source
}
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AsSingleSink(observer: observer, cancel: cancel)
let subscription = _source.subscribe(sink)
return (sink: sink, subscription: subscription)
}
}

18
RxSwift/Observables/Implementations/SingleAsync.swift
View File

@ -6,7 +6,7 @@
// Copyright © 2015 Krunoslav Zaher. All rights reserved.
//
final class SingleAsyncSink<O: ObserverType> : Sink<O>, ObserverType {
fileprivate final class SingleAsyncSink<O: ObserverType> : Sink<O>, ObserverType {
typealias ElementType = O.E
typealias Parent = SingleAsync<ElementType>
typealias E = ElementType
@ -34,22 +34,22 @@ final class SingleAsyncSink<O: ObserverType> : Sink<O>, ObserverType {
return
}
if _seenValue == false {
_seenValue = true
forwardOn(.next(value))
} else {
if _seenValue {
forwardOn(.error(RxError.moreThanOneElement))
dispose()
return
}
_seenValue = true
forwardOn(.next(value))
case .error:
forwardOn(event)
dispose()
case .completed:
if (!_seenValue) {
forwardOn(.error(RxError.noElements))
} else {
if (_seenValue) {
forwardOn(.completed)
} else {
forwardOn(.error(RxError.noElements))
}
dispose()
}

17
RxSwift/Observables/Observable+StandardSequenceOperators.swift
View File

@ -257,12 +257,12 @@ extension ObservableType {
extension ObservableType {
/**
Returns a sequence emitting only item _n_ emitted by an Observable
Returns a sequence emitting only element _n_ emitted by an Observable
- seealso: [elementAt operator on reactivex.io](http://reactivex.io/documentation/operators/elementat.html)
- parameter index: The index of the required item (starting from 0).
- returns: An observable sequence that emits the desired item as its own sole emission.
- parameter index: The index of the required element (starting from 0).
- returns: An observable sequence that emits the desired element as its own sole emission.
*/
public func elementAt(_ index: Int)
-> Observable<E> {
@ -275,12 +275,12 @@ extension ObservableType {
extension ObservableType {
/**
The single operator is similar to first, but throws a `RxError.NoElements` or `RxError.MoreThanOneElement`
if the source Observable does not emit exactly one item before successfully completing.
The single operator is similar to first, but throws a `RxError.noElements` or `RxError.moreThanOneElement`
if the source Observable does not emit exactly one element before successfully completing.
- seealso: [single operator on reactivex.io](http://reactivex.io/documentation/operators/first.html)
- returns: An observable sequence that emits a single item or throws an exception if more (or none) of them are emitted.
- returns: An observable sequence that emits a single element or throws an exception if more (or none) of them are emitted.
*/
public func single()
-> Observable<E> {
@ -289,18 +289,17 @@ extension ObservableType {
/**
The single operator is similar to first, but throws a `RxError.NoElements` or `RxError.MoreThanOneElement`
if the source Observable does not emit exactly one item before successfully completing.
if the source Observable does not emit exactly one element before successfully completing.
- seealso: [single operator on reactivex.io](http://reactivex.io/documentation/operators/first.html)
- parameter predicate: A function to test each source element for a condition.
- returns: An observable sequence that emits a single item or throws an exception if more (or none) of them are emitted.
- returns: An observable sequence that emits a single element or throws an exception if more (or none) of them are emitted.
*/
public func single(_ predicate: @escaping (E) throws -> Bool)
-> Observable<E> {
return SingleAsync(source: asObservable(), predicate: predicate)
}
}
// MARK: groupBy

56
RxSwift/Units/PrimitiveSequence.swift
View File

@ -433,6 +433,26 @@ extension PrimitiveSequence {
return PrimitiveSequence(raw: source.observeOn(scheduler))
}
/**
Wraps the source sequence in order to run its subscription and unsubscription logic on the specified
scheduler.
This operation is not commonly used.
This only performs the side-effects of subscription and unsubscription on the specified scheduler.
In order to invoke observer callbacks on a `scheduler`, use `observeOn`.
- seealso: [subscribeOn operator on reactivex.io](http://reactivex.io/documentation/operators/subscribeon.html)
- parameter scheduler: Scheduler to perform subscription and unsubscription actions on.
- returns: The source sequence whose subscriptions and unsubscriptions happen on the specified scheduler.
*/
public func subscribeOn(_ scheduler: ImmediateSchedulerType)
-> PrimitiveSequence<Trait, Element> {
return PrimitiveSequence(raw: source.subscribeOn(scheduler))
}
/**
Continues an observable sequence that is terminated by an error with the observable sequence produced by the handler.
@ -532,3 +552,39 @@ extension PrimitiveSequenceType where TraitType == CompleteableTrait, ElementTyp
return PrimitiveSequence(raw: Observable.empty())
}
}
extension ObservableType {
/**
The `asSingle` operator throws a `RxError.noElements` or `RxError.moreThanOneElement`
if the source Observable does not emit exactly one element before successfully completing.
- seealso: [single operator on reactivex.io](http://reactivex.io/documentation/operators/first.html)
- returns: An observable sequence that emits a single element or throws an exception if more (or none) of them are emitted.
*/
public func asSingle() -> Single<E> {
return PrimitiveSequence(raw: AsSingle(source: self.asObservable()))
}
/**
The `asMaybe` operator throws a ``RxError.moreThanOneElement`
if the source Observable does not emit at most one element before successfully completing.
- seealso: [single operator on reactivex.io](http://reactivex.io/documentation/operators/first.html)
- returns: An observable sequence that emits a single element, completes or throws an exception if more of them are emitted.
*/
public func asMaybe() -> Maybe<E> {
return PrimitiveSequence(raw: AsMaybe(source: self.asObservable()))
}
}
extension ObservableType where E == Never {
/**
- returns: An observable sequence that completes.
*/
public func asCompleteable()
-> Completeable {
return PrimitiveSequence(raw: self.asObservable())
}
}

13
Sources/AllTestz/main.swift
View File

@ -105,11 +105,24 @@ final class PrimitiveSequenceTest_ : PrimitiveSequenceTest, RxTestCase {
("testSingle_map_producesSingleElement", PrimitiveSequenceTest.testSingle_map_producesSingleElement),
("testSingle_flatMap_producesSingleElement", PrimitiveSequenceTest.testSingle_flatMap_producesSingleElement),
("testSingle_observeOn_producesSingleElement", PrimitiveSequenceTest.testSingle_observeOn_producesSingleElement),
("testSingle_subscribeOn_producesSingleElement", PrimitiveSequenceTest.testSingle_subscribeOn_producesSingleElement),
("testSingle_catchError_producesSingleElement", PrimitiveSequenceTest.testSingle_catchError_producesSingleElement),
("testSingle_retry_producesSingleElement", PrimitiveSequenceTest.testSingle_retry_producesSingleElement),
("testSingle_retryWhen1_producesSingleElement", PrimitiveSequenceTest.testSingle_retryWhen1_producesSingleElement),
("testSingle_retryWhen2_producesSingleElement", PrimitiveSequenceTest.testSingle_retryWhen2_producesSingleElement),
("testSingle_timer_producesSingleElement", PrimitiveSequenceTest.testSingle_timer_producesSingleElement),
("testAsSingle_Empty", PrimitiveSequenceTest.testAsSingle_Empty),
("testAsSingle_One", PrimitiveSequenceTest.testAsSingle_One),
("testAsSingle_Many", PrimitiveSequenceTest.testAsSingle_Many),
("testAsSingle_Error", PrimitiveSequenceTest.testAsSingle_Error),
("testAsSingle_Error2", PrimitiveSequenceTest.testAsSingle_Error2),
("testAsMaybe_Empty", PrimitiveSequenceTest.testAsMaybe_Empty),
("testAsMaybe_One", PrimitiveSequenceTest.testAsMaybe_One),
("testAsMaybe_Many", PrimitiveSequenceTest.testAsMaybe_Many),
("testAsMaybe_Error", PrimitiveSequenceTest.testAsMaybe_Error),
("testAsMaybe_Error2", PrimitiveSequenceTest.testAsMaybe_Error2),
("testAsCompleteable_Empty", PrimitiveSequenceTest.testAsCompleteable_Empty),
("testAsCompleteable_Error", PrimitiveSequenceTest.testAsCompleteable_Error),
] }
}

1
Sources/RxSwift/AsMaybe.swift Symbolic link
View File

@ -0,0 +1 @@
../../RxSwift/Observables/Implementations/AsMaybe.swift

1
Sources/RxSwift/AsSingle.swift Symbolic link
View File

@ -0,0 +1 @@
../../RxSwift/Observables/Implementations/AsSingle.swift

343
Tests/RxSwiftTests/PrimitiveSequenceTest.swift
View File

@ -568,6 +568,22 @@ extension PrimitiveSequenceTest {
])
}
func testSingle_subscribeOn_producesSingleElement() {
let scheduler = TestScheduler(initialClock: 0)
let res = scheduler.start { () -> Observable<Int> in
let singleResult: Single<Int> = Single.just(1)
.subscribeOn(scheduler)
return singleResult.asObservable()
}
XCTAssertEqual(res.events, [
next(201, 1),
completed(201)
])
}
func testSingle_catchError_producesSingleElement() {
let singleResult: Single<Int> = Single.error(testError)
.catchError { _ in Single.just(2) }
@ -657,6 +673,333 @@ extension PrimitiveSequenceTest {
}
}
extension PrimitiveSequenceTest {
func testAsSingle_Empty() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
completed(250),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let single: Single<Int> = xs.asSingle()
return single.asObservable()
}
XCTAssertEqual(res.events, [
error(250, RxError.noElements)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 250)
])
}
func testAsSingle_One() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
next(210, 2),
completed(250),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let single: Single<Int> = xs.asSingle()
return single.asObservable()
}
XCTAssertEqual(res.events, [
next(250, 2),
completed(250)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 250)
])
}
func testAsSingle_Many() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
next(210, 2),
next(220, 3),
completed(250),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let single: Single<Int> = xs.asSingle()
return single.asObservable()
}
XCTAssertEqual(res.events, [
error(220, RxError.moreThanOneElement)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 220)
])
}
func testAsSingle_Error() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
error(210, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let single: Single<Int> = xs.asSingle()
return single.asObservable()
}
XCTAssertEqual(res.events, [
error(210, testError)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 210)
])
}
func testAsSingle_Error2() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
next(205, 2),
error(210, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let single: Single<Int> = xs.asSingle()
return single.asObservable()
}
XCTAssertEqual(res.events, [
error(210, testError)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 210)
])
}
#if TRACE_RESOURCES
func testAsSingleReleasesResourcesOnComplete() {
_ = Observable<Int>.just(1).asSingle().subscribe({ _ in })
}
func testAsSingleReleasesResourcesOnError1() {
_ = Observable<Int>.error(testError).asSingle().subscribe({ _ in })
}
func testAsSingleReleasesResourcesOnError2() {
_ = Observable<Int>.of(1, 2).asSingle().subscribe({ _ in })
}
#endif
}
extension PrimitiveSequenceTest {
func testAsMaybe_Empty() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
completed(250),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let maybe: Maybe<Int> = xs.asMaybe()
return maybe.asObservable()
}
XCTAssertEqual(res.events, [
completed(250)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 250)
])
}
func testAsMaybe_One() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
next(210, 2),
completed(250),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let maybe: Maybe<Int> = xs.asMaybe()
return maybe.asObservable()
}
XCTAssertEqual(res.events, [
next(250, 2),
completed(250)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 250)
])
}
func testAsMaybe_Many() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
next(210, 2),
next(220, 3),
completed(250),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let maybe: Maybe<Int> = xs.asMaybe()
return maybe.asObservable()
}
XCTAssertEqual(res.events, [
error(220, RxError.moreThanOneElement)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 220)
])
}
func testAsMaybe_Error() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
error(210, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let maybe: Maybe<Int> = xs.asMaybe()
return maybe.asObservable()
}
XCTAssertEqual(res.events, [
error(210, testError)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 210)
])
}
func testAsMaybe_Error2() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
next(150, 1),
next(205, 2),
error(210, testError)
])
let res = scheduler.start { () -> Observable<Int> in
let maybe: Maybe<Int> = xs.asMaybe()
return maybe.asObservable()
}
XCTAssertEqual(res.events, [
error(210, testError)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 210)
])
}
#if TRACE_RESOURCES
func testAsMaybeReleasesResourcesOnComplete1() {
_ = Observable<Int>.empty().asMaybe().subscribe({ _ in })
}
func testAsMaybeReleasesResourcesOnComplete2() {
_ = Observable<Int>.just(1).asMaybe().subscribe({ _ in })
}
func testAsMaybeReleasesResourcesOnError1() {
_ = Observable<Int>.error(testError).asMaybe().subscribe({ _ in })
}
func testAsMaybeReleasesResourcesOnError2() {
_ = Observable<Int>.of(1, 2).asMaybe().subscribe({ _ in })
}
#endif
}
extension PrimitiveSequenceTest {
func testAsCompleteable_Empty() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
completed(250, Never.self),
error(260, testError)
])
let res = scheduler.start { () -> Observable<Never> in
let completeable: Completeable = xs.asCompleteable()
return completeable.asObservable()
}
XCTAssertEqual(res.events, [
completed(250)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 250)
])
}
func testAsCompleteable_Error() {
let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createHotObservable([
error(210, testError, Never.self)
])
let res = scheduler.start { () -> Observable<Never> in
let completeable: Completeable = xs.asCompleteable()
return completeable.asObservable()
}
XCTAssertEqual(res.events, [
error(210, testError)
])
XCTAssertEqual(xs.subscriptions, [
Subscription(200, 210)
])
}
#if TRACE_RESOURCES
func testAsCompleteableReleasesResourcesOnComplete() {
_ = Observable<Never>.empty().asCompleteable().subscribe({ _ in })
}
func testAsCompleteableReleasesResourcesOnError() {
_ = Observable<Never>.error(testError).asCompleteable().subscribe({ _ in })
}
#endif
}
extension Never: Equatable {
}