content: package io.activej.datastream.supplier.impl;
  
  import io.activej.common.annotation.ExposedInternals;
  import io.activej.datastream.consumer.AbstractStreamConsumer;
  import io.activej.datastream.supplier.AbstractStreamSupplier;
  import io.activej.datastream.supplier.StreamDataAcceptor;
  import io.activej.datastream.supplier.StreamSupplier;
  import io.activej.reactor.Reactor;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.List;
  
  @ExposedInternals
  public final class OfAnotherReactor<T> extends AbstractStreamSupplier<T> {
  	public static final int MAX_BUFFER_SIZE = 100;
  	public static final Iterator<?> END_OF_STREAM = Collections.emptyIterator();
  
  	public volatile Iterator<T> iterator;
  	public volatile boolean isReady;
  
  	public final StreamSupplier<T> anotherReactorSupplier;
  	public final InternalConsumer internalConsumer;
  	public volatile boolean wakingUp;
  
  	public OfAnotherReactor(Reactor anotherReactor, StreamSupplier<T> anotherReactorSupplier) {
  		this.anotherReactorSupplier = anotherReactorSupplier;
  		this.internalConsumer = Reactor.executeWithReactor(anotherReactor, InternalConsumer::new);
  	}
  
  	private void execute(Runnable runnable) {
  		reactor.execute(runnable);
  	}
  
  	private void wakeUp() {
  		if (wakingUp) return;
  		wakingUp = true;
  		execute(this::onWakeUp);
  	}
  
  	private void onWakeUp() {
  		if (isComplete()) return;
  		wakingUp = false;
  		flush();
  	}
  
  	@Override
  	protected void onInit() {
  		reactor.startExternalTask();
  	}
  
  	@Override
  	protected void onStarted() {
  		internalConsumer.execute(() ->
  			anotherReactorSupplier.streamTo(internalConsumer));
  	}
  
  	@Override
  	protected void onResumed() {
  		isReady = true;
  		flush();
  	}
  
  	@Override
  	protected void onSuspended() {
  		isReady = false;
  		internalConsumer.wakeUp();
  	}
  
  	@Override
  	protected void onComplete() {
  		reactor.completeExternalTask();
  	}
  
  	private void flush() {
  		if (iterator == null) {
  			internalConsumer.wakeUp();
  			return;
  		}
  		Iterator<T> iterator = this.iterator;
  		while (isReady() && iterator.hasNext()) {
  			send(iterator.next());
  		}
  		if (iterator == END_OF_STREAM) {
  			sendEndOfStream();
  		} else if (!iterator.hasNext()) {
  			this.iterator = null;
  			internalConsumer.wakeUp();
  		}
  	}
  
  	@Override
  	protected void onAcknowledge() {
  		internalConsumer.execute(internalConsumer::acknowledge);
  	}
  
  	@Override
  	protected void onError(Exception e) {
  		internalConsumer.execute(() -> internalConsumer.closeEx(e));
  	}
  
  	@Override
  	protected void onCleanup() {
  		this.iterator = null;
  	}
  
  	public final class InternalConsumer extends AbstractStreamConsumer<T> {
  		private List<T> list = new ArrayList<>();
  		private final StreamDataAcceptor<T> toList = item -> {
  			list.add(item);
  			if (list.size() == MAX_BUFFER_SIZE) {
  				flush();
  				suspend();
  			}
  		};
  		volatile boolean wakingUp;
  
  		void execute(Runnable runnable) {
  			reactor.execute(runnable);
  		}
  
  		void wakeUp() {
  			if (wakingUp) return;
  			wakingUp = true;
  			execute(this::onWakeUp);
  		}
  
  		void onWakeUp() {
  			if (isComplete()) return;
  			wakingUp = false;
  			flush();
  			if (OfAnotherReactor.this.isReady) {
  				resume(toList);
  			} else {
  				suspend();
  			}
  		}
  
  		@Override
  		protected void onInit() {
  			reactor.startExternalTask();
  		}
  
  		@Override
  		protected void onEndOfStream() {
  			flush();
  		}
  
  		private void flush() {
  			if (OfAnotherReactor.this.iterator != null) return;
  			if (this.isEndOfStream() && this.list.isEmpty()) {
  				//noinspection unchecked
  				OfAnotherReactor.this.iterator = (Iterator<T>) END_OF_STREAM;
  			} else if (!this.list.isEmpty()) {
  				OfAnotherReactor.this.iterator = this.list.iterator();
  				this.list = new ArrayList<>();
  			} else {
  				return;
  			}
  			OfAnotherReactor.this.wakeUp();
  		}
  
  		@Override
  		protected void onError(Exception e) {
  			OfAnotherReactor.this.execute(() -> OfAnotherReactor.this.closeEx(e));
  		}
  
  		@Override
  		protected void onComplete() {
  			reactor.completeExternalTask();
  		}
  
  		@Override
  		protected void onCleanup() {
  			this.list = null;
  		}
  	}
  
  }
  