如何将rxjava2-Zip函数的arity(从Single/Observable)推广到n个Nullable参数而不丢

Question

需要解决的两个主要问题：

1( 类型检查丢失

使用数组参数Single.zip()版本，我会丢失强类型参数。

2( 源参数不能为Null

我无法将可为null的源值作为Single.zip()函数的参数发送

3( 我想要一个替代方法，以代替未键入Object[]的方法：

4(我不想要可变对象，我不想在我的类中使用var。我想使用val

public static <T, R> Single<R> zipArray(Function<? super Object[], ? extends R> zipper, SingleSource<? extends T>... sources) ...

在haskell中，有一个问题与如何实现广义的"zipn"；以及"；解压缩"；在Haskell？：

在haskell中，我可以用应用函子实现这一点：

f <$> a1 <*> a2 <*> a3 <*> a4 <*> a5 <*> a6 <*> a7 <*> a8 <*> a9 <*> a10 <*> a11

为f :: Int -> Int -> Int -> Int -> Int -> Int -> Int -> String -> String -> String -> Int

和对应于每种类型的CCD_ 5值

库中有一个类似函数的列表：

• 有两个参数：

  public static <T1, T2, R> Single<R> zip(SingleSource<? extends T1> source1, SingleSource<? extends T2> source2,BiFunction<? super T1, ? super T2, ? extends R> zipper) {
  ObjectHelper.requireNonNull(source1, "source1 is null");
  ObjectHelper.requireNonNull(source2, "source2 is null");
  return zipArray(Functions.toFunction(zipper), source1, source2);
  }
  

• 有三个：

  public static <T1, T2, T3, R> Single<R> zip(
  SingleSource<? extends T1> source1, SingleSource<? extends T2> source2,
  SingleSource<? extends T3> source3,
  Function3<? super T1, ? super T2, ? super T3, ? extends R> zipper)
  

等等…

在所有这些情况下，都可以，因为每个参数都是类型化的但有一个限制，直到9个单一来源

在我们的项目中，我们需要更多的资源，因为我们有很多服务需要异步(在我们的案例中是11个参数(。

但问题是，这些论点失去了它们的强类型，更糟糕的是，其中一些可能是Nullable

例如，我们想解决这个用例：

//Given
val bothSubscribed = CountDownLatch(2) // Change this value to 0 to run the test faster
val subscribeThreadsStillRunning = CountDownLatch(1) // Change this value to 0 to run the test faster
val service = { s1: String,
s2: Int,
s3: String?,
s4: Int,
s5: String,
s6: String,
s7: String,
s8: String,
s9: String,
s10: String?,
s11: String ->
val result =
listOf(s1, "$s2", s3 ?: "none", "$s4", s5, s6, s7, s8, s9, s10 ?: "none", s11).joinToString(separator = ";")
Single.just("Values:$result")
}
val createSingle = { value: String ->
Observable
.create<String> { emitter ->
println("Parallel subscribe $value on ${Thread.currentThread().name}")
bothSubscribed.countDown()
subscribeThreadsStillRunning.await(20, TimeUnit.SECONDS)
emitter.onNext(value)
emitter.onComplete()
}
.singleOrError()
.subscribeOn(io())
}
val s1 = createSingle("v1")
val s2 = Single.just(2)
val s3 = null
val s4 = Single.just(4)
val s5 = createSingle("v5")
val s6 = createSingle("v6")
val s7 = createSingle("v7")
val s8 = createSingle("v8")
val s9 = createSingle("v9")
val s10 = null
val s11 = createSingle("v11")
//When
val result = Single.zipArray(
listOf(
s1,
s2,
s3,
s4,
s5,
s6,
s7,
s8,
s9,
s10,
s11
)
) { arrayResult ->
service(
arrayResult[0] as String,
arrayResult[1] as String,
arrayResult[2] as String?,
arrayResult[3] as String,
arrayResult[4] as String,
arrayResult[5] as String,
arrayResult[6] as String,
arrayResult[7] as String,
arrayResult[8] as String,
arrayResult[9] as String?,
arrayResult[10] as String
)
}
//Then
result
.test()
.awaitDone(50, TimeUnit.SECONDS)
.assertSubscribed()
.assertValues("Values:v1;2;none;4;v5;v6;v7;v8;v9;none;v11")

正如你所看到的，如果我这样做，可能会出现问题，例如：

arrayResult[0] as String,
arrayResult[1] as Int,
arrayResult[2] as String?,
arrayResult[3] as Int,
arrayResult[4] as String,
arrayResult[5] as String,
arrayResult[6] as String,
arrayResult[7] as String,
arrayResult[8] as String,
arrayResult[9] as String?,
arrayResult[10] as String

失败的原因：

1(Single.zip()函数都不能将可为null的值作为参数

2( 您可以在数组中更改值的顺序，但它可能会因为类型检查强制转换而失败

Answer 1

一个有11个参数的函数是不干净代码的一个很好的例子。相反，你应该考虑建立一个模型来满足你的需求。像这样，您也可以为每个参数提供有意义的名称。

data class MyObject(...)
class MyMutableObject {
private lateinit var param0: String
private var param1: Int
...
fun setParam0(value: String) {
param0 = value
}
fun setParam1(value: Int) {
param1 = value
}
...
fun toMyObject() = MyObject(
param0,
param1,
...
) 
}

有了这个模型，您可以在每个源上使用zipWith()运算符。

Single.just(MyMutableObject())
.zipWith(source0, MyMutableObject::setParam0)
.zipWith(source1, MyMutableObject::setParam1)
...
.map(MyMutableObject::toMyObject)

如果您考虑将可为空性抽象为Maybe，那么您可以简单地定义一个扩展函数来接收带数据或不带数据的Maybe，并适当地映射它。

inline fun <T, U, R> Single<T>.zipWith(
other: MaybeSource<U>,
crossinline zipper: (T, U) -> R
) = other.zipWith(toMaybe()) { t, u -> zipper(t, u) }
.switchIfEmpty(this)

Answer 2

我使用实现了这个目标

1. Kotlin扩展函数
2. Curried函数(Kotlin允许(
3. 部分应用程序(Kotlin也允许(
4. 函子和应用函子概念(单类和可观察类是应用函子(
5. 将其混合在一起：

首先，zipOver函数，用于不可为null的值：

/**
* Returns a Single that is the result of applying the function inside the context (a Single in this case).
* This function is curried and will be used as an Applicative Functor, so each argument will be given
* one by one
* @param <B> the result value type
* @param applicativeValue
*            a Single that contains the input value of the function
* @return the Single returned when the function is applied to the applicative value.
* Each application will be executed on <b>a new thread</b> if and only if the Single is subscribed on a specific scheduler
*/
infix fun <A, B> Single<(A) -> (B)>.zipOver(applicativeValue: Single<A>): Single<B> =
Single.zip(this, applicativeValue, BiFunction { f, a -> f(a) })

然后，zipOverNullable表示可为空的值：

/**
* Returns a Single that is the result of applying the function inside the context (a Single in this case).
* This function is curried and will be used as an Applicative Functor, so each argument will be given
* one by one
* @param <B> the result value type
* @param applicativeValue
*            a Single that contains the input value of the function and it can be null
* @return the Single returned when the function is applied to the applicative value even when
* it is null.
* Each application will be executed on <b>a new thread</b> if and only if the Single is subscribed on a specific scheduler
*/
infix fun <A, B> Single<(A?) -> (B)>.zipOverNullable(applicativeValue: Single<A>?): Single<B> =
when {
applicativeValue != null -> Single.zip(this, applicativeValue, BiFunction { f, a -> f(a) })
else -> this.map { it(null) }
}

我使用了org.funnational.currying作为curried()函数

把这两者结合起来，你可以写：

//Given
val bothSubscribed = CountDownLatch(0) // Change this value to 2 to run the test slowly
val subscribeThreadsStillRunning = CountDownLatch(0) // Change this value to 1 to run the test slowly
val service: (String, String, String?, String, String, String, String, String, String, String?, String) -> Single<String> = { 
s1: String,
s2: Int,
s3: String?,
s4: Int,
s5: String,
s6: String,
s7: String,
s8: String,
s9: String,
s10: String?,
s11: String ->
val result =
listOf(s1, "$s2", s3 ?: "none", "$s4", s5, s6, s7, s8, s9, s10 ?: "none", s11).joinToString(separator = ";")
Single.just("Values:$result")
}
val createSingle = { value: String ->
Observable
.create<String> { emitter ->
println("Parallel subscribe $value on ${Thread.currentThread().name}")
bothSubscribed.countDown()
subscribeThreadsStillRunning.await(20, TimeUnit.SECONDS)
emitter.onNext(value)
emitter.onComplete()
}
.singleOrError()
.subscribeOn(io())
}
val s1: Single<String> = createSingle("v1")
val s2: Single<Int> = Single.just(2)
// Here, we move the Nullable value outside, so the whole Single<String> is Nullable, and not the value inside the Single`enter code here`
val s3: Single<String>? = null
val s4: Single<String> = Single.just(4)
val s5: Single<String> = createSingle("v5")
val s6: Single<String> = createSingle("v6")
val s7: Single<String> = createSingle("v7")
val s8: Single<String> = createSingle("v8")
val s9: Single<String> = createSingle("v9")
val s10: Single<String>? = null
val s11 = createSingle("v11")
//When
// Here I curry the function, so I can apply one by one the the arguments via zipOver() and preserve the types 
val singleFunction: Single<(String) -> (String) -> (String?) -> (String) -> (String) -> (String) -> (String) -> (String) -> (String) -> (String?) -> (String) -> Single<String>> =
Single.just(service.curried()).subscribeOn(io())
val result = singleFunction
.zipOver(s1)
.zipOver(s2)
.zipOverNullable(s3)
.zipOver(s4)
.zipOver(s5)
.zipOver(s6)
.zipOver(s7)
.zipOver(s8)
.zipOver(s9)
.zipOverNullable(s10)
.zipOver(s11)
.flatMap { it }
//Then
result
.test()
.awaitDone(50, TimeUnit.SECONDS)
.assertSubscribed()
.assertValues("Values:v1;2;none;4;v5;v6;v7;v8;v9;none;v11")

然后它会打印出这样的东西：

Parallel subscribe v11 on RxCachedThreadScheduler-10
Parallel subscribe v8 on RxCachedThreadScheduler-8
Parallel subscribe 4 on RxCachedThreadScheduler-4
Parallel subscribe v5 on RxCachedThreadScheduler-5
Parallel subscribe v9 on RxCachedThreadScheduler-9
Parallel subscribe 2 on RxCachedThreadScheduler-3
Parallel subscribe v6 on RxCachedThreadScheduler-6
Parallel subscribe v1 on RxCachedThreadScheduler-2
Parallel subscribe v7 on RxCachedThreadScheduler-7

现在，如果我这样做：

val result = singleFunction
.zipOver(s1)
.zipOver(s1)
.zipOverNullable(s3)
.zipOver(s1)
.zipOver(s5)
.zipOver(s6)
.zipOver(s7)
.zipOver(s8)
.zipOver(s9)
.zipOverNullable(s10)
.zipOver(s11)
.flatMap { it }

它将在编译时中断