Iterating over internet socket messages with async / await in Swift – Donny Wals

In iOS 13, we gained the power to simply ship and obtain information utilizing internet sockets via URLSession. With async/await, we gained the power to fetch information from servers utilizing the await key phrase and we are able to iterate over asynchronous sequences utilizing async for loops.

We are able to even learn information from a URL one line at a time by calling the traces property on URL:

let url = URL(string: "https://donnywals.com")!

for attempt await line in url.traces {
    // use line
}

Whereas that is actually cool and permits us to construct apps that ingest information in actual time if the server helps streaming our bodies, we can not use the traces property to arrange an online socket connection and pay attention for incoming messages and doubtlessly ship messages over the identical connection too.

On this submit, you’ll study every part it’s essential find out about constructing your individual mechanism to conveniently iterate over messages from an online socket asynchronously. We’ll leverage some present performance from URLSessionWebSocketTask and AsyncThrowingStream to construct our personal AsyncSequence that conveniently wraps our URLSessionWebSocketTask.

Notice that the ensuing code has solely had comparatively restricted testing performed so I can not assure that the offered resolution will probably be 100% right for every part you throw at it. Should you discover any points with the ultimate code, be happy to contact me. Bonus factors if you happen to’re capable of present some concepts for a possible repair.

Utilizing an online socket with out async / await

Earlier than we get began, let’s rapidly assessment use an online socket with out async/await. The code particulars are outlined in this submit. You’ll want to learn it if you wish to study extra about utilizing internet sockets in your apps.

let url = URL(string: "ws://127.0.0.1:8080")!
let socketConnection = URLSession.shared.webSocketTask(with: url)
socketConnection.resume()

func setReceiveHandler() {
    socketConnection.obtain { lead to
        defer { self.setReceiveHandler() }

        do {
            let message = attempt end result.get()
            swap message {
            case let .string(string):
                print(string)
            case let .information(information):
                print(information)
            @unknown default:
                print("unkown message obtained")
            }
        } catch {
            // deal with the error
            print(error)
        }
    }
}

setReceiveHandler()

Discover how, to obtain messages from the socket, I need to name obtain with a completion handler. This technique solely permits me to obtain a single incoming message, so I need to re-set my handler after receiving a message to robotically start listening for the subsequent message.

This can be a nice instance of a state of affairs the place an async for loop reminiscent of for attempt await message in socketConnection would make a variety of sense. Sadly, this isn’t attainable out of the field. Nevertheless, URLSessionWebSocketTask offers some type of assist for async / await so we’re not fully out of luck.

A fundamental implementation of internet sockets with async / await

Whereas URLSessionWebSocketTask doesn’t expose an AsyncSequence that emits incoming messages out of the field, it does include an async model of the obtain technique you noticed earlier.

This permits us to rewrite the instance above as an async technique as follows:

func setReceiveHandler() async {
    do {
        let message = attempt await socketConnection.obtain()

        swap message {
        case let .string(string):
          print(string)
        case let .information(information):
          print(information)
        @unknown default:
          print("unkown message obtained")
        }
    } catch {
        print(error)
    }

    await setReceiveHandler()
}

This code works simply effective, besides we don’t actually have a method to cease the recursion right here. The code you noticed earlier truly has the very same concern; there’s no situation to cease listening for internet socket messages even when the online socket connection has already been closed.

We may enhance our code by solely recursing if:

1. We didn’t encounter any errors
2. The socket connection continues to be energetic

This may look a bit as follows:

func setReceiveHandler() async {
    guard socketConnection.closeCode == .invalid else {
        return
    }

    do {
        let message = attempt await socketConnection.obtain()

        swap message {
        case let .string(string):
          print(string)
        case let .information(information):
          print(information)
        @unknown default:
          print("unkown message obtained")
        }

        await setReceiveHandler()
    } catch {
        print(error)
    }
}

An open internet socket’s closed code is at all times stated to invalid to sign that the connection has not (but) been closed. We are able to leverage this to examine that our connection continues to be energetic earlier than ready for the subsequent message to be obtained.

