Привет, Хабр!
Combine — это фреймворк, представленный Apple в 2019 году, предназначенный для работы с асинхронными потоками данных на платформе Swift. Основная задача Combine заключается в том, чтобы упростить и унифицировать обработку асинхронных событий, таких как сетевые запросы, таймеры, уведомления и пользовательские действия.
Фреймворк является частью экосистемы Swift и доступен начиная с iOS 13, macOS 10.15, watchOS 6.0 и tvOS 13. Если вы работаете с более ранними версиями, обновление Xcode до версии 11 или выше автоматом предоставит вам доступ к Combine.
В этой статье мы рассмотрим основы этого замечательного фреймворка.
После установки Xcode вы сможете использовать Combine в своих проектах, импортировав его следующим образом:
import CombineСоздание и комбинирование потоков данных с помощью Combine
В Combine всё начинается с Publisher — сущности, которая эмитирует значения с течением времени. Это может быть что угодно: массив чисел, результат сетевого запроса, или события пользовательского интерфейса.
Простейший пример Publisher — это массив, преобразованный в Publisher с помощью .publisher:
import Combine
let numbers = [1, 2, 3, 4, 5].publisher
numbers.sink { value in
print("Received value: \(value)")
}Массив numbers превращается в Publisher, который выдает каждое число последовательно. Метод sink — это Subscriber, который "подписывается" на Publisher и выводит каждый полученный элемент.
Если хочется создать собственный Publisher, можно использовать PassthroughSubject или CurrentValueSubject:
import Combine
let subject = PassthroughSubject<String, Never>()
subject.sink { value in
print("Received value: \(value)")
}
subject.send("Hello")
subject.send("Combine")Здесь PassthroughSubject позволяет вручную выдавать значения, используя метод send, можно контролировать, когда и какие данные будут отправлены подписчикам.
Когда дело доходит до объединения данных из разных источников, есть несколько операторов: merge, combineLatest и zip.
mergeобъединяет несколько Publisher-ов в один, эмитируя значения по мере их поступления:
import Combine
let publisher1 = PassthroughSubject<Int, Never>()
let publisher2 = PassthroughSubject<Int, Never>()
let merged = publisher1.merge(with: publisher2)
merged.sink { value in
print("Merged value: \(value)")
}
publisher1.send(1)
publisher2.send(2)
publisher1.send(3)Результат работы merge будет последовательность: 1, 2, 3, т.е. значения от обоих Publisher-ов объединяются в один поток.
combineLatestвыдает пары значений от обоих Publisher-ов, когда оба источника данных предоставили хотя бы одно значение:
import Combine
let publisherA = PassthroughSubject<String, Never>()
let publisherB = PassthroughSubject<String, Never>()
let combined = publisherA.combineLatest(publisherB)
combined.sink { value in
print("Combined value: \(value)")
}
publisherA.send("Hello")
publisherB.send("World")
// Вывод: "Combined value: ("Hello", "World")"
publisherA.send("Hi")
// Вывод: "Combined value: ("Hi", "World")"combineLatest ждет данные от обоих Publisher-ов и затем выдает их в виде пары, каждый раз, когда поступает новое значение от одного из них.
zipработает подобноcombineLatest, но выдает пары только тогда, когда обе стороны готовы передать по одному значению:
import Combine
let pub1 = PassthroughSubject<Int, Never>()
let pub2 = PassthroughSubject<String, Never>()
let zipped = pub1.zip(pub2)
zipped.sink { value in
print("Zipped value: \(value)")
}
pub1.send(1)
pub2.send("One")
// Вывод: "Zipped value: (1, "One")"
pub1.send(2)
pub2.send("Two")
// Вывод: "Zipped value: (2, "Two")"Этот оператор хорош, когда необходимо синхронизировать данные из разных источников.
Допустим, есть приложение, которое принимает пользовательский ввод и результат предсказания модели, и вы хочется комбинировать эти данные для отображения их в UI:
import Combine
// Создаем два Publisher-а: один для ввода пользователя, другой для предсказаний
let userInput = PassthroughSubject<String, Never>()
let prediction = PassthroughSubject<String, Never>()
// Комбинируем данные, используя combineLatest
let combinedData = userInput.combineLatest(prediction)
combinedData.sink { user, predicted in
print("User input: \(user), Model prediction: \(predicted)")
}
// Эмитируем значения
userInput.send("User says: Hello")
prediction.send("Model predicts: Hi")
// Вывод: "User input: User says: Hello, Model prediction: Hi"combineLatest позволяет связать ввод пользователя с результатами работы ML-модели.
Трансформация данных в потоках
Оператор map является основным инструментом для трансформации данных в Combine. Он позволяет взять данные, выдаваемые Publisher-ом, и преобразовать их в новую форму, изменяя каждый элемент потока.
Представим, что есть поток чисел, и хочется умножить каждое из них на 2. Это легко реализуется с помощью map:
import Combine
let numbers = [1, 2, 3, 4, 5].publisher
let multipliedNumbers = numbers.map { $0 * 2 }
multipliedNumbers.sink { value in
print("Transformed value: \(value)")
}Результатом выполнения этого кода будет вывод:
Transformed value: 2
Transformed value: 4
Transformed value: 6
Transformed value: 8
Transformed value: 10Оператор map подходит для предобработки данных перед их передачей в ML модель. Например, если нужно нормализовать входные данные перед подачей их в модель:
let rawInputs = [0.5, 0.75, 1.0, 1.25].publisher
let normalizedInputs = rawInputs.map { $0 / 2.0 }
normalizedInputs.sink { value in
print("Normalized value: \(value)")
}Оператор filter позволяет отфильтровывать значения, пропуская только те, которые удовлетворяют определенным условиям.
Рассмотрим пример, где нужно оставить в потоке только четные числа:
let numbers = [1, 2, 3, 4, 5, 6].publisher
let evenNumbers = numbers.filter { $0 % 2 == 0 }
evenNumbers.sink { value in
print("Filtered value: \(value)")
}Вывод:
Filtered value: 2
Filtered value: 4
Filtered value: 6В контексте ML, filter может быть полезен для удаления выбросов или данных, не соответствующих определенным критериям. Например, если нужно исключить из входных данных все значения, превышающие допустимый предел:
let rawInputs = [0.5, 2.5, 1.0, 3.5].publisher
let validInputs = rawInputs.filter { $0 <= 2.0 }
validInputs.sink { value in
print("Valid input: \(value)")
}flatMap — это оператор, который раскрывает новый Publisher на основе каждого значения, выдаваемое исходным Publisher-ом.
Рассмотрим пример с сетевым запросом, который возвращает данные на основе пользовательского ввода:
import Foundation
let userInput = PassthroughSubject<String, Never>()
let searchResults = userInput.flatMap { query in
URLSession.shared.dataTaskPublisher(for: URL(string: "https://api.example.com/search?q=\(query)")!)
.map { $0.data }
.decode(type: [String].self, decoder: JSONDecoder())
.catch { _ in Just([]) } // Обрабатываем возможные ошибки
}
searchResults.sink { results in
print("Search results: \(results)")
}
userInput.send("apple")flatMap используется для запуска сетевого запроса на основе пользовательского ввода. Каждый раз, когда пользователь вводит новый запрос, flatMap раскрывает новый Publisher, который затем обрабатывает результаты и возвращает их.
Примеры использования
Предположим, есть поток сырых данных, которые нужно нормализовать и отфильтровать перед подачей в модель:
let rawInputs = [1.5, 0.9, 2.5, 3.7, 1.8].publisher
let processedInputs = rawInputs
.filter { $0 < 3.0 } // Оставляем только допустимые значения
.map { $0 / 3.0 } // Нормализуем данные
processedInputs.sink { value in
print("Processed input: \(value)")
}После того как модель выдала свои предсказания, возможно, потребуется постобработка данных, чтобы они стали пригодными для отображения в UI или дальнейшей аналитики:
let modelPredictions = [0.2, 0.8, 1.0, 0.4].publisher
let filteredPredictions = modelPredictions
.filter { $0 > 0.5 } // Оставляем только уверенные предсказания
.map { $0 * 100.0 } // Преобразуем в проценты
filteredPredictions.sink { value in
print("Filtered prediction: \(value)%")
}Управление ошибками и отложенная обработка данных
В Combine ошибки обрабатываются с помощью операторов catch и retry.
Оператор catch позволяет перехватывать ошибки, возникающие в потоке данных, и предоставлять альтернативный Publisher для продолжения работы.
Пример:
import Combine
struct MyError: Error {}
let faultyPublisher = Fail<Int, MyError>(error: MyError())
let safePublisher = faultyPublisher
.catch { _ in Just(0) } // В случае ошибки возвращаем значение 0
.sink { value in
print("Received value: \(value)")
}
// Вывод: "Received value: 0"ИспользуемFail для эмуляции ошибки, а затем обработали ее с помощью catch, который возвращает безопасное значение.
Оператор retry позволяет автоматически повторять выполнение потока данных, если в нем произошла ошибка.
Пример:
import Combine
var attemptCount = 0
let retryingPublisher = Deferred {
attemptCount += 1
return attemptCount < 3 ? Fail(error: URLError(.badServerResponse)) : Just("Success")
}
.retry(3) // Повторяем до трех раз
.sink(
receiveCompletion: { completion in
switch completion {
case .finished:
print("Completed successfully")
case .failure(let error):
print("Failed with error: \(error)")
}
},
receiveValue: { value in
print("Received value: \(value)")
}
)
// Вывод: "Received value: Success"Здесь retry пытается выполнить поток данных до трех раз. На третьей попытке поток завершается успешно.
Работа с асинхронными данными часто требует управления частотой обработки событий, особенно если поток данных поступает быстро и может перегрузить систему. Combine предоставляет два оператора для этой задачи: debounce и throttle.
debounce — это оператор, который откладывает обработку событий до тех пор, пока не пройдет определенное время без поступления новых данных.
Пример:
import Combine
import Foundation
let searchTextPublisher = PassthroughSubject<String, Never>()
let debouncedSearch = searchTextPublisher
.debounce(for: .milliseconds(500), scheduler: RunLoop.main)
.sink { value in
print("Search query: \(value)")
}
searchTextPublisher.send("S")
searchTextPublisher.send("Sw")
searchTextPublisher.send("Swi")
searchTextPublisher.send("Swif")
searchTextPublisher.send("Swift")
// Через 500 мс после последнего ввода: "Search query: Swift"Здесь debounce откладывает отправку данных до тех пор, пока пользователь не закончит ввод текста, избегая излишних запросов.
throttle позволяет пропускать события только через определенные интервалы времени, ограничивая частоту их обработки.
Пример:
import Combine
import Foundation
let eventPublisher = PassthroughSubject<Void, Never>()
let throttledEvents = eventPublisher
.throttle(for: .seconds(1), scheduler: RunLoop.main, latest: true)
.sink {
print("Event received")
}
for _ in 1...5 {
eventPublisher.send()
Thread.sleep(forTimeInterval: 0.3) // Эмуляция частого срабатывания события
}
// Вывод: "Event received" будет напечатано только один раз в секундуthrottle в этом примере пропускает только одно событие в секунду, игнорируя остальные, что позволяет эффективно управлять нагрузкой.
Рассмотрим сценарий, где идет работа с потоком данных из нескольких источников, например, с сенсоров и сервера, и нужно управлять ошибками и задержками:
import Combine
import Foundation
enum SensorError: Error {
case sensorFailure
}
let sensorDataPublisher = PassthroughSubject<Int, SensorError>()
let serverDataPublisher = PassthroughSubject<Int, URLError>()
let combinedPublisher = Publishers.CombineLatest(sensorDataPublisher, serverDataPublisher)
.retry(2) // Пытаемся заново в случае ошибки
.debounce(for: .milliseconds(300), scheduler: RunLoop.main) // Избегаем частой обработки данных
.catch { _ in Just((0, 0)) } // В случае ошибки возвращаем безопасное значение
.sink { sensorData, serverData in
print("Sensor: \(sensorData), Server: \(serverData)")
}
sensorDataPublisher.send(100)
serverDataPublisher.send(200)
// Эмулируем ошибку сенсора
sensorDataPublisher.send(completion: .failure(.sensorFailure))
serverDataPublisher.send(300)Используем CombineLatest для объединения данных от сенсоров и сервера, обрабатываем ошибки с помощью retry, избегаем частой обработки данных с помощью debounce, и, наконец, безопасно обрабатываем ошибки с помощью catch.
Подробнее с библиотекой Combine можно ознакомиться здесь.
Актуальна тема подготовки данных в Pandas? Приходите на открытый урок 19 августа, который пройдет в рамках специализации Machine Learning.
В результате вебинара вы узнаете о методах подготовки данных и научитесь чистить данные при помощи библиотеки Pandas. Записывайтесь на урок по ссылке.
