Subscriptions

struct Resolver {
    // Example using an async stream generator
    func hello(_: Context, _: NoArguments) async -> EventStream<String> {
        let stream = AsyncStream { con in
            for word in ["Hello", "Bonjour", "Ciao"] {
                con.yield(word)
            }
            con.finish()
        }
        return stream.toEventStream()
    }
 
    // Example using pubsub
    func hello(_: Context, _: NoArguments) async -> EventStream<String> {
        pubsub.asyncStream(String.self, for: "HELLO_MESSAGE").toEventStrem()
    }
}

let eventStream: EventStream<Int> = AsyncEventStream(from: MyIntAsyncSequence())

let eventStream = AsyncStream<Int>(...)
    .toEventStream()
 
// Initial value before any stream values
let eventStream1 = AsyncStream<Int>(...)
    .toEventStream(initialValue: 0)
 
// End value after stream finishes (excluding termination and value is lazily loaded; hence the function there)
let eventStream2 = AsyncStream<Int>(...)
    .toEventStream(initialValue: 0, endValue: { 10 })

let sequence = MyAsyncSequence()
 
let eventStream = sequence.toEventStream(
    onTermination: { termination in
        if case .cancelled = termination {
            sequence.cancel()
        }
    }
)

import struct Pioneer.AsyncPubSub
 
let pubsub = AsyncPubSub()

await pubsub.publish(
    for: "POST_CREATED", 
    payload: Post(by: "Jeff Bezos", content: "How much money do I have?")
)

struct Resolver {
    func createPost(ctx: Context, args: CreatePostArgs) async throws -> Post {
        let post = Post(args: args)
        try await post.create(on: ctx.req.db)
        return post
    }
}

struct Resolver {
    func createPost(ctx: Context, args: CreatePostArgs) async throws -> Post {
        let post = Post(args: args)
        try await post.create(on: ctx.req.db)
        await pubsub.publish(for: "POST_CREATED", post)
        return post
    }
}

pubsub.asyncStream(Post.self, for: "POST_CREATED");

struct Resolver {
    func postCreated(ctx: Context, _: NoArguments) async -> EventStream<Post> {
        ctx.pubsub.asyncStream(Post.self, for: "POST_CREATED").toEventStrem()
    }
}

import Foundation
import NIOFoundationCompat
import class Pioneer.Broadcast
import struct Redis.RedisChannelName
import protocol Pioneer.PubSub
import protocol Redis.RedisClient
 
struct RedisPubSub: PubSub {
 
    // MARK: - Actor for distribution
    actor Dispatcher {
        private let redis: RedisClient
        private var broadcasting: [String: Broadcast<Data>] = [:]
 
        init(_ redis: RedisClient) {
            self.redis = redis
        }
 
        /// Get a downstream from the broadcast for the channel given
        func downstream(to channel: String) async throws -> AsyncStream<Data> {
            let broadcast = try await subscribe(to: channel)
            let downstream = await broadcast.downstream()
            return downstream.stream
        }
 
        /// Get the broadcast for the channel if exist, otherwise make a new one
        private func subscribe(to channel: String) async throws -> Broadcast<Data> {
            if let broadcast = broadcasting[channel] {
                return broadcast
            }
            let broadcast = Broadcast<Data>()
            broadcasting[channel] = broadcast
            try await apply(from: .init(channel), to: broadcast)
            return broadcast
        }
 
        /// Apply broadcasting to the Redis channel subscription
        private func apply(from channel: RedisChannelName, to broadcast: Broadcast<Data>) async throws {
            do {
                try await redis.subscribe(
                    to: channel,
                    messageReceiver: { _, msg in
                        guard case .bulkString(.some(let buffer)) = msg else { return }
                        let data = Data(buffer: buffer)
                        Task {
                            await broadcast.publish(data)
                        }
                    },
                    onUnsubscribe: { _, _ in
                        Task {
                            await broadcast.close()
                        }
                    }
                )
                .get()
            } catch {
              await broadcast.close()
              throw error
            }
        }
 
        /// Pubblish the data (which is RESPValueConvertible) to the specific redis channel
        func publish(for channel: String, _ value: Data) async throws {
            let _ = try await redis.publish(value, to: .init(channel)).get()
        }
 
        /// Close the redis channel subscription and all of the downstreams
        func close(for channel: String) async throws {
            try await redis.unsubscribe(from: .init(channel)).get()
            await broadcasting[channel]?.close()
        }
    }
 
    // MARK: -- Protocol required methods
 
    public func asyncStream<DataType: Sendable & Decodable>(_ type: DataType.Type = DataType.self, for trigger: String) -> AsyncThrowingStream<DataType, Error> {
        AsyncThrowingStream<DataType, Error> { con in
            let task = Task {
                do {
                    let stream = try await dispatcher.downstream(to: trigger)
                    for await data in stream {
                        do {
                            let event = try JSONDecoder().decode(DataType.self, data)
                            con.yield(event)
                        } catch {
                            con.finish(throwing: error)
                        }
                    }
                    con.finish()
                } catch {
                    con.finish(throwing: error)
                }
            }
            con.onTermination = { @Sendable _ in
                task.cancel()
            }
        }
    }
 
    public func publish<DataType: Sendable & Encodable>(for trigger: String, payload: DataType) async throws {
        let data = try JSONEncoder().encode(payload)
        try await dispatcher.publish(for: trigger, data)
    }
 
    public func close(for trigger: String) async throws {
        try await dispatcher.close(for: trigger)
    }
 
    // MARK: - Properties
 
    private let dispatcher: Dispatcher
 
    public init(_ redis: RedisClient) {
        self.dispatcher = .init(redis)
    }
}

struct Message: Sendable, Codable { ... }

struct Resolver {
    let pubsub: PubSub = app.environment.isRelease ? RedisPubSub(app.redis) : AsyncPubSub()
 
    func create(ctx: Context, _: NoArguments) async throws -> Message {
        let message = ...
        try await pubsub.publish(message)
        return message
    }
 
    func onCreate(ctx: Context, _: NoArguments) async -> EventStream<Message> {
        pubsub.asyncStream(Message.self, for: "message-create").toEventStream()
    }
}