package cn.netkiller.test;

import java.util.concurrent.Flow;

public class Test {
    public static void main(String[] args) {

        IntPublisher intPublisher = new IntPublisher();
        IntSubscriber intSubscriber = new IntSubscriber();
        intPublisher.subscribe(intSubscriber);
    }

    public static class IntPublisher implements Flow.Publisher<Integer> {

        @Override
        public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
            for (int i = 1; i <= 5; i++) {
                System.out.println("Publishing = " + i);
                // 将数据发给订阅者
                subscriber.onNext(i);
            }
            // 发出完成信号
            subscriber.onComplete();
        }
    }

    public static class IntSubscriber implements Flow.Subscriber<Integer> {
        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("onSubscribe");
        }

        @Override
        public void onNext(Integer item) {
            System.out.println("onNext: " + item);
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("onError：" + throwable);
        }

        @Override
        public void onComplete() {
            System.out.println("onComplete");
        }
    }
}			
			
			

		
package cn.netkiller.test;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.IntStream;

public class SubmissionPublisherTest {
    public static void main(String[] args) throws InterruptedException {
        System.out.println(Thread.currentThread().getName());

        // 1. 定义 String 类型的数据发布者，SubmissionPublisher 实现了 Publisher
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // 2. 创建一个订阅者，用于接收发布者的消息
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                System.out.println("onSubscribe");
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("onNext: " + item);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("Throwable: " + throwable.getMessage());
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };

        // 3. 发布者和订阅者需要建立关系
        publisher.subscribe(subscriber);
        publisher.submit("https://www.netkiller.cn");

        // 4. 发布者开始发布数据
        IntStream.range(0, 11).mapToObj(i -> "publisher -----> message " + i).peek(message -> publisher.submit(message)).forEach(System.out::println);

        // 5. 发布结束后，关闭发布者
        publisher.close();

        // main线程延迟关闭，不然订阅者还没接收完消息，线程就被关闭了
        Thread.currentThread().join(1000);
    }

}
		
			

			
main
onSubscribe
onNext: https://www.netkiller.cn
publisher -----> message 0
publisher -----> message 1
publisher -----> message 2
publisher -----> message 3
publisher -----> message 4
onNext: publisher -----> message 0
onNext: publisher -----> message 1
onNext: publisher -----> message 2
onNext: publisher -----> message 3
onNext: publisher -----> message 4
onNext: publisher -----> message 5
publisher -----> message 5
publisher -----> message 6
publisher -----> message 7
publisher -----> message 8
publisher -----> message 9
publisher -----> message 10
onNext: publisher -----> message 6
onNext: publisher -----> message 7
onNext: publisher -----> message 8
onNext: publisher -----> message 9
onNext: publisher -----> message 10
onComplete			
			
			

			
package cn.netkiller.test;


import lombok.SneakyThrows;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class Test {
    @SneakyThrows
    public static void main(String[] args) {
        Test test = new Test();
        try (final var publisher = new SubmissionPublisher<String>()) {

            final var processor = new UpperCaseProcessor();
            publisher.subscribe(processor);

            processor.subscribe(new Flow.Subscriber<String>() {
                                    private Flow.Subscription subscription;

                                    @Override
                                    public void onSubscribe(Flow.Subscription subscription) {
                                        this.subscription = subscription;
                                        System.out.println("Subscriber.onSubscribe");
                                        subscription.request(1);
                                    }

                                    @Override
                                    public void onNext(String item) {
                                        System.out.println("Subscriber.onNext: " + item);
                                        this.subscription.request(1);
                                    }

                                    @Override
                                    public void onError(Throwable throwable) {
                                        System.out.println("Subscriber.onError: " + throwable);
                                    }

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

            publisher.submit("abc");
            publisher.submit("xyz");
            publisher.submit("neo");
            publisher.submit("netkiller");
        }
        Thread.currentThread().join(1000);
    }


    static class UpperCaseProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            System.out.println("Processor.onSubscribe");
            subscription.request(1);
        }

        @Override
        public void onNext(String item) {
            System.out.println("Processor.onNext: " + item);
            submit(item.toUpperCase());
            this.subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("Processor.onError: " + throwable.getMessage());
            closeExceptionally(throwable);
        }

        @Override
        public void onComplete() {
            System.out.println("Processor.onComplete");
            close();
        }
    }
}
			
			
			

			
Processor.onSubscribe
Subscriber.onSubscribe
Processor.onNext: abc
Processor.onNext: xyz
Processor.onNext: neo
Processor.onNext: netkiller
Processor.onComplete
Subscriber.onNext: ABC
Subscriber.onNext: XYZ
Subscriber.onNext: NEO
Subscriber.onNext: NETKILLER
Subscriber.onComplete