如何使用 Scala 优雅地实现管道模式

Question

我正在寻找使用 Scala 构建管道模式。我希望在我写完管道对象之后，它们可以像这样连接在一起：

Pipeline1 :: Pipeline2 :: Pipeline3 ...

到目前为止，我已经尝试了一些想法。有些工作，有些不工作。但它们似乎都没有完全摆脱样板代码。以下是我得到的最接近的。

首先定义 Pipeline 和 Source 抽象类：

// I is the input type and O is the output type of the pipeline
abstract class Pipeline[I, +O](p: Pipeline[_, _ <: I]) {

  val source = p
  val name: String
  def produce(): O
  def stats():String
}
abstract class Source[+T] extends Pipeline[AnyRef, T](null)

接下来，我创建了两个管道并尝试将它们链接在一起

// this creates a random integer
class RandomInteger extends Source[Int] {
  override val name = "randInt"

  def produce() = {
    scala.Math.round(scala.Math.random.asInstanceOf[Float] * 10)
  }

  def stats()="this pipeline is stateless"
}

// multiply it by ten
class TimesTen(p: Pipeline[_, Int]) extends Pipeline[Int, Int](p) {
  private var count = 0 // this is a simple state of the pipeline
  override val name = "Times"
  def produce = {
    val i = source.produce()
    count += 1 // updating the state
    i * 10
  }
  def stats() = "this pipeline has been called for " + count + " times"
}

object TimesTen {
  // this code achieves the desired connection using ::
  // but this has to be repeated in each pipeline subclass. 
  // how to remove or abstract away this boilerplate code? 
  def ::(that: Pipeline[_, Int]) = new TimesTen(that)
}

这是连接两个管道的主类。

object Pipeline {
  def main(args: Array[String]) {
    val p = new RandomInteger() :: TimesTen
    println(p.source)
    for (i <- 0 to 10)
      println(p.produce())
    println(p.stats())
  }
}

所以这段代码有效。但是我必须在我编写的每个管道类中重复 TimesTen 伴随对象中的代码。这当然是不可取的。有没有更好的方法来做到这一点？反射可能会起作用，但我听说过一些不好的事情，比如任何涉及反射的东西都是糟糕的设计。我也不确定 Scala 是否支持反射。

感谢您的宝贵时间。

更新：我设计了这个玩具问题，使其易于理解。作为一个通用解决方案，并且正如我的应用程序所需要的那样，每个管道对象都有一个状态，该状态理想地封装在对象本身中，而不是暴露给所有其他管道。我已经修改了上面的代码以反映这一点。我希望有一个基于对象的解决方案。我还在试验中，如果找到了会通知你的。

更新 2：经过一番思考，我认为管道的想法实际上只是一个包含一些内部状态的通用函数以及能够与 @ 组成 Function0 函数的能力987654326@ 功能。在 Scala 中，Function0 类没有compose() 或andThen() 方法。

Answer 1

这是使用andThen 的对象的解决方案。这个想法是通过使用输入Unit 来强制创建Function1 对象。连接两个 Pipelines 会创建一个包含两个函数的新 Pipeline。此解决方案允许管道具有内部状态。

进一步的简化是使用apply() 而不是produce()。这留给读者作为练习。

abstract class Pipeline[-I, +O] {

  val name: String
  def produce : I => O
  def stats(): String

  def ->[X](seg:Pipeline[_ >: O, X]):Pipeline[I, X] = {
    val func = this.produce
    val outerName = this.name
    new Pipeline[I, X] {
      val name = outerName + "." + seg.name
      def produce = func andThen seg.produce 
      def stats = seg.stats
    }
  }
}

abstract class Source[+T] extends Pipeline[Unit, T] {
}

class RandomInteger extends Source[Int] {
  override val name = "randInt"
  def produce: Unit => Int = (x:Unit) => scala.Math.round(scala.Math.random.asInstanceOf[Float] * 10) 
  def stats() = "stateless"
}

class TimesTen() extends Pipeline[Int, Int] {
  private var count = 0
  override val name = "times"
  def produce : Int => Int = (x:Int) => {    
    count += 1
    x * 10
  }
  def stats() = "called for " + count + " times"
}


object Main {
  def main(args: Array[String]) {
    val p = new RandomInteger() -> new TimesTen() 

    for (i <- 0 to 10)
      println(p.produce())
    println(p.name)    // print "randInt.times"
    println(p.stats()) // print "called for 11 times"
  }
}

Answer 2

除非我遗漏了什么，否则您的管道对象只是函数，而您的 :: 运算符只是“组合”

val randomInteger: ()=>Int = () => scala.Math.round(scala.Math.random.asInstanceOf[Float] * 10)
val timesTen :Int => Int = x => x*10    
val pipeline: () =>Int = timesTen compose randomInteger

您的“produce()”方法只是“apply()”，但通常使用它是“()”的缩写。少量的图书馆拉皮条将允许您使用运算符进行组合。这是面向对象样板真正妨碍简单功能概念的情况之一。幸运的是，Scala 可以让您避免像这样的许多用例的样板。

Answer 3

object Pipelining {  implicit def toPipe[T](x : T) = new {     def :: [U](f : T => U) = f(x)  }}

import Pipelining._
List(2,3,4) :: (_.map(_*3)) :: (_.map(_.toString)) :: println 

StephaneLD "|> operator like in F#"

http://www.scala-lang.org/node/8747

Answer 4

您的意思是像数据流或函数响应式编程？试试this question。反应式库正在积极开发中——我不知道其余的。