import reactor.core.publisher.Flux; // Flux реализует Publisher

Flux<Integer> publisher = Flux.range(1, 5); // Издатель: поток от 1 до 5

publisher.subscribe(new Subscriber<Integer>() {
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
        s.request(1); // Запрашиваем первый элемент
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Получено: " + item);
        subscription.request(1); // Запрашиваем следующий
    }

    @Override
    public void onError(Throwable t) {
        System.err.println("Ошибка: " + t);
    }

    @Override
    public void onComplete() {
        System.out.println("Завершено");
    }
});

import java.util.HashMap;

public class Task240925 {
    public static void main(String[] args) {
        HashMap<String, Integer> map = new HashMap<>();
        map.put("a", 1);
        map.put(new String("a"), 2);
        
        System.out.println(map.size());
        System.out.println(map.get("a"));
    }
}

public class Animal {
    protected String name;

    public Animal(String name) {
        this.name = name;
    }

    public void makeSound() {
        System.out.println(name + " издает звук.");
    }
}

public class Dog extends Animal {
    public Dog(String name) {
        super(name);
    }

    @Override
    public void makeSound() {
        System.out.println(name + " лает: Гав!");
    }
}

public class Engine {
    private int horsepower;

    public Engine(int horsepower) {
        this.horsepower = horsepower;
    }

    public void start() {
        System.out.println("Двигатель с " + horsepower + " л.с. запущен.");
    }
}

public class Car {
    private String model;
    private Engine engine;  // Композиция: Car "имеет" Engine

    public Car(String model, Engine engine) {
        this.model = model;
        this.engine = engine;
    }

    public void drive() {
        engine.start();
        System.out.println(model + " едет.");
    }
}

public class Main {
    public static void main(String[] args) {
        Engine engine = new Engine(200);
        Car car = new Car("Toyota", engine);
        car.drive();  // Двигатель с 200 л.с. запущен. Toyota едет.
    }
}

public class Vehicle {
    public void move() {
        System.out.println("Движется с помощью двигателя.");
    }
}

public class Car extends Vehicle {
    // Проблема: А если электромобиль или велосипед?
}

Композиция (решение):
public interface Movable {
    void move();
}

public class EngineMovable implements Movable {
    @Override
    public void move() {
        System.out.println("Движется с помощью двигателя.");
    }
}

public class Car {
    private Movable movement;

    public Car(Movable movement) {
        this.movement = movement;
    }

    public void drive() {
        movement.move();
    }
}

Теперь Car может использовать любой Movable (двигатель, электромотор), без привязки к иерархии.

import java.util.function.Function;

public class Task250925 {
    public static void main(String[] args) {
        Function<Integer, Integer> plus5 = x -> x + 5;
        Function<Integer, Integer> times2 = x -> x * 2;
        
        Function<Integer, Integer> composed = plus5.compose(times2);
        System.out.println(composed.apply(3));
    }
}

import reactor.core.publisher.Flux;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

Flux<Integer> fastPublisher = Flux.range(1, 1000); // Быстрый источник: 1000 элементов

fastPublisher.subscribe(new Subscriber<Integer>() {
    private Subscription sub;

    @Override
    public void onSubscribe(Subscription s) {
        sub = s;
        sub.request(5); // Сначала запрашиваем 5 элементов
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Обработано: " + item);
        // Симулируем медленную обработку: Thread.sleep(100); (но в реальности избегайте блокировок!)
        sub.request(1); // После каждого — запрашиваем следующий
    }

    @Override
    public void onError(Throwable t) { /* обработка */ }
    @Override
    public void onComplete() { System.out.println("Готово"); }
});

Здесь подписчик контролирует темп: запросил 5 — получил 5, обработал — запросил ещё. Если не request() — поток остановится. Это асинхронно: издатель не блокируется, а ждёт сигналов.

Flux.interval(Duration.ofMillis(1)) // Бесконечный поток: элемент каждую мс
    .onBackpressureDrop(dropped -> System.out.println("Сброшено: " + dropped)) // Стратегия: сбрасываем лишнее
    .subscribe(item -> {
        try { Thread.sleep(100); } catch (InterruptedException e) {} // Медленный подписчик
        System.out.println("Обработано: " + item);
    });

Здесь быстрый Flux "замедляется" под подписчика: лишние элементы сбрасываются, система не падает.

Ещё опция — cancel(): подписчик может полностью отменить подписку, если данных слишком много или не нужно.