/* Определим блоки */

class A {                                                                     //Класс блока A имеет один выход 'out' который на старте последовательно посылает четыре значения на выход любого другого подключенного блока
    final Stream<Int> out = Stream{1, -1, 1, -1}                              //здесь это (-1)^N но может быть синусоида или что угодно. 
}                                                                        

class B {                                                                     //Класс блока B так же с одним выходом, посылает только одно значение (константу).
    final Unitary<Int> out = Unitary(2)
}

class Base {                                                                  //Базовый класс, скрывающий всю машинерию общую для блоков-матоператоров.

    protected abstract Int doCalc(Array<Int> input)                           //Операция которую должен реализовать субкласс 

    private Array<Queue<Int>> streamBuf = new Array()                         //Буфер для последовательностей входящих значений (для реализации пороговой синхронизации).
    private Array<Option<Int>> unitaryBuf = new Array()                       //Буфер для единичных входящих значений (констант).
                                                                              //Буферы изначально заполнены Queue() и None значениями, и имеющие количество элементов равное количеству подключений к соответсвующему входу.

    final Stream<Int> out = new Empty                                         //Выход, ничего не посылает на старте, "ожидает" вызова .send(Int) метода.

    private void chechIfAllValsReadyAndDoCalcIfSo(){
        while(streamBuf.notContainsEmptyQueue && unitaryBuf.notContainsNone){ //Если получены значения для всех входов,
            out.send(doCalc(                                                  //выполнить операцию и отправить результат на выход,
                streamBuf.map(_.dequeueFirst) ++ unitaryBuf.map(_.get)))      //и очистить буфер входа принимающего последовательность значений (не константу).                                              
        }
    }

    final Stream<Int> in = onReceive((index, value) -> {                      //Вход с логикой заточенной для получения последовательности значений,
        streamBuf[index].enqueue(value)                                       //index - номе подключения к данному входу.
        chechIfAllValsReadyAndDoCalcIfSo() 
    })
    final Unitary<Int> in = onReceive((index, value) -> {                     //Вход для получения единичных значений (констант). 
        unitaryBuf[index] = new Some(value)
        chechIfAllValsReadyAndDoCalcIfSo()
    })
}

class C extends Base {                                                        //Собственно операция суммирование
  protected Int doCalc(Array<Int> input) { return input.sum() } 
}

class P {                                                                     //Класс блока для вывода результата
    final Stream<Int> in = onReceive((_, value) -> { println(value) })
}


/* Создадим несколько блоков */

A a1 = new A() 
A a2 = new A() 
B b = new B() 
C c = new C()
P printer = new P()


/* Соберём из них пару примеров */

a1 ~> c                //Блоки 'a1', 'a2' генерирующие [1, -1, 1, -1], будут подключены с входу 'Stream<Int> in' (т.е. к входу с совместимым типом) блока 'c', 
a2 ~> c ~> printer     //выход блока 'c' будет подключен к 'printer'.
                       //Все четыре блока работают асинхронно, например:
                       // 1. Блок 'a1' получает первым свой квант процессорного времени и отправляет сразу все четыре значения.
                       // 2. Которые попадают в первую очередь в блок 'c' (streamBuf выглядит как [[1,-1,1,-1],[]]), 'c' не отправляет ничего на выход так как вторая очередь пуста.
                       // 3. Блок 'a2' выполняется затем и успевает отправить только 2 значения из 4-х.
                       // 4. Каждый раз при получении нового значения от 'a2', 'c' ставит его в очередь (streamBuf = [[1,-1,1,-1],[1]]) и так как нет пустых очередей 
                       //    забирает по одному элементу из каждой, выполняет операцию, и отправляет результат на свой выход.
                       // 5. Таким образом после обработки 2-х сообщений от 'a2': streamBuf = [[1,-1],[]], блоком 'printer' будет выведено: "2", "-2". 
                       // 6. Соответсвенно по получении оставшихся 2-х сообщений от 'a2': streamBuf = [[],[]], блоком 'printer' будет выведено: "2", "-2", "2", "-2". 
                       //PS: Такой прикольный синтаксис можно сделать в Scala, но в целом это примерно эквивалентно: 
                       //  a1.out.connectTo(c.in)
                       //  a2.out.connectTo(c.in)
                       //  c.out.connectTo(printer.in)

a1 ~> c                //Для блока 'b' генерирующего константу, всё примерно так-же, роме того что он будет подключен к входу 'Unitary<Int> in' с отличной логикой обработки значений.
b  ~> c ~> printer     //Например:
                       // 1. Блок 'a1' опять выполняется первым и отправляет сразу все четыре значения. 
                       // 2. streamBuf = [[1,-1,1,-1]], unitaryBuf = [None]
                       // 3. Блок 'b' выполняется и посылает своё единственное значение: streamBuf = [[1,-1,1,-1]], unitaryBuf = [Some(2)]
                       // 4. Блок 'c' выполняет операцию сложения для всех значений в очереди и отправляет полученные значения на выход.
                       // 5. Блок 'printer' выводит: "3", "1", "3", "1".