在 RxScala 应用程序中调用 Java 库 (rxjava-extras) 代码

Question

我正在 RxScala 应用程序中使用来自 Kafka 的消息，并希望以动态大小的滑动 window 处理消息，即我想将在同一 2 秒内创建的所有消息分组。据我所知，RxScala 不提供这种运算符。这个存储库 (https://github.com/davidmoten/rxjava-extras) 确实提供了一个函数“toListWhile”，但是以 Java 代码的形式。

调用所需函数的简化尝试失败：

import com.github.davidmoten.rx.Transformers
import rx.functions.Func1
import rx.lang.scala.JavaConversions._
import rx.lang.scala.Observable

val o = Observable.from(List(1, 2, 3, 4, 5, 6))

val predicate = new Func1[java.lang.Integer, java.lang.Boolean] {
  override def call(t: Integer): Boolean = true
}

val p = o.compose(Transformers.toListWhile(predicate))

这是我收到的错误：

Error:(75, 45) type mismatch;
 found   : rx.functions.Func1[Integer,Boolean]
 required: rx.functions.Func1[_ >: _, Boolean] where type _ <: Int
Note: Integer <: Any, but Java-defined trait Func1 is invariant in type T.
You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
val p = o.compose(Transformers.toListWhile(predicate))
                                           ^

有人可以帮我从 Scala 调用该代码吗？谢谢！

解决方法

import java.lang.Boolean

import com.github.davidmoten.rx.Transformers
import rx.functions.Func1
import rx.lang.scala.JavaConversions._
import rx.lang.scala.Observable

val o = Observable.from(List[java.lang.Integer](1, 2, 3, 4, 5, 6))

val predicate = new Func1[java.lang.Integer, java.lang.Boolean] {
  override def call(t: Integer): Boolean = true
}

val p = o.compose[java.util.List[java.lang.Integer]](Transformers.toListWhile(predicate))

Answer 1

java.lang.Integer 和 Int 是不同的类型，尽管它们可以相互转换。使用任一

val o = Observable.from(List[java.lang.Integer](1, 2, 3, 4, 5, 6))

或

val predicate = new Func1[Int, java.lang.Boolean] ...

第一个肯定有效，第二个可能无效（使用 Java 定义的泛型类型参数扩展 AnyVal 存在问题，例如 Int）。

在 RxScala 应用程序中调用 Java 库 (rxjava-extras) 代码

Call Java library (rxjava-extras) code in RxScala application

java

scala

scala-java-interop

rx-java

rx-scala