Flowable Streams

/**
 * Creates a Flowable that emits the provided items then completes
 */
public static <T> Flowable<T> just(T item);
public static <T> Flowable<T> just(T item1, T item2);
// ... up to 10 items

/**
 * Creates a Flowable from an array
 */
public static <T> Flowable<T> fromArray(T... array);

/**
 * Creates a Flowable from an Iterable
 */
public static <T> Flowable<T> fromIterable(Iterable<? extends T> source);

/**
 * Creates a Flowable using the provided FlowableOnSubscribe with backpressure strategy
 */
public static <T> Flowable<T> create(FlowableOnSubscribe<T> source, BackpressureStrategy mode);

/**
 * Creates a Flowable from a Publisher (Reactive Streams compatibility)
 */
public static <T> Flowable<T> fromPublisher(Publisher<? extends T> publisher);

/**
 * Creates a Flowable that emits sequential numbers at intervals
 */
public static Flowable<Long> interval(long period, TimeUnit unit);
public static Flowable<Long> interval(long initialDelay, long period, TimeUnit unit);

/**
 * Creates a Flowable that emits a range of integers
 */
public static Flowable<Integer> range(int start, int count);

/**
 * Creates a Flowable that generates items using a generator function
 */
public static <T> Flowable<T> generate(Consumer<Emitter<T>> generator);
public static <S, T> Flowable<T> generate(Callable<S> initialState, BiConsumer<S, Emitter<T>> generator);

/**
 * Buffers all items until the downstream is ready to receive them
 */
public final Flowable<T> onBackpressureBuffer();
public final Flowable<T> onBackpressureBuffer(int capacity);
public final Flowable<T> onBackpressureBuffer(int capacity, Action onOverflow);

/**
 * Drops items when downstream can't keep up
 */
public final Flowable<T> onBackpressureDrop();
public final Flowable<T> onBackpressureDrop(Consumer<? super T> onDrop);

/**
 * Keeps only the latest item when downstream can't keep up
 */
public final Flowable<T> onBackpressureLatest();

/**
 * Reduces backpressure by sampling items at regular intervals
 */
public final Flowable<T> sample(long period, TimeUnit unit);
public final Flowable<T> sample(long period, TimeUnit unit, Scheduler scheduler);

/**
 * Throttles items by only emitting the first item in each time window
 */
public final Flowable<T> throttleFirst(long windowDuration, TimeUnit unit);

/**
 * Throttles items by only emitting the last item in each time window
 */
public final Flowable<T> throttleLast(long intervalDuration, TimeUnit unit);

/**
 * Debounces items by only emitting when no new items arrive for a specified duration
 */
public final Flowable<T> debounce(long timeout, TimeUnit unit);
public final Flowable<T> debounce(long timeout, TimeUnit unit, Scheduler scheduler);

/**
 * Transforms items using a function
 */
public final <R> Flowable<R> map(Function<? super T, ? extends R> mapper);

/**
 * FlatMap variants with backpressure handling
 */
public final <R> Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper);
public final <R> Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper, int maxConcurrency);
public final <R> Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper, boolean delayErrors);

/**
 * ConcatMap maintains order and handles backpressure
 */
public final <R> Flowable<R> concatMap(Function<? super T, ? extends Publisher<? extends R>> mapper);
public final <R> Flowable<R> concatMap(Function<? super T, ? extends Publisher<? extends R>> mapper, int prefetch);

/**
 * SwitchMap for latest values only
 */
public final <R> Flowable<R> switchMap(Function<? super T, ? extends Publisher<? extends R>> mapper);

/**
 * Accumulation with scan
 */
public final <R> Flowable<R> scan(BiFunction<R, ? super T, R> accumulator);
public final <R> Flowable<R> scan(R initialValue, BiFunction<R, ? super T, R> accumulator);

/**
 * Grouping with backpressure handling
 */
public final <K> Flowable<GroupedFlowable<K, T>> groupBy(Function<? super T, ? extends K> keySelector);

/**
 * Subscribe with FlowableSubscriber for full Reactive Streams compliance
 */
public final void subscribe(FlowableSubscriber<? super T> subscriber);

/**
 * Subscribe with simple callbacks (automatically requests unbounded)
 */
public final Disposable subscribe(Consumer<? super T> onNext);
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError);
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete);

