Combine – швейцарский нож iOS-разработчика. Или нет?

Привет! Меня зовут Антон, я iOS-разработчик в Банки.ру. Когда я только начинал изучать Combine, он казался для меня магией. Пара команд и вот у тебя уже есть какие-то данные. Чтобы Combine перестал оставаться черным ящиком давайте заглянем внутрь. Эта статья – мое виденье этого фреймворка.

Небольшая сводка: Combine – фреймворк для работы с асинхронными событиями в декларативном стиле. Он помогает разработчикам управлять потоками данных, избавляя от множества колбэков, ручного управления очередями и других сложностей, связанных с асинхронностью.

Большинство статей описывают 3 сущности: Publisher (издатель), Subscriber (подписчик) и Operator'ы. Но они умалчивают еще об одном игроке – Subscription. Именно подписка управляет всей "жизнью" цепочки: кто кому и когда передаёт данные, и когда всё заканчивается.

В центре внимания Combine: 

• Publisher (издатель) — посылает сигналы  

• Subscriber (подписчик) — подписывается на Publisher и реагирует на поступающие значения 

• Operator'ы — модифицируют, фильтруют или комбинируют значения между Publisher и Subscriber 

Publisher – протокол, который является источником данных

Если проводить аналогии, то это “Человек с микрофоном”, который готов сообщать новости (значения).

Выглядит он вот так:

public protocol Publisher { 
    associatedtype Output 
    associatedtype Failure: Error 
    func receive<S: Subscriber>(subscriber: S) where S.Input == Output, S.Failure == Failure 
} 

За что отвечает: 

• Генерирует значения или ошибки;

• Метод receive() принимает Subscriber и присоединяет указанного подписчика к данному Publisher, после чего данный Subscriber сможет получать значения от Publisher;

• Создаёт Subscription и связывает ее с Subscriber, c помощью метода sink, который возвращает подписку.

public func sink(
    receiveCompletion: @escaping (Subscribers.Completion<Self.Failure>) -> Void,
    receiveValue: @escaping (Self.Output) -> Void
) -> AnyCancellable

Параметры этого метода это два замыкания:

• receiveValue — вызывается при каждом новом значении от паблишера.

• receiveCompletion — вызывается, когда паблишер завершает работу (.finished или .failure).

Если не сохранить подписку, то значение которое отправил Publisher будет утеряно. 

У Publisher есть еще много методов, реализованных через расширения. Их выделяют в отдельную сущность Operator'ы – промежуточные обработчики данных.

Операторов можно рассматривать как “Фильтр в цепи” –  например, усилитель или эквалайзер между микрофоном и динамиком. Они помогают управлять потоком данных, маппить данные, обрабатывать ошибки и тд.

Subscription – контракт между Subscriber и Publisher

Используем аналогию: “Шнур между микрофоном и наушниками” — передаёт звук, но может быть отключён или ограничен.

public protocol Subscription: Cancellable {
    func request(_ demand: Subscribers.Demand)
}

Главные задачи Subscription: 

• Создаётся когда Subscriber подписывается на паблишер Publisher, отвечает за жизненный цикл этой связи. Передача данных от Publisher к Subscriber прервется, если Subscription уйдет из памяти;

• Управляет передачей значений;

• Контролирует объём запрошенных данных, и может быть отменена (через cancel()).

Subscription наследуется от Cancellable:

public protocol Cancellable {
    func cancel()
}

Любая подписка должна уметь отменять получение данных. Когда вызываешь cancel(), паблишер должен прекратить посылку значений подписчику и освободить ресурсы.

Метод request(_:)

func request(_ demand: Subscribers.Demand)

Это ключевой метод Combine для контроля потока данных. Он говорит паблишеру, сколько значений подписчик готов получить. Это указывается во входном параметре. 

Subscribers.Demand – это структура, описывающая сколько значений подписчик может принять. В качестве значений передаются:

public static let unlimited: Subscribers.Demand // подписчик готов принять сколько угодно значений
@inlinable public static func max(_ value: Int) -> Subscribers.Demand // готов принять максимум value значений
public static let none: Subscribers.Demand // Это эквивалентно max(0)

Это делает Combine "pull-based" системой – подписчик запрашивает значения, а не просто "получает по факту". 

Subscriber – получатель данных от Publisher

Его можно представить как “слушателя в зале”. Он может сказать: Прекрати, Продолжай, или Я готов к N сообщениям..

public protocol Subscriber: CustomCombineIdentifierConvertible { 
    associatedtype Input 
    associatedtype Failure: Error 
    func receive(subscription: Subscription) 
    func receive(_ input: Input) -> Subscribers.Demand 
    func receive(completion: Subscribers.Completion<Failure>) 
} 

Subscriber подписывается на Publisher и получает:

1. Subscription (для управления подпиской и запросом данных);

2. Значения (Output);

3. Завершение потока (Completion).

