01. Combine (1) ———— Publisher And Subscriber

(Updated: )
Reactive Programming › 01. Combine

It will take about 5 minutes to finish reading this article.

Next, I will record the learning process of Combine through a series of articles.

1. Publisher

1.1 Concept of Publisher
Publisher is used to declare a protocol type that can transmit a series of values over time. Its source code is as follows:

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

It has two generic parameters: Output represents the output data, and Failure represents failure.

It also has a receive function that accepts an object of Subscriber protocol type, and the Subscriber’s Input and Failure associated types must match the Input and Failure types declared by the Publisher.

Publisher accepts a subscriber by implementing the receive(subscriber:) method. A Publisher can pass elements to one or more Subscriber instances.

1.2 Sample Code

1
2
3
4
5
6
7
8
9
10
let publisher = NotificationCenter.Publisher(center: .default, name: .titleData)
    .compactMap { notification -> String? in
        return (notification.object as? Item)?.title
    }

let subscriber1 = Subscribers.Assign(object: self, keyPath: \.titleText1)
let subscriber2 = Subscribers.Assign(object: self, keyPath: \.titleText2)

publisher.receive(subscriber: subscriber1)
publisher.receive(subscriber: subscriber2)

1.3 Publishers

Apple uses enum to define a Publishers namespace, as follows:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

enum Publishers {
    struct Sequence {
        
    }
    
    struct Catch {
        
    }
    
    struct ReceiveOn {
        
    }
    ......
}

Note: This enumeration is not an enumeration type, but a namespace.

Used to contain various publisher-related types and operators, used to organize and manage publisher-related functions. For example:

  • Publishers.Sequence: used to convert arrays to Publishers.

  • Publishers.Catch: used to handle errors and return an alternate Publisher.

  • Publishers.CombineLatest: used to merge the latest values of multiple Publishers.

  • Publishers.ReceiveOn: used to specify the dispatch queue for receiving events.

The following is a sample code for using Publishers.Sequence

1
2
3
4
5
6
7
8
9
10
11
12
13
14
let numbers = [1, 2, 3, 4, 5]
let publisher = Publishers.Sequence<[Int], Error>(sequence: numbers)
let subscription = publisher
     .sink(receiveCompletion: { completion in
         switch completion {
         case .finished:
             print("Data flow completed.")
         case .failure(let error):
             print("Data flow error: \(error)")
         }
     }, receiveValue: { value in
         print("Received value: \(value)")
     })
subscription.cancel()

The Combine framework provides a large number of this type,
See here: https://developer.apple.com/documentation/combine/publishers.

1.4 Other Publisher

Of course, we can customize our own CustomPublisher by extending the system Publisher protocol. At the same time, the Combine framework also provides us with many extended Publishers. We can use them first. Here are a few:

1.4.1 Subject

1
protocol Subject<Output, Failure> : AnyObject, Publisher

Subject is a Publisher with publish and subscribe capabilities. Common Subject types include:

  • PassthroughSubject: New values can be published to all subscribers.

  • CurrentValueSubject: Can hold and publish the current value, as well as send new values to subscribers.

1.4.2 ConnectablePublisher

1
protocol ConnectablePublisher<Output, Failure> : Publisher

It represents a connectable publisher, which requires a connect method that allows manual connection and disconnection of publishers.

2. Subscriber

2.1 Concept of Subscriber

Subscriber is a protocol that declares types that can receive input from Publisher.

1
2
3
4
5
6
7
8
public protocol Subscriber<Input, Failure> : CustomCombineIdentifierConvertible {
    associatedtype Input
    associatedtype Failure : Error

    func receive(subscription: Subscription)
    func receive(_ input: Self.Input) -> Subscribers.Demand
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

Subscriber instances receive a stream of elements from the publisher, as well as lifecycle events that describe changes to their relationships. It also has two generic parameters: Input and Failure, whose types must match the Output and Failure of their corresponding publishers. It also has three functions for interaction between Subscriber and Publisher.

You connect a subscriber to a publisher by calling the publisher’s subscribe(:) method. After the call, the publisher will call the subscriber’s receive(subscription:) method. The subscriber thus obtains a Subscription instance, which is used to request elements from the publisher and can also choose to cancel the subscription. After the subscriber makes the initial request, the publisher calls receive(:) (possibly asynchronously) to send the newly published elements. If the publisher stops publishing, it calls receive(completion:) with a parameter of type Subscribers.Completion to indicate whether the publishing completed normally or with an error.

Combine provides the following subscribers as operators of the Publisher type:

  • sink(receiveCompletion:receiveValue:) executes an arbitrary closure when a completion signal is received and each time a new element is received.

  • assign(to:on:) writes each newly received value to the property identified by the key path on the specified instance.

2.2 Sample Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
//Define a publisher, here use Just to create a publisher that publishes a single value
let publisher = Just("Hello, Combine!")

// Use the sink operator to create a Subscriber and handle the received values and completion events
let subscriber = Subscribers.Sink<String, Never>(
     receiveCompletion: { completion in
         switch completion {
         case .finished:
             print("Subscription completed.")
         case .failure(let error):
             print("Subscription error: \(error)")
         }
     },
     receiveValue: { value in
         print("Received value: \(value)")
     }
)

// Connect Subscriber to publisher
let subscription = publisher.subscribe(subscriber)
// unsubscribe
subscription.cancel()

In the above example, we first create a publisher, which publishes a string value. We then use Subscribers.Sink to create a Subscriber and specify a closure to receive the completion event and value. Finally, we connect the subscriber to the publisher via the subscribe method and unsubscribe after subscribing.

2.3 Subscribers

Subscriber also has a Subscribers namespace, which contains various publisher-related types and operators and is used to organize and manage publisher-related functions. as follows:

  • Subscribers.Sink: Used to execute closures to handle values and completion events sent by publishers. It is usually used to perform custom operations, such as writing values to properties or performing some logic.

  • Subscribers.Assign: used to assign the value sent by the publisher to the specified attribute. It is typically used for binding with interface elements to associate Combine data flows with user interface updates.

  • Subscribers.Completion: is an enumeration type that represents the completion status of subscriptions. It has two possible values: .finished indicates that the subscription is completed, and .failure(error) indicates that an error occurred in the subscription.

  • Subscribers.Demand: Indicates the number of elements required by subscribers.