🏎 Implement `knownEventLoop` to fix race condition

There was race condition on mutation operations e.g. adding or removing channels which causes fatalError randomly on highload.
Now it is fixed by calling any mutation operations on known eventLoop.

BREAKING CHANGES

now subscribe and unsubscribe methods ends with on eventLoop: EventLoop)

BaseObserver: `clients` variable fix.

Now broadcast works as expected.

2.0 for Vapor 4 🚀

Receive & send websocket messages through convenient observers. Even multiple observers on different endpoints!

💡Types of observers: Classic, Declarative, Bindable. Read about all of them below.

Built for Vapor4.

How it works ?

Declarative observer

WS lib have .default WSID which represents DeclarativeObserver.

💡You can declare your own WSID with another type of observer and your custom class.

You can start working with it this easy way

app.ws.build(.default).serve()

In this case it will start listening for websocket connections at /, but you can change it before you call .serve()

app.ws.build(.default).at("ws").serve()

Ok now it is listening at /ws

Also you can protect your websocket endpoint with middlewares, e.g. you can check auth before connection will be established.

app.ws.build(.default).at("ws").middlewares(AuthMiddleware()).serve()

Ok, looks good, but how to handle incoming data?

As we use .default WSID which represents Declarative observer we can handle incoming data like this

app.ws.build(.default).at("ws").middlewares(AuthMiddleware()).serve().onOpen { client in
    print("client just connected \(client.id)")
}.onText { client, text in
    print("client \(client.id) text: \(text)")
}

there are also available: onClose, onPing, onPong, onBinary, onByteBuffer handlers.

💡Set app.logger.logLevel = .info or app.logger.logLevel = .debug to see more info about connections

Classic observer

You should create new class which inherit from ClassicObserver

import WS

class MyClassicWebSocket: ClassicObserver {
    override func on(open client: AnyClient) {}
    override func on(close client: AnyClient) {}

    override func on(text: String, client: AnyClient) {}
    /// also you can override: `on(ping:)`, `on(pong:)`, `on(binary:)`, `on(byteBuffer:)`
}

and you must declare a WSID for it

extension WSID {
    static var myClassic: WSID<MyClassicWebSocket> { .init() }
}

so then start serving it

app.ws.build(.myClassic).at("ws").serve()

Bindable observer

This kind of observer designed to send and receive events in special format, e.g. in JSON:

{ "event": "<event name>", "payload": <anything> }

or just

{ "event": "<event name>" }

💡By default lib uses JSONEncoder and JSONDecoder, but you can replace them with anything else in setup method.

First of all declare any possible events in EID extension like this

struct Hello: Codable {
  let firstName, lastName: String
}
struct Bye: Codable {
  let firstName, lastName: String
}
extension EID {
    static var hello: EID<Hello> { .init("hello") }
    static var bye: EID<Bye> { .init("bye") }
    // Use `EID<Nothing>` if you don't want any payload
}

Then create your custom bindable observer class

class MyBindableWebsocket: BindableObserver {
    // register all EIDs here
    override func setup() {
        bind(.hello, hello)
        bind(.bye, bye)
        // optionally setup here custom encoder/decoder
        encoder = JSONEncoder() // e.g. with custom `dateEncodingStrategy`
        decoder = JSONDecoder() // e.g. with custom `dateDecodingStrategy`
    }

    // hello EID handler
    func hello(client: AnyClient, payload: Hello) {
        print("Hello \(payload.firstName) \(payload.lastName)")
    }

    // bye EID handler
    func bye(client: AnyClient, payload: Bye) {
        print("Bye \(payload.firstName) \(payload.lastName)")
    }
}

declare a WSID

extension WSID {
    static var myBindable: WSID<MyBindableWebsocket> { .init() }
}

then start serving it

app.ws.build(.myBindable).at("ws").serve()

💡Here you also could provide custom encoder/decoder
e,g, app.ws.build(.myBindable).at("ws").encoder(JSONEncoder()).encoder(JSONDecoder()).serve()

How to send data

Data sending works through Sendable protocol, which have several methods

.send(text: <StringProtocol>) // send message with text
.send(bytes: <[UInt8]>) // send message with bytes
.send(data: <Data>) // send message with binary data
.send(model: <Encodable>) // send message with Encodable model
.send(model: <Encodable>, encoder: Encoder)
.send(event: <EID>) // send bindable event
.send(event: <EID>, payload: T?)

all these methods returns EventLoopFuture<Void>

Using methods listed above you could send messages to one or multiple clients.

To one client e.g. in on(open:) or on(text:)

client.send(...)

To all clients

client.broadcast.send(...)
client.broadcast.exclude(client).send(...) // excluding himself
req.ws(.mywsid).broadcast.send(...)

To clients in channels

client.broadcast.channels("news", "updates").send(...)
req.ws(.mywsid).broadcast.channels("news", "updates").send(...)

To custom filtered clients

e.g. you want to find all ws connections of the current user to send a message to all his devices

req.ws(.mywsid).broadcast.filter { client in
    req.headers[.authorization].first == client.originalRequest.headers[.authorization].first
}.send(...)

Broadcast

You could reach broadcast obejct on app.ws.observer(.mywsid) or req.ws(.mywsid).broadcast or client.broadcast.

This object is a builder, so using it you should filter recipients like this client.broadcast.one(...).two(...).three(...).send()

Available methods

.encoder(Encoder) // set custom data encoder
.exclude([AnyClient]) // exclude provided clients from clients
.filter((AnyClient) -> Bool) // filter clients by closure result
.channels([String]) // filter clients by provided channels
.subscribe([String]) // subscribe filtered clients to channels
.unsubscribe([String]) // unsubscribe filtered clients from channels
.disconnect() // disconnect filtered clients
.send(...) // send message to filtered clients
.count // number of filtered clients

Channels

Subscribe

client.subscribe(to: ...) // will subscribe client to provided channels

To subscribe to news and updates call it like this client.subscribe(to: "news", "updates")

Unsubscribe

client.unsubscribe(from: ...) // will unsubscribe client from provided channels

List

client.channels // will return a list of client channels

Defaults

If you have only one observer in the app you can set it as default. It will give you ability to use it without providing its WSID all the time, so you will call just req.ws() instead of req.ws(.mywsid).

// configure.swift

app.ws.setDefault(.myBindable)

Also you can set custom encoder/decoder for all the observers

// configure.swift

let encoder = JSONEncoder()
encoder.dateEncodingStrategy = .secondsSince1970
app.ws.encoder = encoder

let decoder = JSONDecoder()
decoder.dateDecodingStrategy = .secondsSince1970
app.ws.decoder = decoder

Client

As you may see in every handler you always have client object. This object conforms to AnyClient protocol which contains useful things inside

variables

• id - UUID
• originalRequest - original Request
• eventLoop - next EventLoop
• application - pointer to Application
• channels - an array of channels that client subscribed to
• logger - pointer to Logger
• observer - this client's observer
• sockets - original socket connection of the client
• exchangeMode - client's observer exchange mode

conformanses

• Sendable - so you can use .send(...)
• Subscribable - so you can use .subscribe(...), .unsubscribe(...)
• Disconnectable - so you can call .disconnect() to disconnect that user

Original request gives you ability to e.g. determine connected user:

let user = try client.originalRequest.requireAuthenticated(User.self)