<commit_before>package com.campudus.scycle

import rxscalajs.subscription.AnonymousSubscription
import rxscalajs.{Observable, Observer, _}

object Scycle {

  type DisposeFunction = () => Unit

  type StreamSubscribe[T] = (Observable[T], Observer[T]) => DisposeFunction

  trait ScycleSubject[T] {

    val stream: Observable[T]
    val observer: Observer[T]

  }

  type DriversDefinition = SourcesMap
  type Sources = SourcesMap
  type Sinks = SinksMap
  type SinkProxies = Map[Any, Subject[_]]

  def run(mainFn: Sources => Sinks, drivers: DriversDefinition): () => Unit = {

    if (drivers.inner.isEmpty) {
      throw new IllegalArgumentException("Scycle needs at least one driver to work.")
    } else {
      val sinkProxies = makeSinkProxies(drivers)
      val subscriptions = callDrivers(drivers, sinkProxies)
      val sinks = mainFn(drivers)
      val disposeReplication = replicateMany(sinks, sinkProxies)

      val result = () => {
        disposeSubscriptions(subscriptions)
        disposeReplication()
      }

      result
    }
  }

  def side[X](someNext: X => _): X => X = {
    x => {
      someNext(x)
      x
    }
  }

  private def replicateMany(sinks: Sinks, sinkProxies: SinkProxies): () => Unit = {

    type X = Any

    val disposeFunctions = sinks.inner.keys
      .filter(name => {
        sinkProxies.exists(a => {
          a._1 == name
        })
      })
      .map(name => {
        val subs = sinks.inner(name).subscribe(sinkProxies(name).asInstanceOf[Observer[X]])
        val latest$ = sinks.inner(name).lastOrElse(null)
        val dispose = subs.unsubscribe _
        latest$.map(x => {
          sinkProxies(name).asInstanceOf[Observer[X]].next(x)
        })
        dispose
      })

    () => disposeFunctions.foreach(_.apply())
  }

  private def logToConsoleError(error: Any): Unit = {
    println(error)
  }

  private def disposeSubscriptions(subscriptions: Map[Any, AnonymousSubscription]): Unit = {
    subscriptions.foreach(_._2.unsubscribe())
  }

  private def callDrivers(
    drivers: DriversDefinition,
    sinkProxies: SinkProxies
  ): Map[Any, AnonymousSubscription] = {

    type X = Nothing

    drivers.inner.foldLeft(Map[Any, AnonymousSubscription]()){
      case (m, (name, driver)) =>
        val proxyObservable = sinkProxies(name).asInstanceOf[Observable[X]]
        val subscription = driver.subscribe(proxyObservable)
        m + (name -> subscription)
    }
  }

  private def makeSinkProxies(drivers: DriversDefinition): SinkProxies = {
    drivers.inner.foldLeft(Map[Any, Subject[_]]()){
      case (m, (name, driver)) =>
        m + (name -> driver.createSubject())
    }
  }

  class SinkMapper[K, V <: Observable[_]]

  class SinksMap(val inner: Map[Any, Observable[_]] = Map.empty) {

    def get[K, V <: Observable[_]](k: K)(implicit ev: SinkMapper[K, V]): Option[V] = {
      inner.get(k).asInstanceOf[Option[V]]
    }

    def apply[K, V <: Observable[_]](k: K)(implicit ev: SinkMapper[K, V]): V = get(k).get

    def +[K, V <: Observable[_]](kv: (K, V))(implicit ev: SinkMapper[K, V]): SinksMap = new SinksMap(inner + kv)
    def -[K](k: K): SinksMap = new SinksMap(inner - k)
  }

  class SourcesMapper[K, V <: Driver[_]]

  class SourcesMap(val inner: Map[Any, Driver[_]] = Map.empty) {

    def get[K, V <: Driver[_]](k: K)(implicit ev: SourcesMapper[K, V]): Option[V] = inner.get(k).asInstanceOf[Option[V]]

    def apply[K, V <: Driver[_]](k: K)(implicit ev: SourcesMapper[K, V]): V = get(k).get

    def +[K, V <: Driver[_]](kv: (K, V))(implicit ev: SourcesMapper[K, V]): SourcesMap = new SourcesMap(inner + kv)
    def -[K](k: K): SourcesMap = new SourcesMap(inner - k)
  }

}<commit_msg>Make inner map in SinksMap/SourcesMap private<commit_after>package com.campudus.scycle

import rxscalajs.subscription.AnonymousSubscription
import rxscalajs.{Observable, Observer, _}

