How can I generalize the arity of rxjava2 Zip function (from Single/Observable) to n Nullable arguments without lose its types?

Question

Two Main Problems to solve:

1) Type check is lost

Using the array argument Single.zip() version I lose the strongly typed arguments.

2) Source argument Cannot be Nullable

I cannot send nullable source values as argument of Single.zip() function

3) I want an alternative to the method taking an Object[] not typed:

4) I don't want Mutable Objects, I don't want the use of var in my class, i want to use val

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

In haskell, there is a question related How can I implement generalized "zipn" and "unzipn" in Haskell?:

And in haskell I can achieve this with applicative functors:

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

being f :: Int -> Int -> Int -> Int -> Int -> Int -> Int -> String -> String -> String -> Int

and a1 .. a11 values corresponding each type

There is a list of similar functions in the library:

• With two arguments:

   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);
   }
  

• with three:

    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)
  

And so on...

In all those cases, is just fine, because each argument is typed. But there is a limitation until 9 Single sources

In our project, we needed more sources, because we have a lot of services that we want to reach async (in our case was 11 arguments).

But the issue is the arguments lose their strong types, and worse, some of them could be Nullable

For example we wanted to solve this use case:

//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")

As you can see, problems may occur if I do for example:

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

Fails because:

1) None of the Single.zip() functions can take a nullable value as argument.

2) You can change in the array the order of the values and it can fail because of type-check casting

Answer 1

A function with eleven parameters is a good example for unclean code. Instead you should consider to build a model to serve your needs. Like this you can provide meaningful names for each argument as well.

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,
        ...
    ) 
}

Having this model you could just use the zipWith() operator on each of your sources.

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

If you consider to abstract the nullability as a Maybe, you could simply define an extension function receiving a Maybe with data or without data and map it appropriately.

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)