Andronik_D_S_Synchronizer

src/main/java/hse/java/lectures/lecture6/tasks/synchronizer/StreamWriter.java

@@ -28,9 +28,18 @@ public void attachMonitor(StreamingMonitor monitor) {
     public void run() {
         // Writer threads are intentionally infinite for the task contract.
         while (true) {
-            output.print(message);
-            onTick.run();
+            try {
+                if(!monitor.checkTurn(id)) {
+                    return;
+                }
+                output.print(message);
+                onTick.run();
+                monitor.TickFinished();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            }
         }
     }
 
-}
+}

src/main/java/hse/java/lectures/lecture6/tasks/synchronizer/StreamingMonitor.java

@@ -1,5 +1,45 @@
 package hse.java.lectures.lecture6.tasks.synchronizer;
 
+import java.util.List;
+
 public class StreamingMonitor {
-    // impl your sync here
-}
+    private int currentWriter;
+    private int ticksPerWriter;
+    private int totalWriters;
+    private int[] ticks;
+    private int total;
+
+    StreamingMonitor(List<StreamWriter> tasks, int ticksPerWriter) {
+        this.ticksPerWriter = ticksPerWriter;
+        this.totalWriters = tasks.size();
+        this.ticks = new int[totalWriters + 1];
+        this.total = 0;
+        this.currentWriter = 1;
+    }
+
+    synchronized boolean checkTurn(int writer) throws InterruptedException {
+        while (writer != currentWriter && total < totalWriters * ticksPerWriter) {
+            wait();
+        }
+        return total < totalWriters * ticksPerWriter;
+    }
+
+    synchronized void TickFinished() {
+        ticks[currentWriter]++;
+        total++;
+
+        if (total == totalWriters * ticksPerWriter) {
+            notifyAll();
+            return;
+        }
+
+        int next = currentWriter % totalWriters + 1;
+
+        while (ticks[next] >= ticksPerWriter) {
+            next = next % totalWriters + 1;
+        }
+        currentWriter = next;
+
+        notifyAll();
+    }
+}

src/main/java/hse/java/lectures/lecture6/tasks/synchronizer/Synchronizer.java

@@ -1,5 +1,6 @@
 package hse.java.lectures.lecture6.tasks.synchronizer;
 
+import java.util.ArrayList;
 import java.util.List;
 
 public class Synchronizer {
@@ -23,11 +24,27 @@ public Synchronizer(List<StreamWriter> tasks, int ticksPerWriter) {
      */
     public void execute() {
         // add monitor and sync
+        StreamingMonitor monitor = new StreamingMonitor(tasks, ticksPerWriter);
+        for (StreamWriter writer : tasks) {
+            writer.attachMonitor(monitor);
+        }
+
+        List<Thread> workers = new ArrayList<>();
         for (StreamWriter writer : tasks) {
             Thread worker = new Thread(writer, "stream-writer-" + writer.getId());
             worker.setDaemon(true);
             worker.start();
+            workers.add(worker);
+        }
+
+        for (Thread worker : workers) {
+            try {
+                worker.join();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            }
         }
     }
 
-}
+}