0.0.1 implementation

Author: adam-zethraeus

.gitignore

@@ -0,0 +1,8 @@
+.DS_Store
+/.build
+/Packages
+xcuserdata/
+DerivedData/
+.swiftpm/configuration/registries.json
+.swiftpm/xcode/package.xcworkspace/contents.xcworkspacedata
+.netrc

.swift-format.json

@@ -0,0 +1,79 @@
+{
+  "fileScopedDeclarationPrivacy" : {
+    "accessLevel" : "private"
+  },
+  "indentConditionalCompilationBlocks" : true,
+  "indentSwitchCaseLabels" : false,
+  "indentation" : {
+    "spaces" : 2
+  },
+  "lineBreakAroundMultilineExpressionChainComponents" : false,
+  "lineBreakBeforeControlFlowKeywords" : false,
+  "lineBreakBeforeEachArgument" : false,
+  "lineBreakBeforeEachGenericRequirement" : false,
+  "lineBreakBetweenDeclarationAttributes" : false,
+  "lineLength" : 100,
+  "maximumBlankLines" : 1,
+  "multiElementCollectionTrailingCommas" : true,
+  "noAssignmentInExpressions" : {
+    "allowedFunctions" : [
+      "XCTAssertNoThrow"
+    ]
+  },
+  "prioritizeKeepingFunctionOutputTogether" : false,
+  "reflowMultilineStringLiterals" : {
+    "never" : {
+
+    }
+  },
+  "respectsExistingLineBreaks" : true,
+  "rules" : {
+    "AllPublicDeclarationsHaveDocumentation" : false,
+    "AlwaysUseLiteralForEmptyCollectionInit" : false,
+    "AlwaysUseLowerCamelCase" : true,
+    "AmbiguousTrailingClosureOverload" : true,
+    "AvoidRetroactiveConformances" : true,
+    "BeginDocumentationCommentWithOneLineSummary" : false,
+    "DoNotUseSemicolons" : true,
+    "DontRepeatTypeInStaticProperties" : true,
+    "FileScopedDeclarationPrivacy" : true,
+    "FullyIndirectEnum" : true,
+    "GroupNumericLiterals" : true,
+    "IdentifiersMustBeASCII" : true,
+    "NeverForceUnwrap" : false,
+    "NeverUseForceTry" : false,
+    "NeverUseImplicitlyUnwrappedOptionals" : false,
+    "NoAccessLevelOnExtensionDeclaration" : true,
+    "NoAssignmentInExpressions" : true,
+    "NoBlockComments" : true,
+    "NoCasesWithOnlyFallthrough" : true,
+    "NoEmptyLinesOpeningClosingBraces" : false,
+    "NoEmptyTrailingClosureParentheses" : true,
+    "NoLabelsInCasePatterns" : true,
+    "NoLeadingUnderscores" : false,
+    "NoParensAroundConditions" : true,
+    "NoPlaygroundLiterals" : true,
+    "NoVoidReturnOnFunctionSignature" : true,
+    "OmitExplicitReturns" : false,
+    "OneCasePerLine" : true,
+    "OneVariableDeclarationPerLine" : true,
+    "OnlyOneTrailingClosureArgument" : true,
+    "OrderedImports" : true,
+    "ReplaceForEachWithForLoop" : true,
+    "ReturnVoidInsteadOfEmptyTuple" : true,
+    "TypeNamesShouldBeCapitalized" : true,
+    "UseEarlyExits" : false,
+    "UseExplicitNilCheckInConditions" : true,
+    "UseLetInEveryBoundCaseVariable" : true,
+    "UseShorthandTypeNames" : true,
+    "UseSingleLinePropertyGetter" : true,
+    "UseSynthesizedInitializer" : true,
+    "UseTripleSlashForDocumentationComments" : true,
+    "UseWhereClausesInForLoops" : false,
+    "ValidateDocumentationComments" : false
+  },
+  "spacesAroundRangeFormationOperators" : false,
+  "spacesBeforeEndOfLineComments" : 2,
+  "tabWidth" : 2,
+  "version" : 1
+}

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 adam zethraeus
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Package.resolved

@@ -0,0 +1,36 @@
+// swift-tools-version: 6.1
+
+import PackageDescription
+
+let package = Package(
+  name: "AsyncBroadcaster",
+  platforms: [.iOS(.v18), .macOS(.v15), .tvOS(.v18), .watchOS(.v11), .visionOS(.v2)],
+  products: [
+    .library(
+      name: "AsyncBroadcaster",
+      targets: ["AsyncBroadcaster"])
+  ],
+  dependencies: [
+    .package(
+      url: "https://github.com/apple/swift-collections",
+      from: "1.1.4"
+    ),
+    .package(
+      url: "https://github.com/apple/swift-async-algorithms",
+      from: "1.0.3"
+    ),
+  ],
+  targets: [
+    .target(
+      name: "AsyncBroadcaster",
+      dependencies: [
+        .product(name: "DequeModule", package: "swift-collections"),
+        .product(name: "AsyncAlgorithms", package: "swift-async-algorithms"),
+      ]
+    ),
+    .testTarget(
+      name: "AsyncBroadcasterTests",
+      dependencies: ["AsyncBroadcaster"]
+    ),
+  ]
+)

Package.swift

