Cómo implementar elegantemente el patrón de tubería usando Scala

Question

Estoy buscando construir un patrón de tubería con Scala. Deseo después de escribir los objetos del canal, que podrían estar conectados entre sí de esta manera:

Pipeline1 :: Pipeline2 :: Pipeline3 ... 

he experimentado con algunas ideas hasta el momento. Algunos trabajan y otros no. Pero ninguno de ellos parece deshacerse por completo del código repetitivo. Lo siguiente es lo más cercano que tengo.

En primer lugar definir la clase abstracta de la tubería y Fuente:

// 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) 

A continuación, he creado dos oleoductos y tratar de vincularlos

// 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) 
} 

Ésta es la clase principal, donde dos tuberías están vinculados.

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()) 
    } 
} 

Este código funciona. Pero tendría que repetir el código en el objeto complementario TimesTen en cada clase de interconexión que escribo. Esto ciertamente no es deseable. ¿Hay alguna forma mejor de hacer esto? La reflexión podría funcionar, pero escuché cosas malas al respecto, como que todo lo que involucre la reflexión es un mal diseño. Tampoco estoy seguro del apoyo de Scala para la reflexión.

Gracias por su tiempo.

Actualización: Diseñé este problema de juguete para que sea fácil de entender. Como solución general, y según lo requiera mi aplicación, cada objeto de canalización tiene un estado, que está idealmente encapsulado dentro del propio objeto en lugar de estar expuesto a cualquier otra canalización. He modificado el código anterior para reflejar esto. Ojalá pudiera haber una solución basada en objetos. Todavía estoy experimentando y te dejaré saber si encuentro uno.

Actualización 2: Después de algunos pensamientos, creo que la idea de la tubería es en realidad una función generalizada que contiene algunos estados internos, así como la capacidad de componer una función Function0 con una función Function1. En Scala, la clase Function0 no tiene el método compose() o andThen().

Answer 1

Aquí está la solución con los objetos que usan andThen. La idea es forzar la creación de objetos Function1 utilizando la entrada Unit. La conexión de dos tuberías crea una nueva tubería con las dos funciones juntas. Esta solución permite que las tuberías tengan estados internos.

Una simplificación adicional sería usar apply() en lugar de produce(). Esto se deja como un ejercicio para el lector.

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

A menos que me falta algo, sus objetos de tuberías son sólo funciones, y su operador :: es simplemente "componer"

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 

Tu es simplemente "Produce) (" método "se aplica()", pero es común usar su abreviatura de "()". Una pequeña cantidad de proxenetismo en la biblioteca le permitiría usar un operador para la composición. Este es uno de esos casos donde la repetición orientada a objetos realmente se interpone en el camino de conceptos funcionales simples. Afortunadamente, Scala te permite evitar la repetición de muchos casos de uso como este.

Answer 3

¿Quiere decir como un flujo de datos o programación funcional reactiva? Pruebe this question. La biblioteca reactiva se desarrolla activamente, no sé sobre el resto.

Answer 4

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 

todos los créditos para StephaneLD "|> operador como en F #"

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