Skip to content

Commit

Permalink
Added subscribeBy(DisposableContainer, [...]) extensions.
Browse files Browse the repository at this point in the history
  • Loading branch information
mickverm committed Aug 10, 2021
1 parent 8396845 commit 8d2ee01
Show file tree
Hide file tree
Showing 8 changed files with 844 additions and 39 deletions.
4 changes: 1 addition & 3 deletions build.gradle.kts
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@ dependencies {
api("io.reactivex.rxjava3:rxjava:3.1.0")
implementation(kotlin("stdlib"))

testImplementation("org.funktionale:funktionale-partials:1.0.0-final")
testImplementation("junit:junit:4.12")
testImplementation("org.mockito:mockito-core:1.10.19")

Expand All @@ -60,7 +59,6 @@ val sourcesJar by tasks.creating(Jar::class) {
val dokka by tasks.getting(DokkaTask::class) {
outputFormat = "html"
outputDirectory = "$buildDir/javadoc"

}

//documentation
Expand Down Expand Up @@ -145,7 +143,7 @@ bintray {

setPublications(if (isRelease) release else snapshot)

// dryRun = true
// dryRun = true

with(pkg) {
userOrg = "reactivex"
Expand Down
118 changes: 89 additions & 29 deletions src/main/kotlin/io/reactivex/rxjava3/kotlin/subscribers.kt
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,11 @@ import io.reactivex.rxjava3.annotations.CheckReturnValue
import io.reactivex.rxjava3.annotations.SchedulerSupport
import io.reactivex.rxjava3.core.*
import io.reactivex.rxjava3.disposables.Disposable
import io.reactivex.rxjava3.disposables.DisposableContainer
import io.reactivex.rxjava3.functions.Action
import io.reactivex.rxjava3.functions.Consumer
import io.reactivex.rxjava3.internal.functions.Functions


private val onNextStub: (Any) -> Unit = {}
private val onErrorStub: (Throwable) -> Unit = {}
private val onCompleteStub: () -> Unit = {}
Expand All @@ -33,9 +33,9 @@ private fun (() -> Unit).asOnCompleteAction(): Action {
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Observable<T>.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Disposable = subscribe(onNext.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())

/**
Expand All @@ -45,9 +45,9 @@ fun <T : Any> Observable<T>.subscribeBy(
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Flowable<T>.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Disposable = subscribe(onNext.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())

/**
Expand All @@ -56,8 +56,8 @@ fun <T : Any> Flowable<T>.subscribeBy(
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Single<T>.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onSuccess: (T) -> Unit = onNextStub
onError: (Throwable) -> Unit = onErrorStub,
onSuccess: (T) -> Unit = onNextStub
): Disposable = subscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer())

/**
Expand All @@ -66,9 +66,9 @@ fun <T : Any> Single<T>.subscribeBy(
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Maybe<T>.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onSuccess: (T) -> Unit = onNextStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onSuccess: (T) -> Unit = onNextStub
): Disposable = subscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())

/**
Expand All @@ -77,8 +77,8 @@ fun <T : Any> Maybe<T>.subscribeBy(
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun Completable.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub
): Disposable = when {
// There are optimized versions of the completable Consumers, so we need to use the subscribe overloads
// here.
Expand All @@ -87,14 +87,74 @@ fun Completable.subscribeBy(
else -> subscribe(onComplete.asOnCompleteAction(), Consumer(onError))
}

/**
* Overloaded subscribe function that allows passing named parameters
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Observable<T>.subscribeBy(
container: DisposableContainer,
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Disposable = subscribe(onNext.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction(), container)

/**
* Overloaded subscribe function that allows passing named parameters
*/
@CheckReturnValue
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Flowable<T>.subscribeBy(
container: DisposableContainer,
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Disposable = subscribe(onNext.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction(), container)

/**
* Overloaded subscribe function that allows passing named parameters
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Single<T>.subscribeBy(
container: DisposableContainer,
onError: (Throwable) -> Unit = onErrorStub,
onSuccess: (T) -> Unit = onNextStub
): Disposable = subscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer(), container)

/**
* Overloaded subscribe function that allows passing named parameters
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Maybe<T>.subscribeBy(
container: DisposableContainer,
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onSuccess: (T) -> Unit = onNextStub
): Disposable =
subscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction(), container)

/**
* Overloaded subscribe function that allows passing named parameters
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun Completable.subscribeBy(
container: DisposableContainer,
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub
): Disposable = subscribe(onComplete.asOnCompleteAction(), onError.asOnErrorConsumer(), container)

/**
* Overloaded blockingSubscribe function that allows passing named parameters
*/
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Observable<T>.blockingSubscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Unit = blockingSubscribe(onNext.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())

/**
Expand All @@ -103,35 +163,35 @@ fun <T : Any> Observable<T>.blockingSubscribeBy(
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Flowable<T>.blockingSubscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Unit = blockingSubscribe(onNext.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())

/**
* Overloaded blockingSubscribe function that allows passing named parameters
*/
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Maybe<T>.blockingSubscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onSuccess: (T) -> Unit = onNextStub
) : Unit = blockingSubscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onSuccess: (T) -> Unit = onNextStub
): Unit = blockingSubscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer(), onComplete.asOnCompleteAction())

/**
* Overloaded blockingSubscribe function that allows passing named parameters
*/
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any> Single<T>.blockingSubscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onSuccess: (T) -> Unit = onNextStub
) : Unit = blockingSubscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer())
onError: (Throwable) -> Unit = onErrorStub,
onSuccess: (T) -> Unit = onNextStub
): Unit = blockingSubscribe(onSuccess.asConsumer(), onError.asOnErrorConsumer())