@@ -0,0 +1,81 @@
+# AsyncBroadcaster
+
+Broadcasting/Multicasting for Swift's AsyncSequences
+
+## Features
+
+- **Broadcast**: make an AsyncSequence safe for multiple subscribers.
+- **Yield**: synchronously yield values to an AsyncBroadcaster.
+- **Replay behavior**: Control how new subscribers receive past values
+  - `.none`: No replay
+  - `.latest(n)`: Replay up to n most recent values
+  - `.unbounded`: Replay all historical values
+
+## Installation
+
+### Swift Package Manager
+
+Add the following to your `Package.swift` file:
+
+```swift
+dependencies: [
+    .package(url: "https://github.com/adam-zethraeus/AsyncBroadcaster.git", from: "0.0.1")
+]
+```
+
+## Usage
+
+### Basic Example
+
+```swift
+
+    let stream = [1,2,3].async.broadcast()
+
+    let one = Task {
+      var result: [Int] = []
+      for await i in stream {
+        result.append(i)
+      }
+      return result
+    }
+
+    await #expect(one.value == [1,2,3])
+```
+
+### Synchronously emitting to a Broadcaster's Continuation 
+
+```swift
+let channel = AsyncBroadcaster.makeAsyncBroadcaster(of: Int.self, replaying: .latest(3))
+
+let task1 = Task {
+  var results: [Int] = []
+  for await value in channel.stream {
+    results.append(value)
+  }
+  return results
+}
+
+let task2 = Task {
+  var results: [Int] = []
+  for await value in channel.stream {
+    results.append(value)
+  }
+  return results
+}
+
+try await Task.sleep(for: .milliseconds(100))
+
+channel.continuation.yield(1)
+channel.continuation.yield(2)
+channel.continuation.yield(3)
+channel.continuation.finish()
+let results1 = await task1.value
+let results2 = await task2.value
+
+#expect(results1 == [1, 2, 3])
+#expect(results2 == [1, 2, 3])
+```
+
+## License
+
+MIT

README.md

@@ -0,0 +1,29 @@
+public enum AsyncBuffer: Sendable {
+  case none
+  case latest(Int)
+  case unbounded
+
+  public func prune<T>(elements: inout [T]) {
+    switch self {
+    case .none:
+      elements.removeAll()
+    case .latest(let count):
+      elements = elements.suffix(count)
+    case .unbounded:
+      break
+    }
+  }
+}
+
+extension AsyncStream.Continuation.BufferingPolicy {
+  init(_ asyncBuffer: AsyncBuffer) {
+    switch asyncBuffer {
+    case .none:
+      self = .bufferingNewest(0)
+    case .latest(let count):
+      self = .bufferingNewest(count)
+    case .unbounded:
+      self = .unbounded
+    }
+  }
+}

Sources/AsyncBroadcaster/AsyncBuffer.swift

@@ -0,0 +1,47 @@
+public final class AsyncBroadcaster<Element: Sendable>: AsyncSequence, Sendable {
+
+  public typealias Element = Element
+
+  public init<S: AsyncSequence>(replay: AsyncBuffer, sequence: sending S)
+  where S.Element == Element {
+    let controller = MulticastController<Element>(
+      sequence.map(MulticastController.Event.publish), replay: replay)
+    self.controller = controller
+    self.memory = replay
+  }
+
+  let controller: MulticastController<Element>
+  let memory: AsyncBuffer
+
+  public func makeAsyncIterator() -> Iterator {
+    let underlying = AsyncStream<Element>
+      .makeStream(
+        of: Element.self,
+        bufferingPolicy: .unbounded
+      )
+    controller.handle(.subscribe(underlying.continuation))
+    return Iterator(underlying: underlying.stream.makeAsyncIterator())
+  }
+
+  public struct Iterator: AsyncIteratorProtocol {
+    init(underlying: AsyncStream<Element>.Iterator) {
+      self.underlying = underlying
+    }
+    private var underlying: AsyncStream<Element>.Iterator
+    public mutating func next() async -> Element? {
+      await underlying.next()
+    }
+    @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, *)
+    public mutating func next(isolation: isolated (any Actor)?) async throws(Never)
+      -> Element?
+    {
+      await underlying.next(isolation: isolation)
+    }
+  }
+}
+
+extension AsyncSequence where Self: Sendable, Self.Element: Sendable {
+  public func broadcast(replay: AsyncBuffer = .none) -> AsyncBroadcaster<Element> {
+    AsyncBroadcaster(replay: replay, sequence: self)
+  }
+}

Sources/AsyncBroadcaster/Broadcaster.swift

@@ -0,0 +1,28 @@
+extension AsyncBroadcaster {
+
+  public struct Continuation: Sendable {
+    let continuation: AsyncStream<Element>.Continuation
+
+    public func yield(_ element: sending Element) {
+      continuation.yield(element)
+    }
+
+    public consuming func finish() {
+      continuation.finish()
+    }
+  }
+
+  public static func makeAsyncBroadcaster(
+    of: Element.Type = Element.self, replaying: AsyncBuffer = .none
+  ) -> Subject {
+    let upstream = AsyncStream.makeStream(of: Element.self)
+    return Subject(
+      stream: upstream.stream.broadcast(replay: replaying),
+      continuation: .init(continuation: upstream.continuation))
+  }
+
+  public struct Subject: Sendable {
+    public let stream: AsyncBroadcaster<Element>
+    public let continuation: Continuation
+  }
+}