具有状态的功能侦听器

Question

假设在我的纯Scala程序中，我依赖于Java服务。 此 Java 服务接受侦听器，以便在某些数据更改时通知我。

假设数据是一个元组(x，y(，java 服务在 X 或 Y 更改时调用侦听器，但我只对 X 感兴趣。

为此，我的侦听器必须保存 X 的最后一个值，并且仅在 oldX ！= X 时才转发更新/调用，因此为了拥有这个，我不纯的 scala 侦听器实现必须保存一个 var oldX

val listener = new JavaServiceListener() {
var oldX;
def updated(val x, val y): Unit = {
if (oldX != x) {
oldX = x
//do stuff
}
}
javaService.register(listener)

我将如何在没有 val 或可变集合的情况下为 Scala 中的这种东西设计一个包装器？我不能在JavaServiceListener级别，因为我受方法签名的约束，所以我需要另一层，java侦听器以某种方式转发到上面

Answer 1

我的偏好是将其包装在Monix Observable中，然后您可以使用distinctUntilChanged来消除连续的重复项。 像这样：

import monix.reactive._
val observable = Observable.create(OverflowStrategy.Fail(10)){(sync) =>
val listener = new JavaServiceListener() {
def updated(val x, val y): Unit = {
sync.onNext(x)
}
}
javaService.register(listener)
Cancelable{() => javaService.unregister(listener)}
}
val distinctObservable = observable.distinctUntilChanged

响应式编程允许您使用纯模型，而库处理所有困难的内容。

Answer 2

首先，如果你正在设计一个纯函数程序，你不能返回Unit(两者都不是Future[Unit]，因为Future不会抑制副作用(。

如果性能不是问题，我会在T = Option的地方使用Kleisli[Option, xType, IO[Unit]]。所以你要做的第一件事是定义(添加适当的类型(

def updated(oldX, x): Kleisli[Option, xType, xType] = Kleisli liftF {
if(x != oldX) None
else Some(x)
}
def doStuff(x, y): Kleisli[Option, xType, IO[Unit]] = Kleisli pure {
IO{
//doStuff
}
}

现在你可以把它们写成这样的东西来理解：

val result: Kleisli[Option, xType, IO[Unit]] = for{
xx <- updated(oldX, x)
effect <- doStuff(xx, y)
} yield effect

您可以使用ReaderWriterStateT执行有状态计算，因此oldX保持状态。

Answer 3

我找到了我喜欢的猫和猫效应的解决方案：

trait MyListener {
def onChange(n: Int): Unit
}
class MyDistinctFunctionalListener(private val last: Ref[IO, Int], consumer: Int => Unit) extends MyListener {
override def onChange(newValue: Int): Unit = {
val program =
last
.getAndSet(newValue)
.flatMap(oldValue => notify(newValue, oldValue))
program.unsafeRunSync()
}
private def notify(newValue: Int, oldValue: Int): IO[Unit] = {
if (oldValue != newValue) IO(consumer(newValue)) else IO.delay(println("found duplicate"))
}
}
object MyDistinctFunctionalListener {
def create(consumer: Int => Unit): IO[MyDistinctFunctionalListener] =
Ref[IO].of(0).map(v => new MyDistinctFunctionalListener(v, consumer))
}
val printer: Int => Unit = println(_)
val functionalDistinctPrinterIO =  MyDistinctFunctionalListener.create(printer)
functionalDistinctPrinterIO.map(fl =>
List(1, 1, 2, 2, 3, 3, 3, 4, 5, 5).foreach(fl.onChange)
).unsafeRunSync()

有关处理共享状态的更多内容，请参阅此处 https://github.com/systemfw/scala-italy-2018

这是否值得私有VaR解决方案是值得的