MuxSink implements Sink<Object> { public void add(Sink<Object>... s) { int i = sinks.length; sinks = Arrays.copyOf(sinks, i + s.length); for (Sink<Object> ss : s) { sinks[i++] = ss; } } MuxSink(Sink<Object>... s); @SuppressWarnings("unchecked")  MuxSink(); @Override void put(Object tuple); void add(Sink<Object>... s); void remove(Sink<Object> s); void remove(Sink<Object>... s); Sink<Object>[] getSinks(); @Override int getCount(boolean reset);  }
@Test @SuppressWarnings({"unchecked"}) public void testAdd() { Sink<Object> a = new Sink<Object>() { @Override public void put(Object tuple) { throw new UnsupportedOperationException("Not supported yet."); } @Override public int getCount(boolean reset) { throw new UnsupportedOperationException("Not supported yet."); } }; Sink<Object> b = new Sink<Object>() { @Override public void put(Object tuple) { throw new UnsupportedOperationException("Not supported yet."); } @Override public int getCount(boolean reset) { throw new UnsupportedOperationException("Not supported yet."); } }; MuxSink instance1 = new MuxSink(a, b); Assert.assertEquals("2 sinks", instance1.getSinks().length, 2); MuxSink instance2 = new MuxSink(); instance2.add(a, b); Assert.assertEquals("2 sinks", instance2.getSinks().length, 2); }