Задание 2 – Пользовательская реализация RxJava

Этот проект представляет собой реализацию библиотеки реактивного программирования, аналогичную RxJava, в соответствии с техническим заданием. Реализация включает все требуемые компоненты: базовые интерфейсы, операторы преобразования данных, управление потоками выполнения, обработку ошибок и тестирование.

Реализованный функционал

1. Базовые компоненты реактивного программирования

В проекте реализованы все необходимые базовые компоненты реактивного программирования. Интерфейс Observer<T> предоставляет методы для обработки элементов потока (onNext), ошибок (onError) и завершения (onComplete). Класс Observable<T> реализует основной паттерн реактивного программирования с поддержкой подписки и статическим методом create() для создания потоков данных. Код написан с учетом принципов чистоты и структурированности.

2. Операторы преобразования данных

Реализованы основные операторы преобразования данных: map для трансформации элементов потока и filter для фильтрации на основе предиката. Каждый оператор протестирован и демонстрирует корректную работу с различными типами данных. Операторы поддерживают цепочки преобразований и корректно обрабатывают ошибки.

3. Управление потоками выполнения

Разработана система управления потоками выполнения на основе интерфейса Scheduler. Реализованы три типа планировщиков:

• IOThreadScheduler на основе CachedThreadPool для I/O операций
• ComputationScheduler на основе FixedThreadPool для вычислений
• SingleThreadScheduler для последовательного выполнения задач

Методы subscribeOn и observeOn обеспечивают гибкое управление потоками выполнения, позволяя выполнять подписку и обработку элементов в разных потоках.

4. Дополнительные операторы и управление подписками

Реализован оператор flatMap для преобразования элементов в новые потоки данных с последующим их объединением. Интерфейс Disposable предоставляет механизм отмены подписок и проверки их состояния. Система обработки ошибок позволяет корректно передавать и обрабатывать исключения через метод onError.

5. Тестирование

Разработан комплекс тестов, охватывающий все аспекты реализации:

• Модульные тесты для базовых компонентов
• Тесты обработки ошибок и граничных случаев
• Тесты работы планировщиков в многопоточной среде
• Тесты операторов преобразования и их комбинаций
• Тесты сложных сценариев использования

Архитектура

Проект реализует паттерн "Наблюдатель" (Observer pattern) и следует принципам реактивного программирования. Основные компоненты:

Основные интерфейсы

1. Observer

   public interface Observer<T> {
       void onNext(T item);
       void onError(Throwable t);
       void onComplete();
   }
   

2. Observable

   • Основной класс для работы с потоками данных
   • Поддерживает создание через create()
   • Реализует цепочку операторов

3. Scheduler

   public interface Scheduler {
       void execute(Runnable task);
   }
   

4. Disposable

   public interface Disposable {
       void dispose();
       boolean isDisposed();
   }
   

Операторы преобразования

Map

public <R> Observable<R> map(Function<T, R> mapper) {
    return new Observable<>(observer -> {
        subscribe(
            item -> observer.onNext(mapper.apply(item)),
            observer::onError,
            observer::onComplete
        );
    });
}

Filter

public Observable<T> filter(Predicate<T> predicate) {
    return new Observable<>(observer -> {
        subscribe(
            item -> {
                if (predicate.test(item)) {
                    observer.onNext(item);
                }
            },
            observer::onError,
            observer::onComplete
        );
    });
}

FlatMap

public <R> Observable<R> flatMap(Function<T, Observable<R>> mapper) {
    return new Observable<>(observer -> {
        subscribe(
            item -> mapper.apply(item).subscribe(observer),
            observer::onError,
            observer::onComplete
        );
    });
}

Управление потоками выполнения

Типы планировщиков

1. IOThreadScheduler

   • Использует CachedThreadPool
   • Оптимизирован для I/O операций
   • Создает потоки по требованию
   • Переиспользует неактивные потоки

2. ComputationScheduler

   • Использует FixedThreadPool
   • Размер пула = количество процессоров
   • Оптимизирован для вычислений
   • Фиксированное количество потоков

3. SingleThreadScheduler

   • Использует один поток
   • Последовательное выполнение
   • Гарантированный порядок операций

Использование планировщиков

observable
    .subscribeOn(new IOThreadScheduler())     // Подписка в IO потоке
    .observeOn(new ComputationScheduler())    // Обработка в вычислительном потоке
    .subscribe(...);

Обработка ошибок

Реализована комплексная система обработки ошибок:

1. Передача ошибок через onError()
2. Распространение ошибок по цепочке операторов
3. Очистка ресурсов при ошибках
4. Обработка прерывания потоков

Пример:

observable
    .map(x -> {
        if (x < 0) throw new IllegalArgumentException("Отрицательное значение");
        return x * 2;
    })
    .subscribe(
        item -> System.out.println("Получено: " + item),
        error -> System.out.println("Ошибка: " + error.getMessage()),
        () -> System.out.println("Завершено")
    );

Тестирование

Проект включает комплексные тесты:

Категории тестов

1. ObservableTest

   • Базовая эмиссия
   • Обработка ошибок
   • Операторы преобразования
   • Disposable

2. ConcurrencyTest

   • Работа с планировщиками
   • Многопоточное выполнение
   • Управление подписками

3. OperatorChainTest

   • Комбинации операторов
   • Распространение ошибок
   • Управление ресурсами

4. AdvancedScenariosTest

   • Сложные сценарии использования
   • Граничные случаи
   • Производительность

Примеры использования

Базовое использование

Observable.create(observer -> {
    observer.onNext(1);
    observer.onNext(2);
    observer.onComplete();
})
.subscribe(
    item -> System.out.println("Получено: " + item),
    error -> System.out.println("Ошибка: " + error.getMessage()),
    () -> System.out.println("Завершено")
);

Сложная цепочка операторов

Observable.create(observer -> {
    for (int i = 0; i < 10; i++) {
        observer.onNext(i);
    }
    observer.onComplete();
})
.filter(x -> x % 2 == 0)
.map(x -> x * 2)
.flatMap(x -> Observable.create(observer -> {
    observer.onNext(x * 10);
    observer.onNext(x * 20);
    observer.onComplete();
}))
.subscribeOn(new IOThreadScheduler())
.observeOn(new ComputationScheduler())
.subscribe(
    item -> System.out.println("Результат: " + item),
    error -> System.out.println("Ошибка: " + error.getMessage()),
    () -> System.out.println("Завершено")
);

Управление подпиской

Disposable subscription = Observable.create(observer -> {
    int i = 0;
    while (!Thread.currentThread().isInterrupted()) {
        observer.onNext(i++);
        Thread.sleep(100);
    }
})
.subscribe(
    item -> System.out.println("Получено: " + item),
    error -> System.out.println("Ошибка: " + error.getMessage()),
    () -> System.out.println("Завершено")
);

// Отмена подписки через 500мс
Thread.sleep(500);
subscription.dispose();