0

我在 RxScala 应用程序中使用来自 Kafka 的消息，并希望在动态大小的滑动窗口中处理消息，即我想对在同一 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[_ >: _$7, Boolean] where type _$7 <: 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))
4

1 回答 1

0

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）。

于 2017-03-27T20:32:47.773 回答