@Test public void creation() throws InterruptedException { final var lock = new ReentrantLock(); final var finished = lock.newCondition(); final var running = new AtomicBoolean(true); final var generation = new AtomicLong(); final Stream<Long> stream = _engine.stream() .limit(33) .map(EvolutionResult::generation); try (var publisher = new StreamPublisher<Long>()) { publisher.subscribe(new Subscriber<>() { private Subscription _subscription; @Override public void onSubscribe(final Subscription subscription) { _subscription = subscription; _subscription.request(1); } @Override public void onNext(final Long g) { generation.set(g); _subscription.request(1); } @Override public void onError(final Throwable throwable) { } @Override public void onComplete() { lock.lock(); try { running.set(false); finished.signal(); } finally { lock.unlock(); } } }); publisher.attach(stream); lock.lock(); try { while (running.get()) { finished.await(); } } finally { lock.unlock(); } } Assert.assertEquals(generation.get(), 33); }
public synchronized void attach(final Stream<? extends T> stream) { requireNonNull(stream); synchronized (_lock) { if (_stream != null) { throw new IllegalStateException( "Already attached evolution stream." ); } _stream = stream.takeWhile(e -> _proceed.get()); _thread = new Thread(() -> { try { _stream.forEach(this::submit); close(); } catch(CancellationException e) { Thread.currentThread().interrupt(); close(); } catch (Throwable e) { closeExceptionally(e); } }); _thread.start(); } }
StreamPublisher extends SubmissionPublisher<T> { public synchronized void attach(final Stream<? extends T> stream) { requireNonNull(stream); synchronized (_lock) { if (_stream != null) { throw new IllegalStateException( "Already attached evolution stream." ); } _stream = stream.takeWhile(e -> _proceed.get()); _thread = new Thread(() -> { try { _stream.forEach(this::submit); close(); } catch(CancellationException e) { Thread.currentThread().interrupt(); close(); } catch (Throwable e) { closeExceptionally(e); } }); _thread.start(); } } }
StreamPublisher extends SubmissionPublisher<T> { public synchronized void attach(final Stream<? extends T> stream) { requireNonNull(stream); synchronized (_lock) { if (_stream != null) { throw new IllegalStateException( "Already attached evolution stream." ); } _stream = stream.takeWhile(e -> _proceed.get()); _thread = new Thread(() -> { try { _stream.forEach(this::submit); close(); } catch(CancellationException e) { Thread.currentThread().interrupt(); close(); } catch (Throwable e) { closeExceptionally(e); } }); _thread.start(); } } StreamPublisher(
		final Executor executor,
		final int maxBufferCapacity,
		final BiConsumer<? super Subscriber<? super T>, ? super Throwable> handler
	);  StreamPublisher(final Executor executor, final int maxBufferCapacity);  StreamPublisher(); }
StreamPublisher extends SubmissionPublisher<T> { public synchronized void attach(final Stream<? extends T> stream) { requireNonNull(stream); synchronized (_lock) { if (_stream != null) { throw new IllegalStateException( "Already attached evolution stream." ); } _stream = stream.takeWhile(e -> _proceed.get()); _thread = new Thread(() -> { try { _stream.forEach(this::submit); close(); } catch(CancellationException e) { Thread.currentThread().interrupt(); close(); } catch (Throwable e) { closeExceptionally(e); } }); _thread.start(); } } StreamPublisher(
		final Executor executor,
		final int maxBufferCapacity,
		final BiConsumer<? super Subscriber<? super T>, ? super Throwable> handler
	);  StreamPublisher(final Executor executor, final int maxBufferCapacity);  StreamPublisher(); synchronized void attach(final Stream<? extends T> stream); @Override void close(); }
StreamPublisher extends SubmissionPublisher<T> { public synchronized void attach(final Stream<? extends T> stream) { requireNonNull(stream); synchronized (_lock) { if (_stream != null) { throw new IllegalStateException( "Already attached evolution stream." ); } _stream = stream.takeWhile(e -> _proceed.get()); _thread = new Thread(() -> { try { _stream.forEach(this::submit); close(); } catch(CancellationException e) { Thread.currentThread().interrupt(); close(); } catch (Throwable e) { closeExceptionally(e); } }); _thread.start(); } } StreamPublisher(
		final Executor executor,
		final int maxBufferCapacity,
		final BiConsumer<? super Subscriber<? super T>, ? super Throwable> handler
	);  StreamPublisher(final Executor executor, final int maxBufferCapacity);  StreamPublisher(); synchronized void attach(final Stream<? extends T> stream); @Override void close();  }