package org.reactivestreams;

public interface Publisher<T> {  
    public void subscribe(Subscriber<? super T> s);  
}

public interface Subscriber<T> {  
    public void onSubscribe(Subscription s);  
    public void onNext(T t);  
    public void onError(Throwable t);  
    public void onComplete();  
}

public interface Subscription {  
    public void request(long n);  
    public void cancel();  
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {  
}

import reactor.core.publisher.Mono;

public class SimpleMonoDemo {  
    public static void main(String[] args) {  
        Mono<String> helloWorld = Mono.just("hello world");  
        helloWorld.subscribe(System.out::println);  
    }  
}

import reactor.core.publisher.Mono;

public class ComplexMonoDemo {  
    public static void main(String[] args) {  
        Mono<String> helloWorld = Mono.just("hello world")  
                .filter(s -> !s.isEmpty()) // 过滤空字符串  
                .map(String::toUpperCase); // 转换为大写  
        helloWorld.subscribe(System.out::println);  
    }  
}

import org.reactivestreams.*;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class MinimalReactiveDemo {

    // 发布者实现
    static class SimplePublisher<T> implements Publisher<T> {
        private final List<T> data;

        public SimplePublisher(List<T> data) {
            this.data = data;
        }

        @Override
        public void subscribe(Subscriber<? super T> s) {
            SimpleSubscription<T> subscription = new SimpleSubscription<>(s, data);
            s.onSubscribe(subscription);
        }
    }

    // 订阅者实现
    static class SimpleSubscriber<T> implements Subscriber<T> {
        private Subscription subscription;

        @Override
        public void onSubscribe(Subscription s) {
            System.out.println("[Subscriber] Subscribed. Requesting data...");
            this.subscription = s;
            s.request(1); // 初始请求 1 条数据
        }

        @Override
        public void onNext(T item) {
            System.out.println("[Subscriber] Received: " + item);
            subscription.request(1); // 持续请求下一条数据
        }

        @Override
        public void onError(Throwable t) {
            System.err.println("[Subscriber] Error: " + t.getMessage());
        }

        @Override
        public void onComplete() {
            System.out.println("[Subscriber] Completed.");
        }
    }

    // 订阅关系实现
    static class SimpleSubscription<T> implements Subscription {
        private final Subscriber<? super T> subscriber;
        private final List<T> data;
        private final AtomicInteger index = new AtomicInteger(0);
        private final ExecutorService executor = Executors.newSingleThreadExecutor();
        private volatile boolean cancelled = false;

        public SimpleSubscription(Subscriber<? super T> subscriber, List<T> data) {
            this.subscriber = subscriber;
            this.data = data;
        }

        @Override
        public void request(long n) {
            if (cancelled) return;

            executor.submit(() -> {
                for (int i = 0; i < n && !cancelled; i++) {
                    int currentIndex = index.getAndIncrement();
                    if (currentIndex < data.size()) {
                        subscriber.onNext(data.get(currentIndex));
                    } else {
                        subscriber.onComplete();
                        executor.shutdown();
                        return;
                    }
                }
            });
        }

        @Override
        public void cancel() {
            cancelled = true;
            executor.shutdownNow();
            System.out.println("[Subscription] Cancelled.");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        List<String> items = List.of("Item 1", "Item 2", "Item 3", "Item 4", "Item 5");
        SimplePublisher<String> publisher = new SimplePublisher<>(items);
        SimpleSubscriber<String> subscriber = new SimpleSubscriber<>();

        publisher.subscribe(subscriber);
        Thread.sleep(2000); // 等待异步处理完成
    }
}

[Subscriber] Subscribed. Requesting data...
[Subscriber] Received: Item 1
[Subscriber] Received: Item 2
[Subscriber] Received: Item 3
[Subscriber] Received: Item 4
[Subscriber] Received: Item 5
[Subscriber] Completed.

public static <T> Mono<T> just(T data) {  
    return onAssembly(new MonoJust<>(data));  
}

final class MonoJust<T> extends Mono<T> implements Fuseable.ScalarCallable<T>, Fuseable, SourceProducer<T> 
public abstract class Mono<T> implements CorePublisher<T> 
public interface CorePublisher<T> extends Publisher<T>
interface SourceProducer<O> extends Scannable, Publisher<O>

public void subscribe(CoreSubscriber<? super T> actual) {  
    actual.onSubscribe(Operators.scalarSubscription(actual, value));  
}

Mono.just("hello world").subscribe(System.out::println);

public final Disposable subscribe(Consumer<? super T> consumer) {  
    Objects.requireNonNull(consumer, "consumer");  
    return subscribe(consumer, null, null);  
}

public final Disposable subscribe(@Nullable Consumer<? super T> consumer,  
                                 @Nullable Consumer<? super Throwable> errorConsumer,  
                                 @Nullable Runnable completeConsumer,  
                                 @Nullable Context initialContext) {  
    return subscribeWith(new LambdaMonoSubscriber<>(consumer, errorConsumer, completeConsumer, null, initialContext));  
}

public void request(long n) {  
    if (validate(n)) {  
        if (ONCE.compareAndSet(this, 0, 1)) {  
            Subscriber<? super T> a = actual;  
            a.onNext(value);  
            if (once != 2) {  
                a.onComplete();  
            }  
        }  
    }  
}

Mono<String> helloWorld = Mono.just("hello world")  
        .filter(s -> !s.isEmpty())  
        .map(String::toUpperCase);  
helloWorld.subscribe(System.out::println);

public final Mono<T> filter(final Predicate<? super T> tester) {  
    if (this instanceof Fuseable) {  
        return onAssembly(new MonoFilterFuseable<>(this, tester));  
    }  
    return onAssembly(new MonoFilter<>(this, tester));  
}

public final <R> Mono<R> map(Function<? super T, ? extends R> mapper) {  
    if (this instanceof Fuseable) {  
        return onAssembly(new MonoMapFuseable<>(this, mapper));  
    }  
    return onAssembly(new MonoMap<>(this, mapper));  
}

MonoJust -> MonoFilter -> MonoMap

final class MonoFilter<T> extends InternalMonoOperator<T, T>
abstract class InternalMonoOperator<I, O> extends MonoOperator<I, O> implements Scannable, OptimizableOperator<O, I>
public abstract class MonoOperator<I, O> extends Mono<O> implements Scannable

public final void subscribe(CoreSubscriber<? super O> subscriber) {  
    OptimizableOperator operator = this;  
    try {  
        while (true) {  
            subscriber = operator.subscribeOrReturn(subscriber);  
            if (subscriber == null) {  
                return;  
            }  
            OptimizableOperator newSource = operator.nextOptimizableSource();  
            if (newSource == null) {  
                CorePublisher operatorSource = operator.source();  
                subscriber = Operators.restoreContextOnSubscriberIfPublisherNonInternal(operatorSource, subscriber);  
                operatorSource.subscribe(subscriber);  
                return;  
            }  
            operator = newSource;  
        }  
    } catch (Throwable e) {  
        Operators.reportThrowInSubscribe(subscriber, e);  
    }  
}

public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) {  
    if (actual instanceof ConditionalSubscriber) {  
        return new FluxFilter.FilterConditionalSubscriber<>((ConditionalSubscriber<? super T>) actual, predicate);  
    }  
    return new FluxFilter.FilterSubscriber<>(actual, predicate);  
}

public void onNext(T t) {  
    if (done) {  
        Operators.onNextDropped(t, this.ctx);  
        return;  
    }  
    boolean b;  
    try {  
        b = predicate.test(t);  
    } catch (Throwable e) {  
        onError(Operators.onNextError(t, e, this.ctx, s));  
        return;  
    }  
    if (b) {  
        actual.onNext(t); // 传递给 MapSubscriber
    } else {  
        s.request(1); // 请求下一条数据
    }  
}