Skip to content
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.

Sign up for GitHub

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

Add Stream All group messages #258

Merged
merged 3 commits into from
Feb 21, 2024
Merged

Add Stream All group messages #258

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d1de051
add all the stream all messaging methods
nplasterer Feb 21, 2024
128e258
bump the pod spec
nplasterer Feb 21, 2024
9c95d16
add tests for all the new streaming methods
nplasterer Feb 21, 2024
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
95 changes: 93 additions & 2 deletions Sources/XMTPiOS/Conversations.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ public actor Conversations {
return messages
}

public func streamAllMessages() async throws -> AsyncThrowingStream<DecodedMessage, Error> {
func streamAllV2Messages() async throws -> AsyncThrowingStream<DecodedMessage, Error> {
return AsyncThrowingStream { continuation in
Task {
while true {
Expand Down Expand Up @@ -278,8 +278,99 @@ public actor Conversations {
}
}
}

public func streamAllGroupMessages() -> AsyncThrowingStream<DecodedMessage, Error> {
AsyncThrowingStream { continuation in
Task.detached {
do {
self.streamHolder.stream = try await self.client.v3Client?.conversations().streamAllMessages(
messageCallback: MessageCallback(client: self.client) { message in
do {
continuation.yield(try message.fromFFI(client: self.client))
} catch {
print("Error onMessage \(error)")
}
}
)
} catch {
print("STREAM ERR: \(error)")
}
}
}
}

public func streamAllMessages(includeGroups: Bool = false) async throws -> AsyncThrowingStream<DecodedMessage, Error> {
AsyncThrowingStream<DecodedMessage, Error> { continuation in
@Sendable func forwardStreamToMerged(stream: AsyncThrowingStream<DecodedMessage, Error>) async {
do {
var iterator = stream.makeAsyncIterator()
while let element = try await iterator.next() {
continuation.yield(element)
}
continuation.finish()
} catch {
continuation.finish(throwing: error)
}
}

Task {
await forwardStreamToMerged(stream: try streamAllV2Messages())
}
if (includeGroups) {
Task {
await forwardStreamToMerged(stream: streamAllGroupMessages())
}
}
}
}

public func streamAllGroupDecryptedMessages() -> AsyncThrowingStream<DecryptedMessage, Error> {
AsyncThrowingStream { continuation in
Task.detached {
do {
self.streamHolder.stream = try await self.client.v3Client?.conversations().streamAllMessages(
messageCallback: MessageCallback(client: self.client) { message in
do {
continuation.yield(try message.fromFFIDecrypted(client: self.client))
} catch {
print("Error onMessage \(error)")
}
}
)
} catch {
print("STREAM ERR: \(error)")
}
}
}
}

public func streamAllDecryptedMessages(includeGroups: Bool = false) -> AsyncThrowingStream<DecryptedMessage, Error> {
AsyncThrowingStream<DecryptedMessage, Error> { continuation in
@Sendable func forwardStreamToMerged(stream: AsyncThrowingStream<DecryptedMessage, Error>) async {
do {
var iterator = stream.makeAsyncIterator()
while let element = try await iterator.next() {
continuation.yield(element)
}
continuation.finish()
} catch {
continuation.finish(throwing: error)
}
}

Task {
await forwardStreamToMerged(stream: try streamAllV2DecryptedMessages())
}
if (includeGroups) {
Task {
await forwardStreamToMerged(stream: streamAllGroupDecryptedMessages())
}
}
}
}


public func streamAllDecryptedMessages() async throws -> AsyncThrowingStream<DecryptedMessage, Error> {
func streamAllV2DecryptedMessages() async throws -> AsyncThrowingStream<DecryptedMessage, Error> {
return AsyncThrowingStream { continuation in
Task {
while true {
Expand Down
Loading