/**
 * Convert to Observable (loses backpressure handling)
 */
public final Observable<T> toObservable();

/**
 * Blocking operations
 */
public final T blockingFirst();
public final T blockingLast();
public final T blockingSingle();
public final Iterable<T> blockingIterable();

/**
 * Creates a ParallelFlowable for parallel processing
 */
public final ParallelFlowable<T> parallel();
public final ParallelFlowable<T> parallel(int parallelism);
public final ParallelFlowable<T> parallel(int parallelism, int prefetch);

import io.reactivex.Flowable;
import io.reactivex.BackpressureStrategy;
import io.reactivex.FlowableSubscriber;
import org.reactivestreams.Subscription;

Flowable<Integer> source = Flowable.create(emitter -> {
    for (int i = 1; i <= 1000; i++) {
        if (emitter.isCancelled()) {
            return;
        }
        emitter.onNext(i);
    }
    emitter.onComplete();
}, BackpressureStrategy.BUFFER);

source.subscribe(new FlowableSubscriber<Integer>() {
    private Subscription subscription;
    
    @Override
    public void onSubscribe(Subscription s) {
        subscription = s;
        subscription.request(1); // Request first item
    }
    
    @Override
    public void onNext(Integer integer) {
        System.out.println("Received: " + integer);
        subscription.request(1); // Request next item
    }
    
    @Override
    public void onError(Throwable t) {
        t.printStackTrace();
    }
    
    @Override
    public void onComplete() {
        System.out.println("Complete");
    }
});

// Producer that emits items very quickly
Flowable<Integer> fastProducer = Flowable.create(emitter -> {
    for (int i = 0; i < 1000000; i++) {
        emitter.onNext(i);
    }
    emitter.onComplete();
}, BackpressureStrategy.MISSING);

// Strategy 1: Buffer all items
fastProducer
    .onBackpressureBuffer()
    .observeOn(Schedulers.io())
    .subscribe(item -> {
        Thread.sleep(1); // Slow consumer
        System.out.println("Buffered: " + item);
    });

// Strategy 2: Drop items when buffer is full
fastProducer
    .onBackpressureDrop(dropped -> System.out.println("Dropped: " + dropped))
    .subscribe(item -> System.out.println("Received: " + item));

// Strategy 3: Keep only latest
fastProducer
    .onBackpressureLatest()
    .subscribe(item -> System.out.println("Latest: " + item));

import org.reactivestreams.Publisher;
import java.util.concurrent.Flow;

// Converting to/from Reactive Streams Publisher
Publisher<String> publisher = Flowable.just("Hello", "World");

Flowable<String> fromPublisher = Flowable.fromPublisher(publisher);

// Java 9+ Flow interoperability
Flow.Publisher<String> flowPublisher = fromPublisher.toFlowPublisher();

Flowable.range(1, 1000)
    .parallel(4) // Split into 4 parallel streams
    .runOn(Schedulers.computation()) // Each stream runs on computation scheduler
    .map(i -> i * i) // Square each number in parallel
    .sequential() // Merge back into single stream
    .subscribe(result -> System.out.println("Result: " + result));

Flowable<Integer> source1 = Flowable.range(1, 5);
Flowable<Integer> source2 = Flowable.range(6, 5);

// Merge with backpressure handling
Flowable.merge(source1, source2)
    .subscribe(item -> System.out.println("Merged: " + item));

// Zip with proper flow control
Flowable.zip(source1, source2, (a, b) -> a + b)
    .subscribe(sum -> System.out.println("Sum: " + sum));

/**
 * Functional interface for creating Flowable with backpressure handling
 */
public interface FlowableOnSubscribe<T> {
    void subscribe(FlowableEmitter<T> emitter) throws Exception;
}

/**
 * Emitter for FlowableOnSubscribe
 */
public interface FlowableEmitter<T> extends Emitter<T> {
    void setDisposable(Disposable d);
    void setCancellable(Cancellable c);
    long requested();
    boolean isCancelled();
}

/**
 * Subscriber interface compatible with Reactive Streams
 */
public interface FlowableSubscriber<T> extends Subscriber<T> {
    // Inherits from org.reactivestreams.Subscriber
}

/**
 * Grouped Flowable for groupBy operations
 */
public abstract class GroupedFlowable<K, T> extends Flowable<T> {
    public abstract K getKey();
}