1
- # MultiProducerSingleConsumerChannel
1
+ # MultiProducerSingleConsumerAsyncChannel
2
2
3
3
* Proposal: [ SAA-0016] ( 0016-multi-producer-single-consumer-channel.md )
4
4
* Authors: [ Franz Busch] ( https://github.com/FranzBusch )
@@ -129,24 +129,24 @@ The above motivation lays out the expected behaviors for any consumer/producer
129
129
system and compares them to the behaviors of ` Async[Throwing]Stream ` and
130
130
` Async[Throwing]Channel ` .
131
131
132
- This section proposes a new type called ` MultiProducerSingleConsumerChannel `
132
+ This section proposes a new type called ` MultiProducerSingleConsumerAsyncChannel `
133
133
that implement all of the above-mentioned behaviors. Importantly, this proposed
134
134
solution is taking advantage of ` ~Copyable ` types to model the
135
135
multi-producer-single-consumer behavior. While the current ` AsyncSequence `
136
136
protocols are not supporting ` ~Copyable ` types we provide a way to convert the
137
137
proposed channel to an asynchronous sequence. This leaves us room to support any
138
138
potential future asynchronous streaming protocol that supports ` ~Copyable ` .
139
139
140
- ### Creating a MultiProducerSingleConsumerChannel
140
+ ### Creating a MultiProducerSingleConsumerAsyncChannel
141
141
142
- You can create an ` MultiProducerSingleConsumerChannel ` instance using the
142
+ You can create an ` MultiProducerSingleConsumerAsyncChannel ` instance using the
143
143
` makeChannel(of: backpressureStrategy:) ` method. This method returns you the
144
144
channel and the source. The source can be used to send new values to the
145
145
asynchronous channel. The new API specifically provides a
146
146
multi-producer/single-consumer pattern.
147
147
148
148
``` swift
149
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
149
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
150
150
of : Int .self ,
151
151
backpressureStrategy : .watermark (low : 2 , high : 4 )
152
152
)
192
192
```
193
193
194
194
The above API offers the most control and highest performance when bridging a
195
- synchronous producer to a ` MultiProducerSingleConsumerChannel ` . First, you have
195
+ synchronous producer to a ` MultiProducerSingleConsumerAsyncChannel ` . First, you have
196
196
to send values using the ` send(contentsOf:) ` which returns a ` SendResult ` . The
197
197
result either indicates that more values should be produced or that a callback
198
198
should be enqueued by calling the `enqueueCallback(callbackToken:
@@ -221,7 +221,7 @@ try await source.send(contentsOf: sequence)
221
221
```
222
222
223
223
With the above APIs, we should be able to effectively bridge any system into a
224
- ` MultiProducerSingleConsumerChannel ` regardless if the system is callback-based,
224
+ ` MultiProducerSingleConsumerAsyncChannel ` regardless if the system is callback-based,
225
225
blocking, or asynchronous.
226
226
227
227
### Multi producer
@@ -232,7 +232,7 @@ region than the original source allowing to pass it into a different isolation
232
232
region to concurrently produce elements.
233
233
234
234
``` swift
235
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
235
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
236
236
of : Int .self ,
237
237
backpressureStrategy : .watermark (low : 2 , high : 5 )
238
238
)
@@ -262,7 +262,7 @@ this:
262
262
263
263
``` swift
264
264
// Termination through calling finish
265
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
265
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
266
266
of : Int .self ,
267
267
backpressureStrategy : .watermark (low : 2 , high : 4 )
268
268
)
@@ -280,7 +280,7 @@ If the channel has a failure type it can also be finished with an error.
280
280
281
281
``` swift
282
282
// Termination through calling finish
283
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
283
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
284
284
of : Int .self ,
285
285
throwing : SomeError.self ,
286
286
backpressureStrategy : .watermark (low : 2 , high : 4 )
@@ -301,7 +301,7 @@ will happen automatically when the source is last used or explicitly consumed.
301
301
302
302
``` swift
303
303
// Termination through deiniting the source
304
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
304
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
305
305
of : Int .self ,
306
306
backpressureStrategy : .watermark (low : 2 , high : 4 )
307
307
)
@@ -322,7 +322,7 @@ callback. Termination of the producer happens in the following scenarios:
322
322
323
323
``` swift
324
324
// Termination through task cancellation
325
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
325
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
326
326
of : Int .self ,
327
327
backpressureStrategy : .watermark (low : 2 , high : 4 )
328
328
)
@@ -338,7 +338,7 @@ task.cancel() // Prints Terminated
338
338
339
339
``` swift
340
340
// Termination through deiniting the channel
341
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
341
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
342
342
of : Int .self ,
343
343
backpressureStrategy : .watermark (low : 2 , high : 4 )
344
344
)
@@ -350,7 +350,7 @@ _ = consume channel // Prints Terminated
350
350
351
351
``` swift
352
352
// Termination through finishing the source and consuming the last element
353
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
353
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
354
354
of : Int .self ,
355
355
backpressureStrategy : .watermark (low : 2 , high : 4 )
356
356
)
@@ -367,7 +367,7 @@ await channel.next() // Prints Terminated
367
367
368
368
``` swift
369
369
// Termination through deiniting the last source and consuming the last element
370
- let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel (
370
+ let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel (
371
371
of : Int .self ,
372
372
backpressureStrategy : .watermark (low : 2 , high : 4 )
373
373
)
@@ -394,29 +394,29 @@ producer has been terminated will result in an error thrown from the send method
394
394
``` swift
395
395
#if compiler (>= 6.1 )
396
396
/// An error that is thrown from the various `send` methods of the
397
- /// ``MultiProducerSingleConsumerChannel /Source``.
397
+ /// ``MultiProducerSingleConsumerAsyncChannel /Source``.
398
398
///
399
399
/// This error is thrown when the channel is already finished when
400
400
/// trying to send new elements to the source.
401
- public struct MultiProducerSingleConsumerChannelAlreadyFinishedError : Error { }
401
+ public struct MultiProducerSingleConsumerAsyncChannelAlreadyFinishedError : Error { }
402
402
403
403
/// A multi-producer single-consumer channel.
404
404
///
405
- /// The ``MultiProducerSingleConsumerChannel `` provides a ``MultiProducerSingleConsumerChannel /Source`` to
405
+ /// The ``MultiProducerSingleConsumerAsyncChannel `` provides a ``MultiProducerSingleConsumerAsyncChannel /Source`` to
406
406
/// send values to the channel. The channel supports different back pressure strategies to control the
407
407
/// buffering and demand. The channel will buffer values until its backpressure strategy decides that the
408
408
/// producer have to wait.
409
409
///
410
410
/// This channel is also suitable for the single-producer single-consumer use-case
411
411
///
412
- /// ## Using a MultiProducerSingleConsumerChannel
412
+ /// ## Using a MultiProducerSingleConsumerAsyncChannel
413
413
///
414
- /// To use a ``MultiProducerSingleConsumerChannel `` you have to create a new channel with its source first by calling
415
- /// the ``MultiProducerSingleConsumerChannel /makeChannel(of:throwing:BackpressureStrategy:)`` method.
414
+ /// To use a ``MultiProducerSingleConsumerAsyncChannel `` you have to create a new channel with its source first by calling
415
+ /// the ``MultiProducerSingleConsumerAsyncChannel /makeChannel(of:throwing:BackpressureStrategy:)`` method.
416
416
/// Afterwards, you can pass the source to the producer and the channel to the consumer.
417
417
///
418
418
/// ```
419
- /// let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel(
419
+ /// let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel(
420
420
/// of: Int.self,
421
421
/// backpressureStrategy: .watermark(low: 2, high: 4)
422
422
/// )
@@ -428,8 +428,8 @@ public struct MultiProducerSingleConsumerChannelAlreadyFinishedError: Error { }
428
428
///
429
429
/// ### Asynchronous producing
430
430
///
431
- /// Values can be send to the source from asynchronous contexts using ``MultiProducerSingleConsumerChannel /Source/send(_:)-8eo96``
432
- /// and ``MultiProducerSingleConsumerChannel /Source/send(contentsOf:)``. Backpressure results in calls
431
+ /// Values can be send to the source from asynchronous contexts using ``MultiProducerSingleConsumerAsyncChannel /Source/send(_:)-8eo96``
432
+ /// and ``MultiProducerSingleConsumerAsyncChannel /Source/send(contentsOf:)``. Backpressure results in calls
433
433
/// to the `send` methods to be suspended. Once more elements should be produced the `send` methods will be resumed.
434
434
///
435
435
/// ```
@@ -495,14 +495,14 @@ public struct MultiProducerSingleConsumerChannelAlreadyFinishedError: Error { }
495
495
/// getting cancelled, the source will get notified about the termination if a termination callback has been set
496
496
/// before by calling ``Source/setOnTerminationCallback(_:)``.
497
497
@available (macOS 15.0 , iOS 18.0 , watchOS 11.0 , tvOS 18.0 , visionOS 2.0 , * )
498
- public struct MultiProducerSingleConsumerChannel <Element , Failure : Error >: ~ Copyable {
498
+ public struct MultiProducerSingleConsumerAsyncChannel <Element , Failure : Error >: ~ Copyable {
499
499
/// A struct containing the initialized channel and source.
500
500
///
501
501
/// This struct can be deconstructed by consuming the individual
502
502
/// components from it.
503
503
///
504
504
/// ```swift
505
- /// let channelAndSource = MultiProducerSingleConsumerChannel .makeChannel(
505
+ /// let channelAndSource = MultiProducerSingleConsumerAsyncChannel .makeChannel(
506
506
/// of: Int.self,
507
507
/// backpressureStrategy: .watermark(low: 5, high: 10)
508
508
/// )
@@ -512,12 +512,12 @@ public struct MultiProducerSingleConsumerChannel<Element, Failure: Error>: ~Copy
512
512
@frozen
513
513
public struct ChannelAndStream : ~ Copyable {
514
514
/// The channel.
515
- public var channel: MultiProducerSingleConsumerChannel
515
+ public var channel: MultiProducerSingleConsumerAsyncChannel
516
516
/// The source.
517
517
public var source: Source
518
518
}
519
519
520
- /// Initializes a new ``MultiProducerSingleConsumerChannel `` and an ``MultiProducerSingleConsumerChannel /Source``.
520
+ /// Initializes a new ``MultiProducerSingleConsumerAsyncChannel `` and an ``MultiProducerSingleConsumerAsyncChannel /Source``.
521
521
///
522
522
/// - Parameters:
523
523
/// - elementType: The element type of the channel.
@@ -547,7 +547,7 @@ public struct MultiProducerSingleConsumerChannel<Element, Failure: Error>: ~Copy
547
547
}
548
548
549
549
@available (macOS 15.0 , iOS 18.0 , watchOS 11.0 , tvOS 18.0 , visionOS 2.0 , * )
550
- extension MultiProducerSingleConsumerChannel {
550
+ extension MultiProducerSingleConsumerAsyncChannel {
551
551
/// A struct to send values to the channel.
552
552
///
553
553
/// Use this source to provide elements to the channel by calling one of the `send` methods.
@@ -583,18 +583,18 @@ extension MultiProducerSingleConsumerChannel {
583
583
/// A type that indicates the result of sending elements to the source.
584
584
public enum SendResult : ~ Copyable , Sendable {
585
585
/// An opaque token that is returned when the channel's backpressure strategy indicated that production should
586
- /// be suspended. Use this token to enqueue a callback by calling the ``MultiProducerSingleConsumerChannel /Source/enqueueCallback(callbackToken:onProduceMore:)`` method.
586
+ /// be suspended. Use this token to enqueue a callback by calling the ``MultiProducerSingleConsumerAsyncChannel /Source/enqueueCallback(callbackToken:onProduceMore:)`` method.
587
587
///
588
- /// - Important: This token must only be passed once to ``MultiProducerSingleConsumerChannel /Source/enqueueCallback(callbackToken:onProduceMore:)``
589
- /// and ``MultiProducerSingleConsumerChannel /Source/cancelCallback(callbackToken:)``.
588
+ /// - Important: This token must only be passed once to ``MultiProducerSingleConsumerAsyncChannel /Source/enqueueCallback(callbackToken:onProduceMore:)``
589
+ /// and ``MultiProducerSingleConsumerAsyncChannel /Source/cancelCallback(callbackToken:)``.
590
590
public struct CallbackToken : Sendable , Hashable { }
591
591
592
592
/// Indicates that more elements should be produced and send to the source.
593
593
case produceMore
594
594
595
595
/// Indicates that a callback should be enqueued.
596
596
///
597
- /// The associated token should be passed to the ````MultiProducerSingleConsumerChannel /Source/enqueueCallback(callbackToken:onProduceMore:)```` method.
597
+ /// The associated token should be passed to the ````MultiProducerSingleConsumerAsyncChannel /Source/enqueueCallback(callbackToken:onProduceMore:)```` method.
598
598
case enqueueCallback (CallbackToken)
599
599
}
600
600
@@ -728,7 +728,7 @@ extension MultiProducerSingleConsumerChannel {
728
728
729
729
/// Indicates that the production terminated.
730
730
///
731
- /// After all buffered elements are consumed the subsequent call to ``MultiProducerSingleConsumerChannel /next(isolation:)`` will return
731
+ /// After all buffered elements are consumed the subsequent call to ``MultiProducerSingleConsumerAsyncChannel /next(isolation:)`` will return
732
732
/// `nil` or throw an error.
733
733
///
734
734
/// Calling this function more than once has no effect. After calling finish, the channel enters a terminal state and doesn't accept
@@ -741,7 +741,7 @@ extension MultiProducerSingleConsumerChannel {
741
741
}
742
742
743
743
@available (macOS 15.0 , iOS 18.0 , watchOS 11.0 , tvOS 18.0 , visionOS 2.0 , * )
744
- extension MultiProducerSingleConsumerChannel {
744
+ extension MultiProducerSingleConsumerAsyncChannel {
745
745
/// Converts the channel to an asynchronous sequence for consumption.
746
746
///
747
747
/// - Important: The returned asynchronous sequence only supports a single iterator to be created and
0 commit comments