Use EmittingPublisher in OutputStreamPublisher to remove busy waiting

[danielkec]

Partially solving #1741, fixing race-condition in EmittingPublisher

Added to IoMulti api

        StringBuilder result = new StringBuilder();

        MultiFromOutputStream osMulti = IoMulti.createOutputStream();

        Single<Void> multiFuture = osMulti
                .map(ByteBuffer::array)
                .map(String::new)
                .forEach(result::append);

        PrintWriter printer = new PrintWriter(osMulti);
        printer.println("test1");
        printer.println("test2");
        printer.println("test3");
        printer.close();

        multiFuture.await();
        assertThat(result.toString(), is(equalTo("test1\ntest2\ntest3\n")));

Signed-off-by: Daniel Kec daniel.kec@oracle.com

[danielkec]

There was a lock in write method but I would expect java.io.OutputStream#write(byte[], int, int) should be always presumed as not thread safe(rule 1.3 should be assured by EmittingPublisher anyway)

          synchronized (invocationLock) {
                if (subscriber.isClosed()) {
                    throw new IOException("Output stream already closed.");
                }

                sub.onNext(createBuffer(buffer, offset, length));
            }

[tomas-langer]

LGTM

[tomas-langer]

LGTM

[olotenko]

The changes to BufferedEmittingPublisher are racy.