Combine+Buffer.swift

// Based on this StackOverflow answer: https://stackoverflow.com/a/61273595/4239752

import Foundation
import Combine

extension Publisher {
	/// collects elements from the source sequence until the boundary sequence fires. Then it emits the elements as an array and begins collecting again.
	func buffer<T: Publisher, U>(_ boundary: T) -> AnyPublisher<[Output], Failure> where T.Output == U {
		let subject = PassthroughSubject<[Output], Failure>()

		var buffer: [Output] = []
		let lock = NSRecursiveLock()

		let boundaryDisposable = boundary.sink(
			receiveCompletion: { _ in },
			receiveValue: { _ in
				lock.lock(); defer { lock.unlock() }
				subject.send(buffer)
				buffer = []
		})

		let disposable = self.sink(
			receiveCompletion: { event in
				lock.lock(); defer { lock.unlock() }
				switch event {
				case .finished:
					subject.send(buffer)
					subject.send(completion: .finished)
				case .failure(let error):
					subject.send(completion: .failure(error))
					buffer = []
				}
		},
			receiveValue: { element in
				lock.lock(); defer { lock.unlock() }
				buffer.append(element)
		})

		let completion = AnyCancellable {
			boundaryDisposable.cancel()
			disposable.cancel()
		}

		return subject
			.handleEvents(receiveCancel: completion.cancel)
			.eraseToAnyPublisher()
	}
}