Subscribe2 #226
Subscribe2 #226
Conversation
|
|
||
| while(true) { |
There was a problem hiding this comment.
I'm not sure where this while loop should actually live. 🤔
| if let conversation = conversationsByTopic[envelope.contentTopic] { | ||
| let decoded = try conversation.decode(envelope) | ||
| continuation.yield(decoded) | ||
| } else if envelope.contentTopic.hasPrefix("/xmtp/0/invite-") { | ||
| let conversation = try fromInvite(envelope: envelope) | ||
| conversationsByTopic[conversation.topic] = conversation | ||
| break // Break so we can resubscribe with the new conversation | ||
| topics.append(conversation.topic) | ||
| try await subscription.update(req: client.makeSubscribeRequest(topics: topics).toFFI) |
There was a problem hiding this comment.
Does this update method want the entire list of topics or just the new topics?
There was a problem hiding this comment.
This method is also blocking from the for loop continuing..
There was a problem hiding this comment.
It takes the entire list of topics
|
|
||
| func subscribe(topics: [String]) -> AsyncThrowingStream<Envelope, Error> { | ||
| func subscribe(topics: [String]) -> AsyncThrowingStream<(envelope: Envelope, subscription: LibXMTP.FfiV2Subscription), Error> { |
There was a problem hiding this comment.
A Touple probably isn't the best idea to solve this. But the simplest for now.
| do { | ||
| for try await envelope in client.subscribe(topics: topics) { | ||
| for try await (envelope, subscription) in client.subscribe(topics: topics) { |
There was a problem hiding this comment.
I tried a scenario where we returned the subscription independent of this for loop but the await function never completed. It may be my lack of knowledge of iOS threads/streams. But you can see that work in the previous commit history.
There was a problem hiding this comment.
@zombieobject I've got some iOS threading questions for you next week 🙏
There was a problem hiding this comment.
@nplasterer Thanks for looping me in, I'll have a look. 👨🏼🚒
|
@fabriguespe the tests definitely pass but I'm suspicious. Do you have an example you're running to test this? Can we try it on this branch? |
|
Okay I was able to reproduce that this isn't fixing the stream entirely. I've got an issue with subcription.update blocking the thread from continueing and when I put it in a task it doesn't actually update. You can see the failure in this test https://github.com/xmtp/xmtp-ios/blob/main/Tests/XMTPTests/IntegrationTests.swift#L248-L277 |
|
This may be a red herring. If everyone thinks this improvement is futile I'm happy to abandon this effort. |
|
@zombieobject and I paired on this today and decided that this probably isn't worth our time right now to fix. I think the real fix will likely be in the rust layer changing the current streaming interface to a callback interface like the V3 work does. Since this works fine for now and we have other high priorities going to pause this work. |
WIP
Improvements to subscribeAllMessages use the subscription to update the stream instead of breaking and resubscribing.
All the tests pass but I think we just don't have great testing around this.