<commit_before>import rx.lang.scala.Observable
import rx.lang.scala.Subscription

import concurrent.duration._
import scala.collection.mutable
import scala.concurrent.Future

/* Example source of transform:
* async {
*   val result = join {
*     case obs1(x) & obs2(y) => x + y
*     case obs3(x) => x + 1
*   }
*   println(result)
* }
* */
object Transform extends App {

  val obs1 = Observable.interval(200 millis)
  val obs2 = Observable.interval(400 millis)
  val obs3 = Observable.interval(100 millis)

  val observables = List(obs1, obs2, obs3)
  val subscriptions = observables.map { obs =>
    obs.subscribe(msg => resolveMessage(obs, msg))
  }

  val channels = new mutable.AnyRefMap[Observable[Any], mutable.Queue[Any]]()

  var ids = new mutable.AnyRefMap[Observable[Any], Int]()
  observables.zipWithIndex.foreach {
    case (obs, index) => ids.put(obs, 1 << index)
  }

  def getId[A](obs: Observable[A]): Int = ids.getOrElse(obs, throw Exception)

  class BitField(val value: Int) {
    def set(mask: Int) = new BitField(value | mask)
    def clear(mask: Int) = new BitField(value & ~mask)
    def matches(mask: Int): Boolean = (~value & mask) == 0
    def isSet(mask: Int): Boolean = (value & mask) == mask
  }
  object BitField {
    def empty = new BitField(0)
  }

  class Pattern[A, B](val target: Int, val body: (A => B)) extends BitField(target)
  object Pattern {
    def apply[A, B](target: Int, body: (A => B)) = new Pattern(target, body)
  }

  val patterns = List(
    Pattern(getId(obs1) | getId(obs2), { case (x, y) => x + y }),
    Pattern(getId(obs3), { case (x) => x })
  )

  val continuation: (Any => Unit) = ???

  var state = BitField.empty

  def resolveMessage[A](sender: Observable[A], msg: A) = {
    patterns.synchronized {
      val senderId = getId(sender)
      state.set(senderId)
      patterns.find(p => p.matches(state.value)) match {
        case Some(p) => {
          state.clear(p.value)
          // How to map the arguments to the body?
          val queues = channels.map({ case (obs, _) => getId(obs) })
                               .filter({ case (obs, _) => p.isSet(obs) })
                               .toList
          Future {
              val result = p.body(???)
              continuation(result)
          }
        }
        case None => {
          channels.get(sender) match {
            case Some(c) => c += msg
            case None => channels += (sender -> mutable.Queue(msg))
          }
        }
      }
    }
  }
}<commit_msg>Adds a more exapnend version of the transform<commit_after>import rx.lang.scala.Observable
import rx.lang.scala.Subscription

import concurrent.duration._
import scala.collection.immutable.Queue
import scala.concurrent.Future

/* Example source of transform:
* async {
*   val result = join {
*     case obs1(x) & obs2(y) => x + y
*     case obs3(x) => x + 1
*   }
*   println(result)
* }
* */

class BitField(val value: Int) {
  def set(mask: Int) = new BitField(value | mask)
  def clear(mask: Int) = new BitField(value & ~mask)
  def matches(mask: Int): Boolean = (~value & mask) == 0
  def isSet(mask: Int): Boolean = (value & mask) == mask
}
object BitField {
  def empty = new BitField(0)
}

object Transform extends App {

  val obs1 = Observable.interval(200 millis)
  val obs2 = Observable.interval(400 millis)
  val obs3 = Observable.interval(100 millis)

  val obs1_id = 1 << 0
  val obs2_id = 1 << 1
  val obs3_id = 1 << 2

  // TODO: The Any type should correspond to the Observable type
  var channels = Map[Int, Queue[Any]]()

  val pattern1 = obs1_id | obs2_id
  val pattern2 = obs3_id

  val pattern1_continuation = (v1: Long, v2: Long) => v1 + v2
  val pattern2_continuation = (v1: Long) => v1

  val continuation: (Any => Unit) = ???

  var state = BitField.empty

  obs1.subscribe(
    next => {
      state.synchronized {
        if (state.set(obs1_id).matches(pattern1)) {
          val obs2_value = channels.get(obs2_id) match {
            case Some(x: Long) => x
          }
          Future {
            val result = pattern1_continuation(next, obs2_value)
            continuation(result)
          }
        } else {
          channels.get(obs1_id) match {
            case Some(q) => q.enqueue(next)
            case None => channels += (obs1_id -> Queue(next))
          }
          state.set(obs1_id)
        }
      }
    }
  )

  obs2.subscribe(
    next => {
      state.synchronized {
        if (state.set(obs2_id).matches(pattern1)) {
          val obs1_value = channels.get(obs1_id) match {
            case Some(x: Long) => x
          }
          Future {
            val result = pattern1_continuation(obs1_value, next)
            continuation(result)
          }
        } else {
          channels.get(obs2_id) match {
            case Some(q) => q.enqueue(next)
            case None => channels += (obs2_id -> Queue(next))
          }
          state.set(obs2_id)
        }
      }
    }
  )

  obs3.subscribe(
    next => {
      state.synchronized {
        if (state.set(obs3_id).matches(pattern2)) {
          Future {
            val result = pattern2_continuation(next)
            continuation(result)
          }
        }
      }
    }
  )


}
{'fraction_non_alphanumeric': 0.12980030721966207, 'fraction_numerical': 0.01708909370199693, 'mean_word_length': 2.247506234413965, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 37, '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': '5377245', 'n_tokens_mistral': 1752, 'n_tokens_neox': 1700, 'n_words': 425}