import AudioToolbox
import Foundation
import Cancellation
import Scope
import Sockets
import Streams

public extension Socket
{
	public func createAudioStream() -> ReadableStream<AudioData>
	{
		let tokenSource = CancellationTokenSource()
		let stream = Streams.Stream<AudioData>()
		stream.addDownstreamDisposable(Scope {
			tokenSource.cancel()
		})

		var description: AudioStreamBasicDescription? = nil

		let parse = try! AFSUtility.CreateCustomAudioStreamParser(
			onStreamReady: { (_, asbd, cookieData) in
				description = asbd.pointee
		},
			onPackets: { (_ numBytes: UInt32, _ numPackets: UInt32, _ inputData: UnsafeRawPointer, _ packetDescriptions: UnsafeMutablePointer<AudioStreamPacketDescription>?, _ startTime: UnsafePointer<AudioTimeStamp>?) in
				stream.publish(AudioData.Create(description!, numBytes, numPackets, inputData, packetDescriptions, startTime))
		})

		DispatchQueue.global().async {
			while !tokenSource.isCancellationRequested && self.isValid, let data = self.read(maxBytes: 4 * 1024) {
				try! parse(data)
			}
		}

		return ReadableStream(stream)
	}
}

import AudioToolbox
import Foundation
import Sockets
import Streams

public extension Socket
{
	public func createAudioStream() -> ReadableStream<AudioData>
	{
		let stream = Streams.Stream<AudioData>()
		var description: AudioStreamBasicDescription? = nil

		let parse = try! AFSUtility.CreateCustomAudioStreamParser(
			onStreamReady: { (_, asbd, cookieData) in
				description = asbd.pointee
		},
			onPackets: { (_ numBytes: UInt32, _ numPackets: UInt32, _ inputData: UnsafeRawPointer, _ packetDescriptions: UnsafeMutablePointer<AudioStreamPacketDescription>?, _ startTime: UnsafePointer<AudioTimeStamp>?) in
				stream.publish(AudioData.Create(description!, numBytes, numPackets, inputData, packetDescriptions, startTime))
		})

		DispatchQueue.global().async {
			while let data = self.read(maxBytes: 4 * 1024) {
				try! parse(data)
			}
		}

		return ReadableStream(stream)
	}
}

--- 
+++ 
@@ -1,5 +1,7 @@
 import AudioToolbox
 import Foundation
+import Cancellation
+import Scope
 import Sockets
 import Streams
 
@@ -7,7 +9,12 @@
 {
 	public func createAudioStream() -> ReadableStream<AudioData>
 	{
+		let tokenSource = CancellationTokenSource()
 		let stream = Streams.Stream<AudioData>()
+		stream.addDownstreamDisposable(Scope {
+			tokenSource.cancel()
+		})
+
 		var description: AudioStreamBasicDescription? = nil
 
 		let parse = try! AFSUtility.CreateCustomAudioStreamParser(
@@ -19,7 +26,7 @@
 		})
 
 		DispatchQueue.global().async {
-			while let data = self.read(maxBytes: 4 * 1024) {
+			while !tokenSource.isCancellationRequested && self.isValid, let data = self.read(maxBytes: 4 * 1024) {
 				try! parse(data)
 			}
 		}
Stop publishing to the audio stream when the socket is disposed and when the stream is disposed.
