如何在不丢失其类型的情况下将 rxjava2 Zip 函数（从 Single/Observable）的数量概括为 n 个 Nullable 参数？

Question

要解决的两个主要问题：

1) 类型检查丢失

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

2) 源参数不能为空

我不能发送可为空的源值作为函数的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中，有一个相关的问题如何在Haskell中实现通用的“zipn”和“unzipn”？：

在 haskell 中，我可以使用 applicative functors 来实现这一点：

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

存在 f :: Int -> Int -> Int -> Int -> Int -> Int -> Int -> String -> String -> String -> Int

和a1 .. a11对应每种类型的值

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

• 有两个论点：

   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()函数可以将可空值作为参数。

2）您可以在数组中更改值的顺序，并且由于类型检查转换可能会失败

Answer 1

我已经使用以下方法实现了该目标：

1. Kotlin 扩展函数
2. 柯里化函数（Kotlin 允许）
3. 部分应用（Kotlin 也允许）
4. 函子和应用函子概念（单一类和可观察类是应用函子）
5. 将它们混合在一起：

首先，zipOver对于不可空值的函数：

/**
 * 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对于可为 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 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.funktionale.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 }

它会在编译时中断