<commit_before>package rescala.extra.distributables

import loci.registry.{Binding, Registry}
import loci.transmitter.RemoteRef
import rescala.core.{InitialChange, Scheduler, Struct}
import rescala.extra.lattices.Lattice
import rescala.reactives.{Observe, Signal}
import viscel.shared.Log

import scala.concurrent.Future

object LociDist {

  def distribute[A: Lattice, S <: Struct : Scheduler]
  (signal: Signal[A, S],
   registry: Registry)
  (binding: Binding[A => Unit] {type RemoteCall = A => Future[Unit]}) =
    distributePerRemote(_ => signal, registry)(binding)

  def distributePerRemote[A: Lattice, S <: Struct : Scheduler]
  (signalFun: RemoteRef => Signal[A, S],
   registry: Registry)
  (binding: Binding[A => Unit] {type RemoteCall = A => Future[Unit]})
  : Unit = {

    Log.Server.info(s"starting new distribution for »${binding.name}«")

    registry.bindPerRemote(binding) { remoteRef =>
      val signal: Signal[A, S] = signalFun(remoteRef)
      val signalName           = signal.name.str
      // println(s"binding »$signalName«(${signal.hashCode()}) for »$remoteRef«")
      newValue => {
        //println(s"received value for $signalName: ${newValue.hashCode()}")
        Scheduler[S].forceNewTransaction(signal) { admissionTicket =>
          admissionTicket.recordChange(new InitialChange[S] {
            override val source = signal
            override def writeValue(b: source.Value, v: source.Value => Unit): Boolean = {
              val merged = b.map(Lattice[A].merge(_, newValue)).asInstanceOf[source.Value]
              if (merged != b) {
                v(merged)
                true
              }
              else false
            }
          })
        }
        Log.Server.info(s"update for $signalName complete")
      }
    }

    var observers = Map[RemoteRef, Observe[S]]()

    def registerRemote(remoteRef: RemoteRef): Unit = {
      val signal: Signal[A, S] = signalFun(remoteRef)<commit_msg>make execution a bit more clear<commit_after>package rescala.extra.distributables

import loci.registry.{Binding, Registry}
import loci.transmitter.RemoteRef
import rescala.core.{InitialChange, Scheduler, Struct}
import rescala.extra.lattices.Lattice
import rescala.reactives.{Observe, Signal}
import viscel.shared.Log

import scala.concurrent.Future

object LociDist {

  def distribute[A: Lattice, S <: Struct : Scheduler]
  (signal: Signal[A, S],
   registry: Registry)
  (binding: Binding[A => Unit] {type RemoteCall = A => Future[Unit]}) =
    distributePerRemote(_ => signal, registry)(binding)

  def distributePerRemote[A: Lattice, S <: Struct : Scheduler]
  (signalFun: RemoteRef => Signal[A, S],
   registry: Registry)
  (binding: Binding[A => Unit] {type RemoteCall = A => Future[Unit]})
  : Unit = {

    Log.Server.info(s"starting new distribution for »${binding.name}«")

    registry.bindPerRemote(binding)(remoteRef => newValue => {
      val signal: Signal[A, S] = signalFun(remoteRef)
      val signalName           = signal.name.str
      //println(s"received value for $signalName: ${newValue.hashCode()}")
      Scheduler[S].forceNewTransaction(signal) { admissionTicket =>
        admissionTicket.recordChange(new InitialChange[S] {
          override val source = signal
          override def writeValue(b: source.Value, v: source.Value => Unit): Boolean = {
            val merged = b.map(Lattice[A].merge(_, newValue)).asInstanceOf[source.Value]
            if (merged != b) {
              v(merged)
              true
            }
            else false
          }
        })
      }
      Log.Server.info(s"update for $signalName complete")
    })

    var observers = Map[RemoteRef, Observe[S]]()

    def registerRemote(remoteRef: RemoteRef): Unit = {
      val signal: Signal[A, S] = signalFun(remoteRef)
{'fraction_non_alphanumeric': 0.12628797886393658, 'fraction_numerical': 0.0, 'mean_word_length': 3.092972972972973, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 23, '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': '420233', 'n_tokens_mistral': 1167, 'n_tokens_neox': 1111, 'n_words': 272}