<commit_before>package colossus.core

import scala.concurrent.duration._
import akka.actor.ActorRef

/**
 * This trait must be implemented by any non-head member of a pipeline.  
 */
trait HasUpstream[T] {
  private var _upstream: Option[T] = None
  def setUpstream(up: T) {
    _upstream = Some(up)
  }
  def upstream = _upstream.getOrElse(throw new Exception("Attempt to use uninitialized upstream reference"))
}

/**
 * This must be implemented by any non-tail member of a pipeline
 */
trait HasDownstream[T] {
  def downstream: T
}

/**
 * These are events that propagate to each layer starting from the head and moving downstream
 */
trait DownstreamEvents extends WorkerItemEvents {
  
  def connected() { onConnected() }
  def connectionTerminated(reason: DisconnectCause) { onConnectionTerminated(reason) }
  def idleCheck(period: FiniteDuration) { onIdleCheck(period) }
  def bind() { onBind() }
  def unbind() { onUnbind() }
  override def receivedMessage(message: Any, sender: ActorRef) { onReceivedMessage(sender, message) }

  protected def onConnected() {}
  protected def onConnectionTerminated(reason: DisconnectCause) {}
  protected def onIdleCheck(period: FiniteDuration){}<commit_msg>make upstream a lazy val, slightly improves performance<commit_after>package colossus.core

import scala.concurrent.duration._
import akka.actor.ActorRef

/**
 * This trait must be implemented by any non-head member of a pipeline.  
 */
trait HasUpstream[T] {
  private var _upstream: Option[T] = None
  def setUpstream(up: T) {
    _upstream = Some(up)
  }
  lazy val upstream = _upstream.getOrElse(throw new Exception("Attempt to use uninitialized upstream reference"))
}

/**
 * This must be implemented by any non-tail member of a pipeline
 */
trait HasDownstream[T] {
  def downstream: T
}

/**
 * These are events that propagate to each layer starting from the head and moving downstream
 */
trait DownstreamEvents extends WorkerItemEvents {
  
  def connected() { onConnected() }
  def connectionTerminated(reason: DisconnectCause) { onConnectionTerminated(reason) }
  def idleCheck(period: FiniteDuration) { onIdleCheck(period) }
  def bind() { onBind() }
  def unbind() { onUnbind() }
  override def receivedMessage(message: Any, sender: ActorRef) { onReceivedMessage(sender, message) }

  protected def onConnected() {}
  protected def onConnectionTerminated(reason: DisconnectCause) {}
  protected def onIdleCheck(period: FiniteDuration){}
{'fraction_non_alphanumeric': 0.0949857317570322, 'fraction_numerical': 0.0, 'mean_word_length': 4.801418439716312, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5521622', 'n_tokens_mistral': 726, 'n_tokens_neox': 709, 'n_words': 229}