Implement Tee directly, revert final removals

See #294

Author: fsteeg

metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectThreader.java

@@ -1,4 +1,4 @@
-/* Copyright 2019 Pascal Christoph, hbz.
+/* Copyright 2019 Pascal Christoph (hbz), and others.
  * 
  * Licensed under the Apache License, Version 2.0 the "License";
  * you may not use this file except in compliance with the License.
@@ -15,14 +15,16 @@
 
 package org.metafacture.flowcontrol;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.metafacture.framework.FluxCommand;
 import org.metafacture.framework.ObjectPipe;
 import org.metafacture.framework.ObjectReceiver;
 import org.metafacture.framework.Tee;
 import org.metafacture.framework.annotations.Description;
 import org.metafacture.framework.annotations.In;
 import org.metafacture.framework.annotations.Out;
-import org.metafacture.framework.helpers.DefaultTee;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,37 +37,72 @@
  * @param <T> Object type
  *
  * @author Pascal Christoph (dr0i)
+ * @author Fabian Steeg (fsteeg)
  * 
  */
 @In(Object.class)
 @Out(Object.class)
 @Description("Incoming objects are distributed to the added receivers, running in their own threads.")
 @FluxCommand("thread-object-tee")
-public class ObjectThreader<T> extends DefaultTee<ObjectReceiver<T>> implements ObjectPipe<T, ObjectReceiver<T>> {
+public class ObjectThreader<T> implements Tee<ObjectReceiver<T>>, ObjectPipe<T, ObjectReceiver<T>> {
 
     private static final Logger LOG = LoggerFactory.getLogger(ObjectThreader.class);
+    private final List<ObjectReceiver<T>> receivers = new ArrayList<ObjectReceiver<T>>();
     private int objectNumber = 0;
 
     @Override
     public void process(final T obj) {
-        getReceivers().get(objectNumber).process(obj);
-        if (objectNumber == getReceivers().size() - 1) {
+        receivers.get(objectNumber).process(obj);
+        if (objectNumber == receivers.size() - 1) {
             objectNumber = 0;
         } else {
             objectNumber++;
         }
     }
 
+    @Override
+    public Tee<ObjectReceiver<T>> addReceiver(final ObjectReceiver<T> receiver) {
+        LOG.info("Adding thread {}", (receivers.size() + 1));
+        ObjectPipeDecoupler<T> opd = new ObjectPipeDecoupler<>();
+        opd.setReceiver(receiver);
+        receivers.add(opd);
+        return this;
+    }
+
     @Override
     public <R extends ObjectReceiver<T>> R setReceiver(final R receiver) {
-        return super.setReceiver(new ObjectPipeDecoupler<T>().setReceiver(receiver));
+        receivers.clear();
+        addReceiver(receiver);
+        return receiver;
     }
 
     @Override
-    public Tee<ObjectReceiver<T>> addReceiver(final ObjectReceiver<T> receiver) {
-        LOG.info("Adding thread {}", (getReceivers().size() + 1));
-        ObjectPipeDecoupler<T> opd = new ObjectPipeDecoupler<>();
-        opd.setReceiver(receiver);
-        return super.addReceiver(opd);
+    public <R extends ObjectReceiver<T>> R setReceivers(R receiver, ObjectReceiver<T> lateralReceiver) {
+        receivers.clear();
+        addReceiver(receiver);
+        addReceiver(lateralReceiver);
+        return receiver;
+    }
+
+    @Override
+    public void resetStream() {
+        receivers.forEach(ObjectReceiver::resetStream);
+    }
+
+    @Override
+    public void closeStream() {
+        receivers.forEach(ObjectReceiver::closeStream);
+    }
+
+    @Override
+    public Tee<ObjectReceiver<T>> removeReceiver(ObjectReceiver<T> receiver) {
+        receivers.remove(receiver);
+        return this;
+    }
+
+    @Override
+    public Tee<ObjectReceiver<T>> clearReceivers() {
+        receivers.clear();
+        return this;
     }
 }

metafacture-framework/src/main/java/org/metafacture/framework/helpers/DefaultTee.java

@@ -35,7 +35,7 @@ public class DefaultTee<T extends Receiver> implements Tee<T> {
     private final List<T> receivers = new ArrayList<T>();
 
     @Override
-    public <R extends T> R setReceiver(final R receiver) {
+    public final <R extends T> R setReceiver(final R receiver) {
         receivers.clear();
         receivers.add(receiver);
         onChangeReceivers();
@@ -52,7 +52,7 @@ public final <R extends T> R setReceivers(final R receiver, final T lateralRecei
     }
 
     @Override
-    public Tee<T> addReceiver(final T receiver) {
+    public final Tee<T> addReceiver(final T receiver) {
         receivers.add(receiver);
         onChangeReceivers();
         return this;