Как элегантно реализовать шаблон конвейера с помощью Scala

Я собираюсь построить шаблон конвейера с помощью 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 с Функция1. В Scala класс Function0 не имеет метода compose() или andThen().