Implementation of my C# Task (pipe and filters) in Scala

Task.scala

import scala.collection.mutable.ListBuffer

object HelloWorld {

	def Start() {
		println("OnStart")
	}

	def Complete() {
		println("OnComplete")
	}

	def Success() {
		println("OnSuccess")
	}

	def Error(error: String) {
		println("OnError: " + error)
	}

	def main(args: Array[String]){

		var task = new Task[OrderMessage]()
		task.wrap(new LoggingAspect[OrderMessage]())
		task.wrap(new ExceptionLoggingAspect[OrderMessage]())
		task.register(new Step1())
		task.register(new Step2())
		task.register(new Step3())

		var msg1 = new OrderMessage(Start, Complete, Success, Error) { PretendErrorHappened = true }		
		task.invoke(msg1)
		println("msg1 result: " + msg1.Result + "\n")

		var msg2 = new OrderMessage(Start, Complete, Success, Error) { PretendErrorHappened = false }
		task.invoke(msg2)		
		println("msg2 result: " + msg2.Result)
	}
}

// looks like function declarations in Scala are immutable so the only opportunity the caller has to assign them is via the constructor
class OrderMessage(start: () => Unit, complete: () => Unit, success: () => Unit, err: (String) => Unit) extends BaseMessage {
	def OnStart = start()
	def OnComplete = complete()
	def OnSuccess = success()
	def OnError(error: String) = err(error)

	var PretendErrorHappened: Boolean = false
	var Result: Int = 0
}

class LoggingAspect[T] extends Handler[T] {

	def Handle(msg: T){
		println("LoggingAspect before")
		Inner.Handle(msg)
		println("LoggingAspect after")
	}

	var Inner: Handler[T] = _
}

class ExceptionLoggingAspect[T] extends Handler[T] {

	def Handle(msg: T){
		try {
		  Inner.Handle(msg)
		} catch {
		  case e: Exception => println("ExceptionLoggingAspect: " + e);
		}
	}

	var Inner: Handler[T] = _
}

class Step1 extends Filter[OrderMessage] {
	def Execute(msg: OrderMessage){
		println("step 1")
	}
}

class Step2 extends Filter[OrderMessage] {
	def Execute(msg: OrderMessage){
		if(msg.PretendErrorHappened) {
			msg.OnError("bad robot!")
			msg.Stop = true;
			return
			
			// throw new Exception("oh oh! something bad happened!")
		}
		println("step 2")
	}
}

class Step3 extends Filter[OrderMessage] {
	def Execute(msg: OrderMessage){
		msg.Result = 3
		println("step 3")
	}
}


/* pipeline api */

trait Filter[T] {
	def Execute(msg: T)
}

trait Handler[T] {
	def Handle(msg: T)
	var Inner: Handler[T]
}

trait BaseMessage {
	var Stop: Boolean = false
	def OnStart()
	def OnComplete()
	def OnSuccess()
	def OnError(error: String)
}

class Task[T <: BaseMessage] {
	
	val pipeline: Pipeline[T] = new Pipeline[T]();
	var handler: Handler[T] = pipeline;

	def wrap(h: Handler[T]){
		h.Inner = this.handler
		this.handler = h
	}

	def register(filter : Filter[T]){
		pipeline.add(filter);
	}

	def invoke(msg: T){
		this.handler.Handle(msg)		
	}

	class Pipeline[T1 <: BaseMessage] extends Handler[T1] {

		def add(filter: Filter[T1]){
			filters.append(filter)
		}

		def Handle(msg: T1) {
			msg.OnStart()
			try {
				filters.foreach(f => { if(msg.Stop) return; f.Execute(msg) })
				msg.OnSuccess()
			} catch {
			  case e: Exception => {
			  	msg.OnError(e.getMessage())
			  	throw e;
			  }
			} finally {
				msg.OnComplete()
			}
		}
		
		var Inner: Handler[T1] = _

		private val filters = new ListBuffer[Filter[T1]]()	
	}	
}