That is significantly better already as a result of we respect closed sockets and failures a lot nicer now, however we may enhance the readability of this code a tiny bit by leveraging a whereas loop as a substitute of recursively calling the setReceiveHandler operate:

func setReceiveHandler() async {
    var isActive = true

    whereas isActive && socketConnection.closeCode == .invalid {
        do {
            let message = attempt await socketConnection.obtain()

            swap message {
            case let .string(string):
              print(string)
            case let .information(information):
              print(information)
            @unknown default:
              print("unkown message obtained")
            }
        } catch {
            print(error)
            isActive = false
        }
    }
}

To me, this model of the code is barely simpler to learn however that may not be the case for you. It’s functionally equal so you possibly can select to make use of whichever possibility fits you finest.

Whereas this code works, I’m not fairly pleased with the place we’ve landed proper now. There’s a variety of logic on this operate and I would favor to separate dealing with the incoming values from the calls to socketConnection.obtain() by some means. Ideally, I ought to have the ability to write the next:

do {
    for attempt await message in socketConnection {
        swap message {
        case let .string(string):
            print(string)
        case let .information(information):
            print(information)
        @unknown default:
            print("unkown message obtained")
      }
} catch {
    // deal with error
}

That is a lot, a lot nicer from a call-site perspective and it could enable us to place the ugly bits elsewhere.

To do that, we are able to leverage the facility of AsyncStream which permits us to construct a customized async sequence of values.

Utilizing AsyncStream to emit internet socket messages

Given our finish objective, there are a number of methods for us to get the place we need to be. The simplest approach could be to put in writing a operate in an extension on URLSessionWebSocketTask that might encapsulate the whereas loop you noticed earlier. This implementation would look as follows:

typealias WebSocketStream = AsyncThrowingStream<URLSessionWebSocketTask.Message, Error>

public extension URLSessionWebSocketTask {    
    var stream: WebSocketStream {
        return WebSocketStream { continuation in
            Process {
                var isAlive = true

                whereas isAlive && closeCode == .invalid {
                    do {
                        let worth = attempt await obtain()
                        continuation.yield(worth)
                    } catch {
                        continuation.end(throwing: error)
                        isAlive = false
                    }
                }
            }
        }
    }
}

To make the code slightly bit simpler to learn, I’ve outlined a typealias for my AsyncThrowingStream so we don’t have to have a look at the identical lengthy kind signature everywhere.

The code above creates an occasion of AsyncThrowingStream that asynchronously awaits new values from the online socket so long as the online socket is taken into account energetic and hasn’t been closed. To emit incoming messages and potential errors, the continuation’s yield and end strategies are used. These strategies will both emit a brand new worth (yield) or finish the stream of values with an error (end).

This code works nice in lots of conditions, however there may be one concern. If we resolve to shut the online socket connection from the app’s facet by calling cancel(with:motive:) on our socketConnection, our WebSocketStream doesn’t finish. As a substitute, will probably be caught ready for messages, and the decision website will probably be caught too.

Process {
    attempt await Process.sleep(for: .seconds(5))
    attempt await socketConnection.cancel(with: .goingAway, motive: nil)
}

Process {    
    do {
        for attempt await message in socketConnection.stream2 {
            // deal with incoming messages
        }
    } catch {
        // deal with error
    }

    print("this is able to by no means be printed")
}

If every part works as anticipated, our internet socket connection will shut after 5 seconds. At that time, our for loop ought to finish and our print assertion ought to execute, because the asynchronous stream is not energetic. Sadly, this isn’t the case, so we have to discover a higher strategy to mannequin our stream.

URLSessionWebSocketTask doesn’t present a approach for us to detect cancellation. So, I’ve discovered that it’s best to make use of an object that wraps the URLSessionWebSocketTask, and to cancel the duty via that object. This permits us to each finish the async stream we’re offering to callers and shut the online socket reference to one technique name.

Right here’s what that object appears like:

class SocketStream: AsyncSequence {
    typealias AsyncIterator = WebSocketStream.Iterator
    typealias Ingredient = URLSessionWebSocketTask.Message

    non-public var continuation: WebSocketStream.Continuation?
    non-public let job: URLSessionWebSocketTask

    non-public lazy var stream: WebSocketStream = {
        return WebSocketStream { continuation in
            self.continuation = continuation

            Process {
                var isAlive = true

                whereas isAlive && job.closeCode == .invalid {
                    do {
                        let worth = attempt await job.obtain()
                        continuation.yield(worth)
                    } catch {
                        continuation.end(throwing: error)
                        isAlive = false
                    }
                }
            }
        }
    }()

    init(job: URLSessionWebSocketTask) {
        self.job = job
        job.resume()
    }

    deinit {
        continuation?.end()
    }

    func makeAsyncIterator() -> AsyncIterator {
        return stream.makeAsyncIterator()
    }

    func cancel() async throws {
        job.cancel(with: .goingAway, motive: nil)
        continuation?.end()
    }
}

There’s a bunch of code right here, but it surely’s not too dangerous. The primary few traces are all about organising some kind aliases and properties for comfort. The lazy var stream is actually the very same code that you simply’ve already within the URLSessionWebSocketTask extension from earlier than.

When our SocketStream‘s deinit is named we make it possible for we finish our stream. There’s additionally a cancel technique that closes the socket connection in addition to the stream. As a result of SocketStream conforms to AsyncSequence we should present an Iterator object that’s used after we attempt to iterate over our SocketStreams. We merely ask our inside stream object to make an iterator and use that as our return worth.

Utilizing the code above appears as follows:

let url = URL(string: "ws://127.0.0.1:8080")!
let socketConnection = URLSession.shared.webSocketTask(with: url)
let stream = SocketStream(job: socketConnection)

Process {  
    do {
        for attempt await message in stream {
            // deal with incoming messages
        }
    } catch {
        // deal with error
    }

    print("this will probably be printed as soon as the stream ends")
}

To cancel our stream after 5 seconds similar to earlier than, you possibly can run the next job in parallel with our iterating job:

Process {
    attempt await Process.sleep(for: .seconds(5))
    attempt await stream.cancel()
}

Process {
    // iterate...
}

Whereas that is fairly cool, we do have a little bit of a problem right here due to the next little bit of code:

non-public lazy var stream: WebSocketStream = {
    return WebSocketStream { continuation in
        self.continuation = continuation

        Process {
            var isAlive = true

            whereas isAlive && job.closeCode == .invalid {
                do {
                    let worth = attempt await job.obtain()
                    continuation.yield(worth)
                } catch {
                    continuation.end(throwing: error)
                    isAlive = false
                }
            }
        }
    }
}()

The duty that we run our whereas loop in received’t finish except we finish our stream from inside our catch block. If we manually shut the online socket connection utilizing the cancel technique we write earlier, the decision to obtain() won’t ever obtain an error nor a worth which signifies that will probably be caught without end.

Essentially the most dependable strategy to repair that is to return to the callback primarily based model of obtain to drive your async stream:

non-public lazy var stream: WebSocketStream = {
    return WebSocketStream { continuation in
        self.continuation = continuation
        waitForNextValue()
    }
}()

non-public func waitForNextValue() {
    guard job.closeCode == .invalid else {
        continuation?.end()
        return
    }

    job.obtain(completionHandler: { [weak self] lead to
        guard let continuation = self?.continuation else {
            return
        }

        do {
            let message = attempt end result.get()
            continuation.yield(message)
            self?.waitForNextValue()
        } catch {
            continuation.end(throwing: error)
        }
    })
}

With this strategy we don’t have any lingering duties, and our name website is as clear and concise as ever; we’ve solely modified a few of our inside logic.

In Abstract

Swift Concurrency offers many helpful options for writing higher code, and Apple rapidly adopted async / await for present APIs. Nevertheless, some APIs that might be helpful are lacking, reminiscent of iterating over internet socket messages.

On this submit, you discovered use async streams to create an async sequence that emits internet socket messages. You first noticed a completely async / await model that was neat, however had reminiscence and job lifecycle points. Then, you noticed a model that mixes a callback-based strategy with the async stream.

The result’s a simple strategy to iterate over incoming internet socket messages with async / await. In case you have any questions, feedback, or enhancements for this submit, please do not hesitate to succeed in out to me on Twitter.