/**
* Overloaded blockingSubscribe function that allows passing named parameters
*/
@SchedulerSupport(SchedulerSupport.NONE)
fun Completable.blockingSubscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub
): Unit = blockingSubscribe(onComplete.asOnCompleteAction(), onError.asOnErrorConsumer())
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
/*
* Copyright (c) 2021-present, RxKotlin Contributors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package io.reactivex.rxjava3.kotlin

import io.reactivex.rxjava3.disposables.CompositeDisposable
import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection
import io.reactivex.rxjava3.subjects.CompletableSubject
import org.junit.Assert.*
import org.junit.Test
import java.io.IOException

class CompletableConsumersTest {

private fun CompositeDisposable.isEmpty(): Boolean = size() == 0
private fun CompositeDisposable.isNotEmpty(): Boolean = size() > 0

private val disposables = CompositeDisposable()
private val subject = CompletableSubject.create()
private val events = mutableListOf<Any>()

@Test
fun errorIntrospectionNormal() {
val disposable = subject.subscribeBy(disposables) as LambdaConsumerIntrospection
assertFalse(disposable.hasCustomOnError())
}

@Test
fun errorIntrospectionCustom() {
val disposable = subject.subscribeBy(disposables, onError = {}) as LambdaConsumerIntrospection
assertTrue(disposable.hasCustomOnError())
}

@Test
fun onErrorNormal() {
subject.subscribeBy(
disposables,
onError = events::add
)

assertTrue(disposables.isNotEmpty())
assertTrue(events.isEmpty())

subject.onComplete()

assertTrue(disposables.isEmpty())
assertTrue(events.isEmpty())
}

@Test
fun onErrorError() {
subject.subscribeBy(
disposables,
onError = events::add
)

assertTrue(disposables.isNotEmpty())
assertTrue(events.isEmpty())

subject.onError(IOException())

assertTrue(disposables.isEmpty())
assertEquals(1, events.size)
assertTrue(events[0] is IOException)
}

@Test
fun onCompleteNormal() {
subject.subscribeBy(
disposables,
onError = events::add,
onComplete = { events.add("completed") }
)

assertTrue(disposables.isNotEmpty())
assertTrue(events.isEmpty())

subject.onComplete()

assertTrue(disposables.isEmpty())
assertEquals(listOf("completed"), events)
}

@Test
fun onCompleteError() {
subject.subscribeBy(
disposables,
onError = events::add,
onComplete = { events.add("completed") }
)

assertTrue(disposables.isNotEmpty())
assertTrue(events.isEmpty())

subject.onError(IOException())

assertTrue(disposables.isEmpty())
assertEquals(1, events.size)
assertTrue(events[0] is IOException)
}

@Test
fun onCompleteDispose() {
val disposable = subject.subscribeBy(
disposables,
onError = events::add,
onComplete = { events.add("completed") }
)

assertFalse(disposable.isDisposed)
assertTrue(disposables.isNotEmpty())
assertTrue(events.isEmpty())

disposable.dispose()

assertTrue(disposable.isDisposed)
assertTrue(disposables.isEmpty())
assertTrue(events.isEmpty())
}
}
12 changes: 5 additions & 7 deletions src/test/kotlin/io/reactivex/rxjava3/kotlin/ExtensionTests.kt
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.core.ObservableEmitter
import io.reactivex.rxjava3.schedulers.TestScheduler
import io.reactivex.rxjava3.functions.*
import org.funktionale.partials.invoke
import org.junit.Assert.assertEquals
import org.junit.Assert.fail
import org.junit.Test
Expand Down Expand Up @@ -248,11 +247,6 @@ class ExtensionTests : KotlinTests() {
inOrder.verifyNoMoreInteractions()
}

val funOnSubscribe: (Int, ObservableEmitter<in String>) -> Unit = { counter, subscriber ->
subscriber.onNext("hello_$counter")
subscriber.onComplete()
}

val asyncObservable: (ObservableEmitter<in Int>) -> Unit = { subscriber ->
thread {
Thread.sleep(50)
Expand All @@ -270,7 +264,11 @@ class ExtensionTests : KotlinTests() {
get() = listOf(1, 3, 2, 5, 4).toObservable()

val onSubscribe: (ObservableEmitter<in String>) -> Unit
get() = funOnSubscribe(p1 = counter++) // partial applied function
get() = {
it.onNext("hello_$counter")
it.onComplete()
counter ++
}

val observable: Observable<String>
get() = Observable.create(onSubscribe)
Expand Down
Loading

0 comments on commit 8d2ee01

Please sign in to comment.