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

Show file tree
Hide file tree
Changes from all commits
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 {
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 {
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
68 changes: 64 additions & 4 deletions Tests/XMTPTests/GroupTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,6 +394,7 @@ class GroupTests: XCTestCase {
let fixtures = try await localFixtures()

let expectation1 = expectation(description: "got a conversation")
expectation1.expectedFulfillmentCount = 2

Task(priority: .userInitiated) {
for try await _ in try await fixtures.aliceClient.conversations.streamAll() {
Expand All @@ -402,23 +403,82 @@ class GroupTests: XCTestCase {
}

_ = try await fixtures.bobClient.conversations.newGroup(with: [fixtures.alice.address])
// _ = try await fixtures.bobClient.conversations.newConversation(with: fixtures.alice.address)
_ = try await fixtures.bobClient.conversations.newConversation(with: fixtures.alice.address)

await waitForExpectations(timeout: 3)
}

func testCanStreamGroupsAndConversationsWorksConvos() async throws {
func testCanStreamAllMessages() async throws {
let fixtures = try await localFixtures()

let expectation1 = expectation(description: "got a conversation")
expectation1.expectedFulfillmentCount = 2
let convo = try await fixtures.bobClient.conversations.newConversation(with: fixtures.alice.address)
let group = try await fixtures.bobClient.conversations.newGroup(with: [fixtures.alice.address])
try await fixtures.aliceClient.conversations.sync()
Task(priority: .userInitiated) {
for try await _ in try await fixtures.aliceClient.conversations.streamAllMessages(includeGroups: true) {
expectation1.fulfill()
}
}

try await group.send(content: "hi")
try await convo.send(content: "hi")

await waitForExpectations(timeout: 3)
}

func testCanStreamAllDecryptedMessages() async throws {
let fixtures = try await localFixtures()

let expectation1 = expectation(description: "got a conversation")
expectation1.expectedFulfillmentCount = 2
let convo = try await fixtures.bobClient.conversations.newConversation(with: fixtures.alice.address)
let group = try await fixtures.bobClient.conversations.newGroup(with: [fixtures.alice.address])
try await fixtures.aliceClient.conversations.sync()
Task(priority: .userInitiated) {
for try await _ in try await fixtures.aliceClient.conversations.streamAll() {
for try await _ in try await fixtures.aliceClient.conversations.streamAllDecryptedMessages(includeGroups: true) {
expectation1.fulfill()
}
}

_ = try await fixtures.bobClient.conversations.newConversation(with: fixtures.alice.address)
try await group.send(content: "hi")
try await convo.send(content: "hi")

await waitForExpectations(timeout: 3)
}

func testCanStreamAllGroupMessages() async throws {
let fixtures = try await localFixtures()

let expectation1 = expectation(description: "got a conversation")

let group = try await fixtures.bobClient.conversations.newGroup(with: [fixtures.alice.address])
try await fixtures.aliceClient.conversations.sync()
Task(priority: .userInitiated) {
for try await _ in try await fixtures.aliceClient.conversations.streamAllGroupMessages() {
expectation1.fulfill()
}
}

try await group.send(content: "hi")

await waitForExpectations(timeout: 3)
}

func testCanStreamAllGroupDecryptedMessages() async throws {
let fixtures = try await localFixtures()

let expectation1 = expectation(description: "got a conversation")
let group = try await fixtures.bobClient.conversations.newGroup(with: [fixtures.alice.address])
try await fixtures.aliceClient.conversations.sync()
Task(priority: .userInitiated) {
for try await _ in try await fixtures.aliceClient.conversations.streamAllGroupDecryptedMessages() {
expectation1.fulfill()
}
}

try await group.send(content: "hi")

await waitForExpectations(timeout: 3)
}
Expand Down
2 changes: 1 addition & 1 deletion XMTP.podspec
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ Pod::Spec.new do |spec|
#

spec.name = "XMTP"
spec.version = "0.8.9"
spec.version = "0.8.10"
spec.summary = "XMTP SDK Cocoapod"

# This description is used to generate tags and improve search results.
Expand Down
Loading