Simple understanding of responsive programming

Responsive programming

• Reference website
• Programming model
• The basic object
  • Flow.Publisher<T>
  • Flow.Subscription<T>
  • Flow.Subscriber<T>
  • Flow.Processor<T>
• Simple actual combat
  • Main task
  • The first stage
  • The second stage

Reference website

• github Address
• Responsive flow
• geek - In depth analysis of Java New characteristics - Fan Xuelei

Programming model

Non blocking asynchronous programming , Mapping is essentially a callback function in code , Corresponding to the reactive programming model is the traditional Command programming ;

• Instruction programming model : Synchronous blocking , Tell the computer What to do , What controls is the state
• Responsive programming model : Asynchronous non-blocking , Tell the computer What to do , Control is the goal

The basic object

With Java 9 Of api For example :

Flow.Publisher<T>

Publisher , The object of data input , T Represents the type of data

    @FunctionalInterface
    public static interface Publisher<T> {
        public void subscribe(Subscriber<? super T> subscriber);
    }

Flow.Subscription<T>

An intermediary agent , Publishers and subscribers have no direct contact , Instead, it controls the transmission of data Separate from data and changes in data , Thus reducing the coupling between functions

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

Flow.Subscriber<T>

subscriber ,T Describe the type of data , The reactions in four cases are specified respectively :

• What if you receive a subscription invitation ？ This behavior is caused by onSubscribe
• The implementation of this method determines . What if data is received ？ This behavior is caused by onNext
• The implementation of this method determines . What to do if you encounter an error ？ This behavior is caused by onError
• The implementation of this method determines . What to do if the data transmission is completed ？ This behavior is caused by onComplete The implementation of this method determines .

 public static interface Subscriber<T> {

        public void onSubscribe(Subscription subscription);

        public void onNext(T item);

        public void onError(Throwable throwable);

        public void onComplete();
    }

Flow.Processor<T>

processor , At the same time Publisher and Subscriber, That is to do a connecting role , It is an intermediate process of data processing during the execution of a stream ;

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

Simple try

Main task

 Use Flow Of api  Realization : a=b+c
a:  Suppose you need to know the day of the week when an event ends 
b: The start time of the data is the day of the week   Don't work hard on Friday, Saturday and Sunday , Until Monday  , The actual start time is Monday ,
c: It will take several days to deal with this matter  ,  Monday   It takes two days ,  Tuesday, Wednesday and Thursday take a day ,

NOTE	No multithreading , Just want to know something about the responsive code execution

The first stage

bc Do not split , Fast business implementation

Publisher

public class TimePublisher implements Flow.Publisher<LocalDate> {

    public final LinkedList<LocalDate> items;
    public TimePublisher(LinkedList<LocalDate> items) {
        this.items = items;
    }
    @Override
    public void subscribe(Flow.Subscriber subscriber) {
        subscriber.onSubscribe(new TimeSubscription(subscriber, items));
    }
}

Subscription

public class TimeSubscription implements Flow.Subscription {

    private final Flow.Subscriber<LocalDate> subscriber;

    private LinkedList<LocalDate> list;

    public TimeSubscription(Flow.Subscriber<LocalDate> subscriber, LinkedList<LocalDate> list) {
        this.subscriber = subscriber;
        this.list = list;
    }

    @Override
    public void request(long n) {
        if (list.isEmpty()) {
            subscriber.onComplete();
            return;
        }
        for (long i = 0; i < n; i++) {
            LocalDate item = list.remove();
            subscriber.onNext(item);
        }
    }


    @Override
    public void cancel() {
        // Mark as delete ;  Close resources, etc 
    }
}

Subscriber

public class TimeSubscriber implements Flow.Subscriber<LocalDate> {

    private Flow.Subscription subscription;

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

    @Override
    public void onNext(LocalDate item) {
        DayOfWeek dayOfWeek = item.getDayOfWeek();
        int days = startAfterDay(dayOfWeek);
        LocalDate localDate = item.plusDays(days);
        System.out.println(" The completion date is : " + localDate);

        subscription.request(1);
    }

