Github.com/agelessman/…
The Publishers are at the top of the pipline, and their primary role is to send data. This article introduces the Publishers in Combine.
[TOC]
Just
Just can be regarded as the simplest publisher. It sends data in the same way as pass-through data. The most commonly used scenario is with flatMap:
class JustViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
struct Student: Decodable {
let name: String
}
let json = "" "[{" name ":" xiao Ming "}, {" name ":" little red "}, {" name ":" li lei "}] "" "
init(a) {
let publisher = PassthroughSubject<String.Never>()
cancellable = publisher
.flatMap { value in
Just(value.data(using: .utf8)!)
.decode(type: [Student].self, decoder: JSONDecoder())
.catch { _ in
Just([Student(name: Anonymous)])
}
}
.sink(receiveCompletion: { _ in
print("It's over.")
}, receiveValue: { someValue in
print(someValue)
})
publisher.send(json)
}
}
Copy the codeThe output is as follows:
[MCMarbleDiagramSwiftUI.JustViewObservableObject.Student(name: "Xiao Ming"),
MCMarbleDiagramSwiftUI.JustViewObservableObject.Student(name: "Little red"),
MCMarbleDiagramSwiftUI.JustViewObservableObject.Student(name: "Li lei")]
Copy the codeBecause the flatMap closure requires that the return value be a Publisher, it is appropriate to use Just in the above code, which maps JSON data to an array of models.
Just is also used in the catch above to return a default value when an error occurs. It is worth noting that the catch also requires a publsiher return.
Future
final public class Future<Output.Failure> : Publisher where Failure : Error {
/// A type that represents a closure to invoke in the future, when an element or error is available.
///
/// The promise closure receives one parameter: a `Result` that contains either a single element published by a ``Future``, or an error.
public typealias Promise = (Result<Output.Failure- > >)Void
/// Creates a publisher that invokes a promise closure when the publisher emits an element.
///
/// - Parameter attemptToFulfill: A ``Future/Promise`` that the publisher invokes when the publisher emits an element or terminates with an error.
public init(_ attemptToFulfill: @escaping (@escaping Future<Output.Failure>.Promise) - >Void)
/// Attaches the specified subscriber to this publisher.
///
/// Implementations of ``Publisher`` must implement this method.
///
/// The provided implementation of ``Publisher/subscribe(_:)-4u8kn``calls this method.
///
/// - Parameter subscriber: The subscriber to attach to this ``Publisher``, after which it can receive values.
final public func receive<S> (subscriber: S) where Output = = S.Input.Failure = = S.Failure.S : Subscriber
}
Copy the codeFuture is a publisher that specializes in processing asynchronous functions. After analyzing the above code, we find that it is initialized with a closure whose return value is a Result
,>
Let’s use an example:
class FutureViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
init(a) {
cancellable = Future<Bool.Error> { promise in
/// simulate the asynchronous process
DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
return promise(.success(true))
}
}
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)}}}Copy the codeIn the above code, the use of DispatchQueue. Main. AsyncAfter simulates an asynchronous process, in essence, the asynchronous process can be any asynchronous process, such as the normal development of network requests, or of some systems API and so on.
It’s a good practice to add an existing asynchronous procedure to a Pipline using a Future, but note that the Future is invoked immediately after creation, rather than waiting for a request from a subscriber. To verify this, we can modify the code:
cancellable = Future<Bool.Error> { promise in
print("diaoyuong")
/// simulate the asynchronous process
DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
return promise(.success(true))
}
}
.print()
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})Copy the codeSimply add a print to the closure, which looks like this:
diaoyuong
receive subscription: (Future)
request unlimited
receive value: (true)
true
receive finished
finished
Copy the codeAs you can see, the closure was actually called before the request was received. To resolve this problem, wrap a Deferred around the Future:
cancellable = Deferred {
return Future<Bool.Error> { promise in
/// simulate the asynchronous process
DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
return promise(.success(true))
}
}
}
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})Copy the codeEmpty
/// A publisher that never publishes any values, and optionally finishes immediately.
///
/// You can create a "Never" publisher -- one which Never pushes values and Never completes or fails -- with the initiator `Empty(completeImmediately: false)`.
@available(macOS 10.15.iOS 13.0.tvOS 13.0.watchOS 6.0.*)
public struct Empty<Output.Failure> : Publisher.Equatable where Failure : Error {
/// Creates an empty publisher.
///
/// - Parameter completeImmediately: A Boolean value that indicates whether the publisher should immediately finish.
public init(completeImmediately: Bool = true)
/// Creates an empty publisher with the given completion behavior and output and failure types.
///
/// Use this initializer to connect the empty publisher to subscribers or other publishers that have specific output and failure types.
///
/// - Parameters:
/// - completeImmediately: A Boolean value that indicates whether the publisher should immediately finish.
/// - outputType: The output type exposed by this publisher.
/// - failureType: The failure type exposed by this publisher.
public init(completeImmediately: Bool = true.outputType: Output.Type.failureType: Failure.Type)
/// A Boolean value that indicates whether the publisher immediately sends a completion.
///
/// If `true`, the publisher finishes immediately after sending a subscription to the subscriber. If `false`, it never completes.
public let completeImmediately: Bool
}
Copy the codeEmpty does not send any data and has the option to terminate the Pipline immediately, which by default terminates the Piline immediately after the Publisher is created.
We can use this feature to terminate the Pipline as soon as we hear an error.
class EmptyPublisherViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
init(a) {
enum MyError: Error {
case custom
}
let publisher = PassthroughSubject<Int.MyError>()
cancellable = publisher
.catch { _ in
Empty<Int.MyError>()
}
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)
})
publisher.send(1)
publisher.send(completion: Subscribers.Completion.failure(MyError.custom))
}
}
Copy the codeThe above code uses a catch to listen for errors and, if an error occurs, sends an Emptypublisher to terminate the Pipline using its property to terminate the pipline immediately. ReplaceError (with: 0) The difference is that the latter does not terminate pipline and replaces errors with default values.
Fail
class FailViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
init(a) {
enum MyError: Error {
case custom
}
cancellable = Fail<Int.MyError>(error: MyError.custom)
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)}}}Copy the codeBoth Fail and Empty can kill pipline immediately, but Fail doesn’t use many scenarios, so I won’t go into more here.
Publishers.Sequence
class SequenceViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
init(a) {
let list: Array = [1.2.3]
let set: Set = ["1"."2"."3"]
let dict: Dictionary = ["name": "Zhang"."age": "20"]
let str: String = "How are you?"
cancellable = list.publisher
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)}}}Copy the codePublishers.Sequence solves data flow problems like this collection data type, and any object that implements the Sequence protocol can automatically create a publisher by calling.publisher and then using all the features in Combine.
In real development, the following four types are most commonly used:
ArraySetDictionayString
Let’s look at the String example:
cancellable = "How are you?".publisher
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})Copy the codePrint result:
How are you finishedCopy the codeIsn’t that interesting? String is essentially a collection of characters, using the above code, easy to achieve character transfer.
Record
/// A publisher that allows for recording a series of inputs and a completion, for later playback to each subscriber.
@available(macOS 10.15.iOS 13.0.tvOS 13.0.watchOS 6.0.*)
public struct Record<Output.Failure> : Publisher where Failure : Error {
/// Creates a publisher to interactively record a series of outputs and a completion.
///
/// - Parameter record: A recording instance that can be retrieved after completion to create new record publishers to replay the recording.
public init(record: (inout Record<Output.Failure>.Recording) - >Void)
/// Creates a record publisher from an existing recording.
///
/// - Parameter recording: A previously-recorded recording of published elements and a completion.
public init(recording: Record<Output.Failure>.Recording)
/// Creates a record publisher to publish the provided elements, followed by the provided completion value.
///
/// - Parameters:
/// - output: An array of output elements to publish.
/// - completion: The completion value with which to end publishing.
public init(output: [Output].completion: Subscribers.Completion<Failure>)
}
Copy the codeRecord is actually a very powerful and useful publisher, powerful in that it can be encoded and decoded, just like its name, can be saved, parsed, and passed. The idea is to save the data and send it when the subscription is received.
As you can see from the code above, he has three initialization methods:
- use
record: (inout Record<Output, Failure>.Recording) -> VoidClosure initialization
let recordPublisher = Record<String.MyCustomError> { recording in
recording.receive("You")
recording.receive("Good")
recording.receive("吗")
recording.receive(completion: Subscribers.Completion.finished)
}
Copy the code- use
recording: Record<Output, Failure>.RecordingInitialize, which is passing in aRecordingType instance
var recording = Record<String.MyCustomError>.Recording()
recording.receive("You")
recording.receive("Good")
recording.receive("吗")
recording.receive(completion: Subscribers.Completion.finished)
let recordPublisher = Record<String.MyCustomError>(recording: recording)
Copy the code- use
output: [Output], completion: Subscribers.Completion<Failure>Parameter initialization
let recordPublisher = Record<String.MyCustomError>(output: ["You"."Good"."吗"], completion: Subscribers.Completion.finished)
Copy the codeAll three of the above initialization methods have the same effect. I personally prefer the first one, which I find more readable. Once initialized, it looks a lot like Publishers.Sequence
cancellable = recordPublisher
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})Copy the codePrint result:
How are you finishedCopy the codeThe point comes, said above, it is to support codec, what does that mean? That is, we can save our pre-set Publisher model, for example, in JSON, which can be parsed from JSON anywhereRecord.
Examples of coding:
let jsonEncoder = JSONEncoder(a)let jsonEncoded = try? jsonEncoder.encode(recordPublisher)
if let jsonData = jsonEncoded {
let jsonStr = String(data: jsonData, encoding: .utf8)
print(jsonStr ?? "Coding error")}Copy the code{"recording": {"completion": {"success":true},"output": ["You"."Good"."吗"]}}
Copy the codeExamples of decoding:
let jsonDecoder = JSONDecoder(a)let jsonDecoded = try? jsonDecoder.decode(Record<String.MyCustomError>.self, from: jsonEncoded!)
if let record = jsonDecoded {
print(record)
}
Copy the codeRecord<String.MyCustomError>(recording: Combine.Record<Swift.String.MCMarbleDiagramSwiftUI.MyCustomError>.Recording(state: Combine.Record<Swift.String.MCMarbleDiagramSwiftUI.MyCustomError>.Recording.State.complete, output: ["You"."Good"."吗"], completion: Combine.Subscribers.Completion<MCMarbleDiagramSwiftUI.MyCustomError>.finished))
Copy the codeDeferred
/// A publisher that awaits subscription before running the supplied closure to create a publisher for the new subscriber.
@available(macOS 10.15.iOS 13.0.tvOS 13.0.watchOS 6.0.*)
public struct Deferred<DeferredPublisher> : Publisher where DeferredPublisher : Publisher {
reatePublisher: () -> DeferredPublisher
/// Creates a deferred publisher.
///
/// - Parameter createPublisher: The closure to execute when calling `subscribe(_:)`.
public init(createPublisher: @escaping() - >DeferredPublisher)
}
Copy the codeIn short, the core idea of Deferred is to create publisher only after the subscription is received. As you can see from the code above, it creates publisher by initializing a closure argument to the function:
cancellable = Deferred {
return Future<Bool.Error> { promise in
/// simulate the asynchronous process
DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
return promise(.success(true))
}
}
}
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})Copy the codeDeferred and Future are a perfect match. Future will execute immediately regardless of whether there are subscribers or not, and with Deferred it is perfectly possible to add any asynchronous problems to the Pipline.
Also, you can use Publisher, which is more performance intensiveDeferredWrapper, which is initialized only when the subscription is received.
MakeConnectable
class MakeConnectableViewObservableObject: ObservableObject {
var cancellable1: AnyCancellable?
var cancellable2: AnyCancellable?
var cancellable: Cancellable?
init(a) {
let publisher = Just("Hello, world")
.share()
.makeConnectable()
cancellable1 = publisher
.sink(receiveCompletion: {
print("Stream 1 received: \ [$0)")
}, receiveValue: {
print("Stream 1 received: \ [$0)")})DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
self.cancellable2 = publisher
.sink(receiveCompletion: {
print("Stream 2 received: \ [$0)")
}, receiveValue: {
print("Stream 2 received: \ [$0)")})}DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
self.cancellable = publisher.connect()
}
}
}
Copy the codeIn publisher’s normal life cycle, a connection is established as soon as a subscription is received, and in some scenarios, we need to wait for a condition to be met before establishing a connection. The code above is an example. Since Just calls.share(), we want to wait until all subscriptions have been processed before establishing a connection, at which point it is necessary to make Just Connectable.
As simple as that, publisher simply calls.makeconnectable () to become ConnectablePublisher. When you become ConnectablePublisher, you can call the following two methods:
connectautoconnect
In the example above, we wait for the pipline to complete and call self.cancellable = publisher.connect() to open the entire Pipline. This is the most commonly used example.
let publisher = Just("Hello, world")
.share()
.makeConnectable()
.autoconnect()
Copy the codeIf.autoconnect() is called immediately after.makeconnectable (), publisher is doing nothing, and in real development, you should never write code like this. We will also use these two features in the following Timer.
@Published
class PublishedViewObservableObject: ObservableObject {
@Published var text: String = ""
var cancellable: AnyCancellable?
init(a) {
cancellable = $text
.sink(receiveValue: {
print($0)
})
text = "hello, world"}}Copy the code@Published is commonly used in the ObservableObject model in SwiftUI. If the View depends on this property, the View is automatically updated when the property changes.
When we call $text, it returns publisher. Why? Let’s look at the Published definition:
@propertyWrapper public struct Published<Value> {
/// A publisher for properties marked with the `@Published` attribute.
public struct Publisher : Publisher {
/// The kind of values published by this publisher.
public typealias Output = Value
/// The kind of errors this publisher might publish.
///
/// Use `Never` if this `Publisher` does not publish errors.
public typealias Failure = Never
/// Attaches the specified subscriber to this publisher.
///
/// Implementations of ``Publisher`` must implement this method.
///
/// The provided implementation of ``Publisher/subscribe(_:)-4u8kn``calls this method.
///
/// - Parameter subscriber: The subscriber to attach to this ``Publisher``, after which it can receive values.
public func receive<S> (subscriber: S) where Value = = S.Input.S : Subscriber.S.Failure = = Published<Value>.Publisher.Failure
}
/// The property for which this instance exposes a publisher.
///
/// The ``Published/projectedValue`` is the property accessed with the `$` operator.
public var projectedValue: Published<Value>.Publisher { mutating get set}}Copy the codePropertyWrapper @propertyWrapper We will have a special article on it later, Published Inside there is a property projectedValue, which is a Publisher, which we can access with the $sign.
Published is not used in SwiftUI, it is a Feature of Combine so we can use it in any class:
class MyClass {
@Published var text: String = ""
var cancellable: AnyCancellable?
init(a) {
cancellable = $text
.sink(receiveValue: {
print($0)
})
text = "hello, MyClass"}}Copy the codeNotificationCenter
extension Notification.Name {
static let myCustomNotification = Notification.Name("myCustomNotification")}class NotificationCenterPublisherViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
init(a) {
cancellable = NotificationCenter.default
.publisher(for: .myCustomNotification)
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})NotificationCenter.default.post(name: .myCustomNotification, object: nil)}}/// Prints:
/// name = myCustomNotification, object = nil, userInfo = nil
Copy the codeCombine also provides publisher capability for NotificationCenter, which is clearly explained in the above code and won’t be explained here.
Timer
extension Timer {
/// Returns a publisher that repeatedly emits the current date on the given interval.
///
/// - Parameters:
/// - interval: The time interval on which to publish events. For example, D. a value of '0.5' an event approximately every half-second.
/// - tolerance: The allowed timing variance when emitting events. Defaults to `nil`, which allows any variance.
/// - runLoop: The run loop on which the timer runs.
/// - mode: The run loop mode in which to run the timer.
/// - options: Scheduler options passed to the timer. Defaults to `nil`.
/// - Returns: A publisher that repeatedly emits the current date on the given interval.
public static func publish(every interval: TimeInterval.tolerance: TimeInterval? = nil.on runLoop: RunLoop.in mode: RunLoop.Mode.options: RunLoop.SchedulerOptions? = nil) -> Timer.TimerPublisher
/// A publisher that repeatedly emits the current date on a given interval.
final public class TimerPublisher : ConnectablePublisher {
/// The kind of values published by this publisher.
public typealias Output = Date
/// The kind of errors this publisher might publish.
///
/// Use `Never` if this `Publisher` does not publish errors.
public typealias Failure = Never
final public let interval: TimeInterval
final public let tolerance: TimeInterval?
final public let runLoop: RunLoop
final public let mode: RunLoop.Mode
final public let options: RunLoop.SchedulerOptions?
/// Creates a publisher that repeatedly emits the current date on the given interval.
///
/// - Parameters:
/// - interval: The interval on which to publish events.
/// - tolerance: The allowed timing variance when emitting events. Defaults to `nil`, which allows any variance.
/// - runLoop: The run loop on which the timer runs.
/// - mode: The run loop mode in which to run the timer.
/// - options: Scheduler options passed to the timer. Defaults to `nil`.
public init(interval: TimeInterval.tolerance: TimeInterval? = nil.runLoop: RunLoop.mode: RunLoop.Mode.options: RunLoop.SchedulerOptions? = nil)
/// Connects to the publisher, allowing it to produce elements, and returns an instance with which to cancel publishing.
///
/// - Returns: A ``Cancellable`` instance that you use to cancel publishing.
final public func connect(a) -> Cancellable}}Copy the codeTimer is more suitable for publisher. Generally speaking, we use Timer because we need to receive data at a fixed time interval.
As you can see from the above code, if you want to add Combine to elements in the Foundation framework, you can simply write an extension and, for the Timer, bind it to a TimerPublisher.
Pay attention to,TimerPublisherTo achieve theConnectablePublisherProtocol, so it can be accessed.autoconnect()and.connect().
If we want to start the timer automatically, the code looks like this:
cancellable = Timer.publish(every: 1.0, on: .main, in: .common)
.autoconnect()
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)})Copy the codeIf you want to manually control the connection, the code looks like this:
class TimerViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
var timerPublisher = Timer.publish(every: 1.0, on: .main, in: .common)
var connectable: Cancellable?
@Published var count: Int = 1
init(a){}func startTimer(a) {
cancellable = timerPublisher
.sink(receiveCompletion: {
print($0)
}, receiveValue: { _ in
self.count + = 1
})
connectable = timerPublisher
.connect()
}
func stopTimer(a) {
connectable?.cancel()
}
}
Copy the codeNotice when calledconnectable? .cancel(), terminates the Pipline, and the subscription process needs to be restarted.
The effect is as follows:
KeyValueObservingPublisher
Those who are engaged in iOS development must be familiar with KVO. In short, we can monitor the changes of properties of an object. We won’t introduce them in detail, but we can get more detailed information by Using key-value Observing in Swift here.
class MyObjectToObserve: NSObject {
@objc dynamic var myDate = NSDate(timeIntervalSince1970: 0) / / 1970
func updateDate(a) {
myDate = myDate.addingTimeInterval(Double(2 << 30)) // Adds about 68 years.}}Copy the codeclass MyObserver: NSObject {
@objc var objectToObserve: MyObjectToObserve
var observation: NSKeyValueObservation?
init(object: MyObjectToObserve) {
objectToObserve = object
super.init()
observation = observe(
\.objectToObserve.myDate,
options: [.old, .new]
) { object, change in
print("myDate changed from: \(change.oldValue!), updated to: \(change.newValue!)")}}}Copy the codeAlthough we rarely use KVO in our normal coding, Combine also adds extensions to kVO as follows:
final class MyKeyValueClass: NSObject {
@objc dynamic var name: String = ""
@objc dynamic var age: Int = 20
}
class KeyValueObservingPublisherViewObservableObject: ObservableObject {
var cancellable: AnyCancellable?
init(a) {
let keyValueClass = MyKeyValueClass()
cancellable = keyValueClass.publisher(for: \.name)
.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)
})
keyValueClass.name = "James"}}Copy the code- Used in objC classes
@objc dynamicDeclaration attributes - use
keyValueClass.publisher(for: \.name)Create a publisher
DataTaskPublisher
extension URLSession {
/// Returns a publisher that wraps a URL session data task for a given URL.
///
/// The publisher publishes data when the task completes, or terminates if the task fails with an error.
/// - Parameter url: The URL for which to create a data task.
/// - Returns: A publisher that wraps a data task for the URL.
public func dataTaskPublisher(for url: URL) -> URLSession.DataTaskPublisher
/// Returns a publisher that wraps a URL session data task for a given URL request.
///
/// The publisher publishes data when the task completes, or terminates if the task fails with an error.
/// - Parameter request: The URL request for which to create a data task.
/// - Returns: A publisher that wraps a data task for the URL request.
public func dataTaskPublisher(for request: URLRequest) -> URLSession.DataTaskPublisher
public struct DataTaskPublisher : Publisher {
/// The kind of values published by this publisher.
public typealias Output = (data: Data, response: URLResponse)
/// The kind of errors this publisher might publish.
///
/// Use `Never` if this `Publisher` does not publish errors.
public typealias Failure = URLError
public let request: URLRequest
public let session: URLSession
public init(request: URLRequest.session: URLSession)}}Copy the codeURLSession. Shared. DataTaskPublisher (for url: url) is the most frequently used publisher in the network request, it is worth noting that the there are the following:
Output = (data: Data, response: URLResponse): Its output type is a tupleFailure = URLError: The error type is URLError- use
URLorURLRequestCreating a network request
Result.Publisher
class ResultPublisherViewViewObservableObject: ObservableObject {
var cancellable1: AnyCancellable?
var cancellable2: AnyCancellable?
init(a) {
enum MyError: Error {
case custom
}
let publisher = Result<Int.MyError>.Publisher(.success(1))
.share()
cancellable1 = publisher
.sink(receiveCompletion: {
print("Stream 1 received: \ [$0)")
}, receiveValue: {
print("Stream 1 received: \ [$0)")
})
cancellable2 = publisher
.sink(receiveCompletion: {
print("Stream 2 received: \ [$0)")
}, receiveValue: {
print("Stream 2 received: \ [$0)")}}}/// Prints:
/// Stream 1 received: 1
/// Stream 1 received: finished
/// Stream 2 received: finished
Copy the codeResult.publisher is mainly used to send success or failure events, if it is. Success, the data is sent upon receipt of the request and the pipline is terminated, while. Faulure immediately ends the Pipline.
In fact, Result.Publisher is similar to Just, except that:
Result.PublisherYou can sendData + success + failureJustCan only senddata