title: Writing the contents of a single input stream concurrently to multiple output streams
body: <p>I'm trying to push the contents of a single input stream to multiple output streams obtained from tcp sockets. I didn't find any existing solution, so I built something from scratch but I really feel I'm reinventing the wheel. My requirements are:</p>
  
  <ul>
  <li>Java. Would be nice if it runs on Android as well, but that's optional.</li>
  <li>Up to around 10 clients</li>
  <li>Clients can be frequently added/removed</li>
  <li>Each client should receive roughly the same bytes/sec</li>
  <li>Throughput should decline as little as possible when a new client is added</li>
  <li>Solution should not be specific to certain data (i.e. while I'm testing with raw h264 the solution should handle text streams just as fine)</li>
  </ul>
  
  <p>So, firstly, is there a lib that meets these requirements?</p>
  
  <p>If not, how can I improve the performance of my own solution (see below). I use it like this:</p>
  
  <ul>
  <li>On application startup I run an instance of this class in a new thread.</li>
  <li>If a client connects, the output stream of the respective socket is attached to that instance</li>
  </ul>
  
  <p>While it works, the throughput on each client with N clients attached is extremely unstable. There seems to be no throughput function with respect to N. (Disclaimer: I testet from a single client using multiple threads) I think the performance of this solution is influenced mainly by two things:</p>
  
  <ul>
  <li>Thread synchronization for the consumer collection, adding/removing consumers will block writing to all streams.</li>
  <li>Iteration time on that collection since iterator() will probably create a new instance every time but I need that remove() function.</li>
  </ul>
  
  <p>I'll appreciate any suggestions, thanks.</p>
  
  <pre><code>public class InfiniteStreamingResource implements Runnable {
  
      private LinkedHashSet&lt;OutputStream&gt; consumers;
      private byte[] buf;
      private boolean running;
      InputStream stream;
  
      Logger logger = Logger.getLogger(InfiniteStreamingResource.class);
  
      public InfiniteStreamingResource(InputStream stream, int bufSize) {
          this.stream = stream
          consumers = new LinkedHashSet&lt;&gt;(100);
          buf = new byte[bufSize];
      }
  
      public synchronized void attachConsumer(OutputStream consumer) {
          consumers.add(consumer);
      }
  
      public synchronized void stop() {
          running = false;
      }
  
      @Override
      public void run() {
          running = true;
          int bytesRead;
          Iterator&lt;OutputStream&gt; it;
          OutputStream current;
          try {
              while ((bytesRead = stream.read(buf)) &gt; 0 &amp;&amp; running) {
                  synchronized(this) {
                      it = consumers.iterator();
                      while (it.hasNext()) {
                          current = it.next();
                          try {
                              current.write(buf, 0, bytesRead);
                          } catch (IOException e) {
                              if (e instanceof SocketException) {
                                  it.remove();
                                  try {
                                      current.close();
                                  } catch (IOException inner) {
                                      //ignore
                                  }
                              } else {
                              e.printStackTrace();
                              }
                          }
                      }
                  }
              }
          } catch (IOException e) {
              e.printStackTrace();
          }
  
          for (OutputStream consumer : consumers) { //if stopped, close any remaining streams
              try {
                  consumer.close();
              } catch (IOException e) {
                  //ignore
              }
          }
  
      }
  
  
  }
  </code></pre>