То есть Subscriber описывает как именно обрабатываются события от паблишера.

Связанные типы подписчика:

• associatedtype Input // тип значений, которые получает подписчик. (должен совпадать с Publisher.Output)

• associatedtype Failure: Error // тип ошибки, которую может выдать паблишер. (должен совпадать с Publisher.Failure)

Описание методов подписчика:

Давайте подытожим:

Теперь – как все это работает на примере кастомной цепочки

// MARK: - Custom Publisher
struct MyPublisher: Publisher {
    typealias Output = Int
    typealias Failure = Never
    
    func receive<S>(subscriber: S)
    where S : Subscriber, MyPublisher.Failure == S.Failure, MyPublisher.Output == S.Input {
        // Создаём подписку и передаём её подписчику
        let subscription = MySubscription(subscriber: subscriber)
        subscriber.receive(subscription: subscription)
    }
}

// MARK: - Custom Subscription
final class MySubscription<S: Subscriber>: Subscription where S.Input == Int {
    private var subscriber: S?
    private var current = 1
    private let max = 5
    
    init(subscriber: S) {
        self.subscriber = subscriber
    }
    
    func request(_ demand: Subscribers.Demand) {
        // Если подписчик запросил данные
        guard demand > .none else {
            return
        }
        
        // Отправляем несколько значений
        while current <= max {
            _ = subscriber?.receive(current)
            current += 1
        }
        
        // Завершаем поток
        subscriber?.receive(completion: .finished)
    }
    
    func cancel() {
        print("Подписка отменена")
        subscriber = nil
    }
}
// MARK: - Custom Subscriber
final class MySubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = Never
    
    func receive(subscription: Subscription) {
        print("Подписка получена")
        // Запрашиваем все значения
        subscription.request(.unlimited)
    }
    
    func receive(_ input: Int) -> Subscribers.Demand {
        print("Получено значение:", input)
        // Можно вернуть .none (не запрашивать дополнительно)
        return .none
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print("Завершено:", completion)
    }
}


// MARK: - Пример использования
do {
let publisher = MyPublisher()
let subscriber = MySubscriber()
publisher.subscribe(subscriber)
 }

Вывод в консоль:

• Подписка получена

• Получено значение: 1

• Получено значение: 2

• Получено значение: 3

• Получено значение: 4

• Получено значение: 5

• Завершено: finished

Пошаговое объяснение

1. publisher.subscribe(subscriber)
   – Паблишер получает подписчика.
   – Создаёт MySubscription и вызывает subscriber.receive(subscription:).

2. MySubscriber.receive(subscription:)
   – Сохраняет ссылку на подписку.
   – Запрашивает .unlimited (все возможные значения).

3. MySubscription.request(_:)
   – Отправляет значения 1…5 в subscriber.receive(_:).
   – После этого вызывает receive(completion: .finished).

4. Поток завершается.

Как это выглядит на схеме:

В этом примере мы последовательно while сurrent <= max запросили 5 значений, при этом MySubscriber никак не ограничивает поток значений, ведь subscription.request(.unlimited), давайте это исправим.

Для MySubscriber

Метод:

func receive(subscription: Subscription) {
        print("Подписка получена")
        // Запрашиваем все значения
        subscription.request(.unlimited)
    }

Поменяем на:

func receive(subscription: Subscription) {
        print("Подписка получена")
        // Запрашиваем все значения
        subscription.request(.max(3))
    }

Для MySubscription

Метод:

  func request(_ demand: Subscribers.Demand) {
        // Если подписчик запросил данные
        guard demand > .none else {
            return
        }
        
        // Отправляем несколько значений
        while current <= max {
            _ = subscriber?.receive(current)
            current += 1
        }
        
        // Завершаем поток
        subscriber?.receive(completion: .finished)
    }

Поменяем на:

func request(_ demand: Subscribers.Demand) {
        // Если подписчик запросил данные
        guard
            demand > .none else {
            return
        }
        
        // Отправляем несколько значений
        while current <= demand.max ?? max {
            _ = subscriber?.receive(current)
            current += 1
            
        }
        
        // Завершаем поток
        subscriber?.receive(completion: .finished)
    }

Вывод в консоль:

• Подписка получена

• Получено значение: 1

• Получено значение: 2

• Получено значение: 3

• Завершено: finished

Теперь мы ограничили вызванные значения до 3.

Что будет если мы дважды вызовем publisher.subscribe(subscriber)?

publisher.subscribe(subscriber)
publisher.subscribe(subscriber)

У нас дважды выводится:

• Подписка получена

• Получено значение: 1

• Получено значение: 2

• Получено значение: 3

• Завершено: finished

• Подписка получена

• Получено значение: 1

• Получено значение: 2

• Получено значение: 3

• Завершено: finished

Каждый вызов subscribe создаёт новый экземпляр MySubscription и новую независимую цепочку. Это можно проверить, если добавить в MySubscription:

deinit {
        print("