Merge branch 'kele/static-can-message' of https://github.com/xmtp/xmt…

…p-ios into kele/static-can-message
xmtp · Nov 29, 2023 · 3240be6 · 3240be6
2 parents 6393110 + 77324ec
commit 3240be6
Show file tree

Hide file tree

Showing 8 changed files with 495 additions and 345 deletions.
diff --git a/Sources/XMTP/Conversation.swift b/Sources/XMTP/Conversation.swift
@@ -118,6 +118,15 @@ public enum Conversation: Sendable {
 		}
 	}
 
+	public func decrypt(_ envelope: Envelope) throws -> DecryptedMessage {
+		switch self {
+		case let .v1(conversationV1):
+			return try conversationV1.decrypt(envelope: envelope)
+		case let .v2(conversationV2):
+			return try conversationV2.decrypt(envelope: envelope)
+		}
+	}
+
 	public func encode<Codec: ContentCodec, T>(codec: Codec, content: T) async throws -> Data where Codec.T == T {
 		switch self {
 		case let .v1:
@@ -211,6 +220,15 @@ public enum Conversation: Sendable {
 		}
 	}
 
+	public func streamDecryptedMessages() -> AsyncThrowingStream<DecryptedMessage, Error> {
+		switch self {
+		case let .v1(conversation):
+			return conversation.streamDecryptedMessages()
+		case let .v2(conversation):
+			return conversation.streamDecryptedMessages()
+		}
+	}
+
 	/// List messages in the conversation
 	public func messages(limit: Int? = nil, before: Date? = nil, after: Date? = nil, direction: PagingInfoSortDirection? = .descending) async throws -> [DecodedMessage] {
 		switch self {

diff --git a/Sources/XMTP/ConversationV1.swift b/Sources/XMTP/ConversationV1.swift
@@ -162,6 +162,17 @@ public struct ConversationV1 {
 		}
 	}
 
+	public func streamDecryptedMessages() -> AsyncThrowingStream<DecryptedMessage, Error> {
+		AsyncThrowingStream { continuation in
+			Task {
+				for try await envelope in client.subscribe(topics: [topic.description]) {
+					let decoded = try decrypt(envelope: envelope)
+					continuation.yield(decoded)
+				}
+			}
+		}
+	}
+
 	var ephemeralTopic: String {
 		topic.description.replacingOccurrences(of: "/xmtp/0/dm-", with: "/xmtp/0/dmE-")
 	}

diff --git a/Sources/XMTP/ConversationV2.swift b/Sources/XMTP/ConversationV2.swift
@@ -137,11 +137,15 @@ public struct ConversationV2 {
 		let envelopes = try await client.apiClient.envelopes(topic: topic.description, pagination: pagination)
 
 		return try envelopes.map { envelope in
-			let message = try Message(serializedData: envelope.message)
-			return try MessageV2.decrypt(generateID(from: envelope), topic, message.v2, keyMaterial: keyMaterial, client: client)
+			try decrypt(envelope: envelope)
 		}
 	}
 
+	func decrypt(envelope: Envelope) throws -> DecryptedMessage {
+		let message = try Message(serializedData: envelope.message)
+		return try MessageV2.decrypt(generateID(from: envelope), topic, message.v2, keyMaterial: keyMaterial, client: client)
+	}
+
 	var ephemeralTopic: String {
 		topic.replacingOccurrences(of: "/xmtp/0/m", with: "/xmtp/0/mE")
 	}
@@ -172,6 +176,18 @@ public struct ConversationV2 {
 		}
 	}
 
+	public func streamDecryptedMessages() -> AsyncThrowingStream<DecryptedMessage, Error> {
+		AsyncThrowingStream { continuation in
+			Task {
+				for try await envelope in client.subscribe(topics: [topic.description]) {
+					let decoded = try decrypt(envelope: envelope)
+
+					continuation.yield(decoded)
+				}
+			}
+		}
+	}
+
 	public var createdAt: Date {
 		Date(timeIntervalSince1970: Double(header.createdNs / 1_000_000) / 1000)
 	}