Fix forked group bug (#350)

* reproduce failing test

* now reproduces in test

* get the tests passing

Package.swift

@@ -25,7 +25,7 @@ let package = Package(
 		.package(url: "https://github.com/1024jp/GzipSwift", from: "5.2.0"),
 		.package(url: "https://github.com/bufbuild/connect-swift", exact: "0.12.0"),
 		.package(url: "https://github.com/apple/swift-docc-plugin.git", from: "1.0.0"),
-		.package(url: "https://github.com/xmtp/libxmtp-swift", exact: "0.5.1-beta2"),
+		.package(url: "https://github.com/xmtp/libxmtp-swift", exact: "0.5.2-beta0"),
 	],
 	targets: [
 		// Targets are the basic building blocks of a package. A target can define a module or a test suite.

Sources/XMTPiOS/Conversations.swift

@@ -297,22 +297,24 @@ 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 MessageV3(client: self.client, ffiMessage: message).decode())
-							} catch {
-								print("Error onMessage \(error)")
-							}
-						}
-					)
-				} catch {
-					print("STREAM ERR: \(error)")
+				let messageCallback = MessageCallback(client: self.client) { message in
+					if let decodedMessage = MessageV3(client: self.client, ffiMessage: message).decodeOrNull() {
+						continuation.yield(decodedMessage)
+					}
+				}
+
+				guard let stream = try await client.v3Client?.conversations().streamAllMessages(messageCallback: messageCallback) else {
+					continuation.finish(throwing: GroupError.streamingFailure)
+					return
+				}
+
+				continuation.onTermination = { @Sendable reason in
+					stream.end()
 				}
 			}
 		}
 	}
+
 
 	public func streamAllMessages(includeGroups: Bool = false) async throws -> AsyncThrowingStream<DecodedMessage, Error> {
 		AsyncThrowingStream<DecodedMessage, Error> { continuation in

Tests/XMTPTests/GroupTests.swift

@@ -587,6 +587,54 @@ class GroupTests: XCTestCase {
 		await waitForExpectations(timeout: 3)
 	}
 
+	func testCanStreamAndUpdateNameWithoutForkingGroup() async throws {
+		let fixtures = try await localFixtures()
+
+		let expectation = expectation(description: "got a message")
+		expectation.expectedFulfillmentCount = 5
+
+		Task(priority: .userInitiated) {
+			for try await _ in try await fixtures.bobClient.conversations.streamAllGroupMessages(){
+				expectation.fulfill()
+			}
+		}
+
+		let alixGroup = try await fixtures.aliceClient.conversations.newGroup(with: [fixtures.bob.address])
+		try await alixGroup.updateGroupName(groupName: "hello")
+		try await alixGroup.send(content: "hello1")
+
+		try await fixtures.bobClient.conversations.sync()
+
+		let boGroups = try await fixtures.bobClient.conversations.groups()
+		XCTAssertEqual(boGroups.count, 1, "bo should have 1 group")
+		let boGroup = boGroups[0]
+		try await boGroup.sync()
+
+		let boMessages1 = try await boGroup.messages()
+		XCTAssertEqual(boMessages1.count, 2, "should have 2 messages on first load received \(boMessages1.count)")
+
+		try await boGroup.send(content: "hello2")
+		try await boGroup.send(content: "hello3")
+		try await alixGroup.sync()
+
+		let alixMessages = try await alixGroup.messages()
+		for message in alixMessages {
+			print("message", message.encodedContent.type, message.encodedContent.type.typeID)
+		}
+		XCTAssertEqual(alixMessages.count, 5, "should have 5 messages on first load received \(alixMessages.count)")
+
+		try await alixGroup.send(content: "hello4")
+		try await boGroup.sync()
+
+		let boMessages2 = try await boGroup.messages()
+		for message in boMessages2 {
+			print("message", message.encodedContent.type, message.encodedContent.type.typeID)
+		}
+		XCTAssertEqual(boMessages2.count, 5, "should have 5 messages on second load received \(boMessages2.count)")
+
+		await waitForExpectations(timeout: 3)
+	}
+
 	func testCanStreamAllMessages() async throws {
 		let fixtures = try await localFixtures()
 

XMTP.podspec

@@ -16,7 +16,7 @@ Pod::Spec.new do |spec|
   #
 
   spec.name         = "XMTP"
-  spec.version      = "0.12.0"
+  spec.version      = "0.12.1"
   spec.summary      = "XMTP SDK Cocoapod"
 
   # This description is used to generate tags and improve search results.
@@ -44,5 +44,5 @@ Pod::Spec.new do |spec|
   spec.dependency "web3.swift"
   spec.dependency "GzipSwift"
   spec.dependency "Connect-Swift", "= 0.12.0"
-  spec.dependency 'LibXMTP', '= 0.5.1-beta2'
+  spec.dependency 'LibXMTP', '= 0.5.2-beta0'
 end

XMTPiOSExample/XMTPiOSExample.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved

@@ -59,8 +59,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/xmtp/libxmtp-swift",
       "state" : {
-        "revision" : "61cf883b0a6e3bdd00a1d41f6b93556c35124b7b",
-        "version" : "0.5.1-beta2"
+        "revision" : "ef9421889bffd121d09e2c175314ccdcc1bc1d2e",
+        "version" : "0.5.2-beta0"
       }
     },
     {