Skip to content

fix: realtime reconnection #261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 16 commits into from
Mar 19, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions Examples/SlackClone/AppView.swift
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ final class AppViewModel {
var session: Session?
var selectedChannel: Channel?

var realtimeConnectionStatus: RealtimeClientV2.Status?

init() {
Task { [weak self] in
for await (event, session) in await supabase.auth.authStateChanges {
Expand All @@ -27,23 +29,46 @@ final class AppViewModel {
}
}
}

Task {
for await status in await supabase.realtimeV2.statusChange {
realtimeConnectionStatus = status
}
}
}
}

@MainActor
struct AppView: View {
@Bindable var model: AppViewModel
let log = LogStore.shared

@State var logPresented = false

@ViewBuilder
var body: some View {
if model.session != nil {
NavigationSplitView {
ChannelListView(channel: $model.selectedChannel)
.toolbar {
ToolbarItem {
Button("Log") {
logPresented = true
}
}
}
} detail: {
if let channel = model.selectedChannel {
MessagesView(channel: channel).id(channel.id)
}
}
.sheet(isPresented: $logPresented) {
List {
ForEach(0 ..< log.messages.count, id: \.self) { i in
Text(log.messages[i].description)
}
}
}
} else {
AuthView()
}
Expand Down
25 changes: 20 additions & 5 deletions Examples/SlackClone/Logger.swift
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,21 @@ extension Logger {
static let main = Self(subsystem: "com.supabase.SlackClone", category: "app")
}

final class SupabaseLoggerImpl: SupabaseLogger, @unchecked Sendable {
@Observable
final class LogStore: SupabaseLogger {
private let lock = NSLock()
private var loggers: [String: Logger] = [:]

static let shared = LogStore()

@MainActor
var messages: [SupabaseLogMessage] = []

func log(message: SupabaseLogMessage) {
Task {
await add(message: message)
}

lock.withLock {
if loggers[message.system] == nil {
loggers[message.system] = Logger(
Expand All @@ -29,11 +39,16 @@ final class SupabaseLoggerImpl: SupabaseLogger, @unchecked Sendable {
let logger = loggers[message.system]!

switch message.level {
case .debug: logger.debug("\(message)")
case .error: logger.error("\(message)")
case .verbose: logger.info("\(message)")
case .warning: logger.notice("\(message)")
case .debug: logger.debug("\(message, privacy: .public)")
case .error: logger.error("\(message, privacy: .public)")
case .verbose: logger.info("\(message, privacy: .public)")
case .warning: logger.notice("\(message, privacy: .public)")
}
}
}

@MainActor
private func add(message: SupabaseLogMessage) {
messages.insert(message, at: 0)
}
}
4 changes: 2 additions & 2 deletions Examples/SlackClone/Supabase.swift
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,10 @@ let decoder: JSONDecoder = {
}()

let supabase = SupabaseClient(
supabaseURL: URL(string: "http://127.0.0.1:54321")!,
supabaseURL: URL(string: "http://localhost:54321")!,
supabaseKey: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0",
options: SupabaseClientOptions(
db: .init(encoder: encoder, decoder: decoder),
global: SupabaseClientOptions.GlobalOptions(logger: SupabaseLoggerImpl())
global: SupabaseClientOptions.GlobalOptions(logger: LogStore.shared)
)
)
13 changes: 9 additions & 4 deletions Examples/SlackClone/UserStore.swift
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,17 @@ final class UserStore {
let channel = await supabase.realtimeV2.channel("public:users")
let changes = await channel.postgresChange(AnyAction.self, table: "users")

let prenseces = await channel.presenceChange()
let presences = await channel.presenceChange()

await channel.subscribe()

let userId = try await supabase.auth.session.user.id
try await channel.track(UserPresence(userId: userId, onlineAt: Date()))
Task {
let statusChange = await channel.statusChange
for await _ in statusChange.filter({ $0 == .subscribed }) {
let userId = try await supabase.auth.session.user.id
try await channel.track(UserPresence(userId: userId, onlineAt: Date()))
}
}

Task {
for await change in changes {
Expand All @@ -36,7 +41,7 @@ final class UserStore {
}

Task {
for await presence in prenseces {
for await presence in presences {
let joins = try presence.decodeJoins(as: UserPresence.self)
let leaves = try presence.decodeLeaves(as: UserPresence.self)

Expand Down
3 changes: 3 additions & 0 deletions Package.swift
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ let package = Package(
dependencies: [
"Auth",
"_Helpers",
"TestHelpers",
.product(name: "SnapshotTesting", package: "swift-snapshot-testing"),
.product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
],
Expand Down Expand Up @@ -115,6 +116,7 @@ let package = Package(
name: "RealtimeTests",
dependencies: [
"Realtime",
"TestHelpers",
.product(name: "CustomDump", package: "swift-custom-dump"),
]
),
Expand All @@ -132,6 +134,7 @@ let package = Package(
]
),
.testTarget(name: "SupabaseTests", dependencies: ["Supabase"]),
.target(name: "TestHelpers"),
]
)

Expand Down
10 changes: 5 additions & 5 deletions Sources/Auth/AuthClient.swift
Original file line number Diff line number Diff line change
Expand Up @@ -199,9 +199,9 @@ public actor AuthClient {
public func onAuthStateChange(
_ listener: @escaping AuthStateChangeListener
) async -> some AuthStateChangeListenerRegistration {
let handle = eventEmitter.attachListener(listener)
await emitInitialSession(forHandle: handle)
return handle
let token = eventEmitter.attachListener(listener)
await emitInitialSession(forToken: token)
return token
}

/// Listen for auth state changes.
Expand Down Expand Up @@ -908,9 +908,9 @@ public actor AuthClient {
return session
}

private func emitInitialSession(forHandle handle: AuthStateChangeListenerHandle) async {
private func emitInitialSession(forToken token: ObservationToken) async {
let session = try? await session
eventEmitter.emit(.initialSession, session: session, handle: handle)
eventEmitter.emit(.initialSession, session: session, token: token)
}

private func prepareForPKCE() -> (codeChallenge: String?, codeChallengeMethod: String?) {
Expand Down
22 changes: 3 additions & 19 deletions Sources/Auth/AuthStateChangeListener.swift
Original file line number Diff line number Diff line change
Expand Up @@ -5,35 +5,19 @@
// Created by Guilherme Souza on 17/02/24.
//

import _Helpers
import ConcurrencyExtras
import Foundation

/// A listener that can be removed by calling ``AuthStateChangeListenerRegistration/remove()``.
///
/// - Note: Listener is automatically removed on deinit.
public protocol AuthStateChangeListenerRegistration: Sendable, AnyObject {
public protocol AuthStateChangeListenerRegistration: Sendable {
/// Removes the listener. After the initial call, subsequent calls have no effect.
func remove()
}

final class AuthStateChangeListenerHandle: AuthStateChangeListenerRegistration {
let _onRemove = LockIsolated((@Sendable () -> Void)?.none)

public func remove() {
_onRemove.withValue {
if $0 == nil {
return
}

$0?()
$0 = nil
}
}

deinit {
remove()
}
}
extension ObservationToken: AuthStateChangeListenerRegistration {}

public typealias AuthStateChangeListener = @Sendable (
_ event: AuthChangeEvent,
Expand Down
42 changes: 14 additions & 28 deletions Sources/Auth/Internal/EventEmitter.swift
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
import ConcurrencyExtras
import Foundation
@_spi(Internal) import _Helpers

protocol EventEmitter: Sendable {
func attachListener(
_ listener: @escaping AuthStateChangeListener
) -> AuthStateChangeListenerHandle
) -> ObservationToken

func emit(
_ event: AuthChangeEvent,
session: Session?,
handle: AuthStateChangeListenerHandle?
token: ObservationToken?
)
}

Expand All @@ -18,7 +19,7 @@ extension EventEmitter {
_ event: AuthChangeEvent,
session: Session?
) {
emit(event, session: session, handle: nil)
emit(event, session: session, token: nil)
}
}

Expand All @@ -27,31 +28,24 @@ final class DefaultEventEmitter: EventEmitter {

private init() {}

let listeners = LockIsolated<[ObjectIdentifier: AuthStateChangeListener]>([:])
let emitter = _Helpers.EventEmitter<(AuthChangeEvent, Session?)?>(
initialEvent: nil,
emitsLastEventWhenAttaching: false
)

func attachListener(
_ listener: @escaping AuthStateChangeListener
) -> AuthStateChangeListenerHandle {
let handle = AuthStateChangeListenerHandle()
let key = ObjectIdentifier(handle)

handle._onRemove.setValue { [weak self] in
self?.listeners.withValue {
$0[key] = nil
}
}

listeners.withValue {
$0[key] = listener
) -> ObservationToken {
emitter.attach { event in
guard let event else { return }
listener(event.0, event.1)
}

return handle
}

func emit(
_ event: AuthChangeEvent,
session: Session?,
handle: AuthStateChangeListenerHandle? = nil
token: ObservationToken? = nil
) {
NotificationCenter.default.post(
name: AuthClient.didChangeAuthStateNotification,
Expand All @@ -62,14 +56,6 @@ final class DefaultEventEmitter: EventEmitter {
]
)

let listeners = listeners.value

if let handle {
listeners[ObjectIdentifier(handle)]?(event, session)
} else {
for listener in listeners.values {
listener(event, session)
}
}
emitter.emit((event, session), to: token)
}
}
2 changes: 1 addition & 1 deletion Sources/Realtime/V2/PushV2.swift
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ actor PushV2 {

func send() async -> PushStatus {
do {
try await channel?.socket?.ws?.send(message)
try await channel?.socket?.ws.send(message)

if channel?.config.broadcast.acknowledgeBroadcasts == true {
return await withCheckedContinuation {
Expand Down
9 changes: 5 additions & 4 deletions Sources/Realtime/V2/RealtimeChannelV2.swift
Original file line number Diff line number Diff line change
Expand Up @@ -33,19 +33,20 @@ public actor RealtimeChannelV2 {
let logger: (any SupabaseLogger)?

private let callbackManager = CallbackManager()
private let statusStream = SharedStream<Status>(initialElement: .unsubscribed)

private let statusEventEmitter = EventEmitter<Status>(initialEvent: .unsubscribed)

private var clientChanges: [PostgresJoinConfig] = []
private var joinRef: String?
private var pushes: [String: PushV2] = [:]

public private(set) var status: Status {
get { statusStream.lastElement }
set { statusStream.yield(newValue) }
get { statusEventEmitter.lastEvent.value }
set { statusEventEmitter.emit(newValue) }
}

public var statusChange: AsyncStream<Status> {
statusStream.makeStream()
statusEventEmitter.stream()
}

init(
Expand Down
Loading