    private int startAfterDay(DayOfWeek dayOfWeek) {
        return switch (dayOfWeek) {
            case MONDAY -> 2;
            case TUESDAY, THURSDAY, WEDNESDAY -> 1;
            case FRIDAY -> 5;
            case SATURDAY -> 4;
            case SUNDAY -> 3;
        };
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}

Main Method

    public static void main(String[] args) {
        // First step ,  Declare a publisher 
        LinkedList<LocalDate> linkedList = new LinkedList();
        linkedList.add(LocalDate.now());

        TimePublisher timePublisher = new TimePublisher(linkedList);
        // Second parts , Declaration process ,
        TimeSubscriber timeSubscriber = new TimeSubscriber();
        timePublisher.subscribe(timeSubscriber);
    }

The second stage

Disassemble the execution process into two processes , Use Processor,

Add a process

The day of the week is the middle result of the flow , It is both the upstream result and the downstream data source ; Try to write the code

Intermediate processor for calculating the day of the week

Here, the code of the first phase is split directly , therefore Publisher and Subscription unchanged , and subscriber It becomes Processor, Became an intermediate processor

public class TimeProcessor implements Flow.Processor<LocalDate,DayOfWeek> {

    private Flow.Subscription subscription;

    public TimeProcessor() {

    }

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


    @Override
    public void onNext(LocalDate item) {
        DayOfWeek dayOfWeek = item.getDayOfWeek();
        emit(dayOfWeek);
        subscription.request(1);
    }



    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
    // -------   Next assembly line ,

    private void emit(DayOfWeek dayOfWeek) {
        subscribe(daySubscriber);
        daySubscription.emit(dayOfWeek);
    }

    DaySubscription daySubscription;
    @Override
    public void subscribe(Flow.Subscriber subscriber) {
        if (daySubscription == null) {
            return;
        }
        DaySubscription subscription = new DaySubscription((DaySubscriber) subscriber);
        this.daySubscription = subscription;
        daySubscriber.onSubscribe(subscription);
    }

    DaySubscriber daySubscriber;
    public TimeProcessor downStream(DaySubscriber daySubscriber) {
        this.daySubscriber = daySubscriber;
        return this;
    }

}

DaySubscription

Downstream intermediaries

demo In order to save trouble, it is written directly to the publisher's constructor ; But the intermediate state adds a method to send data ; After the data is transmitted, the mediation will call the subscriber to consume the message

public class DaySubscription implements Flow.Subscription {


    private final DaySubscriber dayProcessor;
    private DayOfWeek dayOfWeek;

    public DaySubscription(DaySubscriber daySubscriber) {
        this.dayProcessor = daySubscriber;
    }

    public void emit(DayOfWeek dayOfWeek) {
        this.dayOfWeek = dayOfWeek;
        request(1);
    }
    @Override
    public void request(long n) {
        dayProcessor.onNext(dayOfWeek);
    }

    @Override
    public void cancel() {

    }
}

DaySubscriber

Downstream subscribers

This place is for simplicity , Back pressure is not used ; Because as demo, No asynchrony is used . therefore onNext Just execute the logic of consumption data

public class DaySubscriber<T> implements Flow.Subscriber<DayOfWeek> {

    private Flow.Subscription daySubscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.daySubscription = subscription;
    }

    @Override
    public void onNext(DayOfWeek item) {
        int day =  switch (item) {
            case MONDAY -> 2;
            case TUESDAY, THURSDAY, WEDNESDAY -> 1;
            case FRIDAY -> 5;
            case SATURDAY -> 4;
            case SUNDAY -> 3;
        };
        System.out.printf(" The task begins with : %s ,  Will be in  %d  The execution starts after days   ", item, day);
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}

Main

Consider simplicity ; There is only one intermediate processor to be familiar with the basic logic of a response

Essentially, you need to specify the steps , The code is similar to the above flowchart

     public static void main(String[] args) {
        LinkedList<LocalDate> linkedList = new LinkedList<>();
        linkedList.add(LocalDate.now());
        TimePublisher timePublisher = new TimePublisher(linkedList);

        timePublisher.subscribe(new TimeProcessor().downStream(new DaySubscriber()));
    }