A basic Scala implementation of the messaging pipeline as described at http://eventuallyconsistent.net/2013/08/12/messaging-as-a-programming-model-part-1/

PipeAndFilters.scala

import scala.collection.mutable.ListBuffer
 
val msg1 = new Network1(1);
val msg2 = new Network1(2);
val msg3 = new Network1(3);
 
val pipeline = new PipeLine[Network1]();
pipeline.register((x: Network1) => new GetUndeliveredOrders(x))
.register((x: Network1) => new GetDeliveryInfo(x))
.register((x: Network1) => new GetPodImage(x))
 
pipeline.executeWithRetries(msg1, 3);
pipeline.executeWithRetries(msg2, 3);
pipeline.executeWithRetries(msg3, 3);
 
 
// message
class Network1(anId:Int){
var id: Int = anId
}
 
// pipeline
class PipeLine[T]
{
def register(action:(T) => {}) : PipeLine[T] = {
list.append(action)
return this
}
 
def execute(input:T){
list.foreach(action => action(input))
}
 
def executeWithRetries(input:T, retries:Int){
try {
this.execute(input)
} catch {
case e: Exception => {
if(retries > 0){
Thread.sleep(5000)
executeWithRetries(input, retries-1)
}
}
}
}
 
private val list = new ListBuffer[(T) => {}]()
}
 
 
// filters
 
class GetUndeliveredOrders(input:Network1){
println("GetUndeliveredOrders " + input.id)
 
if(input.id == 2)
throw new Exception("oops!")
}
 
class GetDeliveryInfo(input:Network1){
println("GetDeliveryInfo " + input.id)
}
 
class GetPodImage(input:Network1){
println("GetPodImage " + input.id)
}