Add more decrypted streaming

xmtp · Nov 27, 2023 · c489f1a · c489f1a
1 parent e205664
commit c489f1a
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 0 deletions.
diff --git a/Sources/XMTP/Conversation.swift b/Sources/XMTP/Conversation.swift
@@ -220,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
@@ -176,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)
 	}