object Scycle {

  type DisposeFunction = () => Unit

  type StreamSubscribe[T] = (Observable[T], Observer[T]) => DisposeFunction

  trait ScycleSubject[T] {

    val stream: Observable[T]
    val observer: Observer[T]

  }

  type DriversDefinition = SourcesMap
  type Sources = SourcesMap
  type Sinks = SinksMap
  type SinkProxies = Map[Any, Subject[_]]

  def run(mainFn: Sources => Sinks, drivers: DriversDefinition): () => Unit = {

    if (drivers.inner.isEmpty) {
      throw new IllegalArgumentException("Scycle needs at least one driver to work.")
    } else {
      val sinkProxies = makeSinkProxies(drivers)
      val subscriptions = callDrivers(drivers, sinkProxies)
      val sinks = mainFn(drivers)
      val disposeReplication = replicateMany(sinks, sinkProxies)

      val result = () => {
        disposeSubscriptions(subscriptions)
        disposeReplication()
      }

      result
    }
  }

  def side[X](someNext: X => _): X => X = {
    x => {
      someNext(x)
      x
    }
  }

  private def replicateMany(sinks: Sinks, sinkProxies: SinkProxies): () => Unit = {

    type X = Any

    val disposeFunctions = sinks.inner.keys
      .filter(name => {
        sinkProxies.exists(a => {
          a._1 == name
        })
      })
      .map(name => {
        val subs = sinks.inner(name).subscribe(sinkProxies(name).asInstanceOf[Observer[X]])
        val latest$ = sinks.inner(name).lastOrElse(null)
        val dispose = subs.unsubscribe _
        latest$.map(x => {
          sinkProxies(name).asInstanceOf[Observer[X]].next(x)
        })
        dispose
      })

    () => disposeFunctions.foreach(_.apply())
  }

  private def logToConsoleError(error: Any): Unit = {
    println(error)
  }

  private def disposeSubscriptions(subscriptions: Map[Any, AnonymousSubscription]): Unit = {
    subscriptions.foreach(_._2.unsubscribe())
  }

  private def callDrivers(
    drivers: DriversDefinition,
    sinkProxies: SinkProxies
  ): Map[Any, AnonymousSubscription] = {

    type X = Nothing

    drivers.inner.foldLeft(Map[Any, AnonymousSubscription]()){
      case (m, (name, driver)) =>
        val proxyObservable = sinkProxies(name).asInstanceOf[Observable[X]]
        val subscription = driver.subscribe(proxyObservable)
        m + (name -> subscription)
    }
  }

  private def makeSinkProxies(drivers: DriversDefinition): SinkProxies = {
    drivers.inner.foldLeft(Map[Any, Subject[_]]()){
      case (m, (name, driver)) =>
        m + (name -> driver.createSubject())
    }
  }

  class SinkMapper[K, V <: Observable[_]]

  class SinksMap(private[Scycle] val inner: Map[Any, Observable[_]] = Map.empty) {

    def get[K, V <: Observable[_]](k: K)(implicit ev: SinkMapper[K, V]): Option[V] = {
      inner.get(k).asInstanceOf[Option[V]]
    }

    def apply[K, V <: Observable[_]](k: K)(implicit ev: SinkMapper[K, V]): V = get(k).get

    def +[K, V <: Observable[_]](kv: (K, V))(implicit ev: SinkMapper[K, V]): SinksMap = new SinksMap(inner + kv)
    def -[K](k: K): SinksMap = new SinksMap(inner - k)
  }

  class SourcesMapper[K, V <: Driver[_]]

  class SourcesMap(private[Scycle] val inner: Map[Any, Driver[_]] = Map.empty) {

    def get[K, V <: Driver[_]](k: K)(implicit ev: SourcesMapper[K, V]): Option[V] = inner.get(k).asInstanceOf[Option[V]]

    def apply[K, V <: Driver[_]](k: K)(implicit ev: SourcesMapper[K, V]): V = get(k).get

    def +[K, V <: Driver[_]](kv: (K, V))(implicit ev: SourcesMapper[K, V]): SourcesMap = new SourcesMap(inner + kv)
    def -[K](k: K): SourcesMap = new SourcesMap(inner - k)
  }

}
{'fraction_non_alphanumeric': 0.14518738204367754, 'fraction_numerical': 0.0005392289026691831, 'mean_word_length': 3.212947189097104, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 39, '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': '4551123', 'n_tokens_mistral': 2609, 'n_tokens_neox': 2469, 'n_words': 547}