Conversation as field in Message

data/src/main/java/mapper/CursorToMessageImpl.kt

@@ -24,6 +24,7 @@ import android.net.Uri
 import android.provider.Telephony.*
 import com.google.android.mms.pdu_alt.PduHeaders
 import manager.KeyManager
+import model.Conversation
 import model.Message
 import util.extensions.map
 import javax.inject.Inject
@@ -77,12 +78,13 @@ class CursorToMessageImpl @Inject constructor(
 
             id = keys.newId()
             contentId = cursor.getLong(columnsMap.msgId)
-            threadId = cursor.getLong(columnsMap.threadId)
             date = cursor.getLong(columnsMap.date)
             dateSent = cursor.getLong(columnsMap.dateSent)
             read = cursor.getInt(columnsMap.read) != 0
             locked = cursor.getInt(columnsMap.locked) != 0
 
+            conversation = Conversation().apply { id = cursor.getLong(columnsMap.threadId) }
+
             when (type) {
                 "sms" -> {
                     address = cursor.getString(columnsMap.smsAddress) ?: ""

data/src/main/java/repository/MessageRepositoryImpl.kt

@@ -34,7 +34,6 @@ import io.reactivex.Flowable
 import io.reactivex.Maybe
 import io.reactivex.Observable
 import io.reactivex.android.schedulers.AndroidSchedulers
-import io.reactivex.rxkotlin.Flowables
 import io.reactivex.schedulers.Schedulers
 import io.realm.Realm
 import io.realm.RealmResults
@@ -64,54 +63,24 @@ class MessageRepositoryImpl @Inject constructor(
         private val cursorToRecipient: CursorToRecipient,
         private val prefs: Preferences) : MessageRepository {
 
-    override fun getConversations(archived: Boolean): Flowable<List<Pair<Conversation, Message>>> {
-        val realm = Realm.getDefaultInstance()
-
-        val conversationFlowable = realm.where(Conversation::class.java)
-                .equalTo("archived", archived)
-                .equalTo("blocked", false)
-                .findAllAsync()
-                .asFlowable()
-                .filter { it.isLoaded }
-                .map { realm.copyFromRealm(it) }
-
-        val messageFlowable = realm.where(Message::class.java)
-                .sort("date", Sort.DESCENDING)
+    override fun getConversations(archived: Boolean): RealmResults<Message> {
+        return Realm.getDefaultInstance()
+                .where(Message::class.java)
                 .distinctValues("threadId")
+                .equalTo("conversation.archived", archived)
+                .equalTo("conversation.blocked", false)
+                .sort("date", Sort.DESCENDING)
                 .findAllAsync()
-                .asFlowable()
-                .filter { it.isLoaded }
-                .map { realm.copyFromRealm(it) }
-
-        return Flowables.combineLatest(conversationFlowable, messageFlowable, { conversations, messages ->
-            val conversationMap = conversations.associateBy { conversation -> conversation.id }
-
-            messages.mapNotNull { message ->
-                val conversation = conversationMap[message.threadId]
-                if (conversation == null) null else Pair(conversation, message)
-            }
-        })
     }
 
-    override fun getConversationsForWidget(): List<Pair<Conversation, Message>> {
+    override fun getConversationsForWidget(): List<Message> {
         val realm = Realm.getDefaultInstance()
-
-        val conversations = realm.copyFromRealm(realm.where(Conversation::class.java)
-                .equalTo("archived", false)
-                .equalTo("blocked", false)
-                .findAll())
-
-        val messages = realm.copyFromRealm(realm.where(Message::class.java)
-                .sort("date", Sort.DESCENDING)
+        return realm.copyFromRealm(realm.where(Message::class.java)
                 .distinctValues("threadId")
-                .findAll())
-
-        val conversationMap = conversations.associateBy { conversation -> conversation.id }
-
-        return messages.mapNotNull { message ->
-            val conversation = conversationMap[message.threadId]
-            if (conversation == null) null else Pair(conversation, message)
-        }
+                .equalTo("conversation.archived", false)
+                .equalTo("conversation.blocked", false)
+                .sort("date", Sort.DESCENDING)
+                .findAllAsync())
     }
 
     override fun getBlockedConversations(): Flowable<List<Conversation>> {
@@ -383,7 +352,6 @@ class MessageRepositoryImpl @Inject constructor(
 
         // Insert the message to Realm
         val message = Message().apply {
-            this.threadId = threadId
             this.address = address
             this.body = body
             this.date = System.currentTimeMillis()
@@ -393,10 +361,12 @@ class MessageRepositoryImpl @Inject constructor(
             type = "sms"
             read = true
             seen = true
+
+            conversation = getOrCreateConversation(threadId)
         }
         val realm = Realm.getDefaultInstance()
         var managedMessage: Message? = null
-        realm.executeTransaction { managedMessage = realm.copyToRealm(message) }
+        realm.executeTransaction { managedMessage = realm.copyToRealmOrUpdate(message) }
 
         // Insert the message to the native content provider
         val values = ContentValues().apply {
@@ -427,14 +397,14 @@ class MessageRepositoryImpl @Inject constructor(
             this.date = System.currentTimeMillis()
 
             id = messageIds.newId()
-            threadId = getOrCreateConversation(address).blockingGet().id
             boxId = Telephony.Sms.MESSAGE_TYPE_INBOX
             type = "sms"
 
+            conversation = getOrCreateConversation(address).blockingGet()
         }
         val realm = Realm.getDefaultInstance()
         var managedMessage: Message? = null
-        realm.executeTransaction { managedMessage = realm.copyToRealm(message) }
+        realm.executeTransaction { managedMessage = realm.copyToRealmOrUpdate(message) }
 
         // Insert the message to the native content provider
         val values = ContentValues().apply {

data/src/main/java/repository/SyncRepositoryImpl.kt

@@ -97,6 +97,17 @@ class SyncRepositoryImpl @Inject constructor(
         val lastSync = realm.where(Message::class.java)?.max("date")?.toLong() ?: 0
         realm.insert(SyncLog())
 
+
+        // Sync messages
+        val messageCursor = cursorToMessage.getMessagesCursor()
+        val messageColumns = CursorToMessage.MessageColumns(messageCursor)
+        val messages = messageCursor.mapWhile(
+                { cursor -> cursorToMessage.map(Pair(cursor, messageColumns)) },
+                { message -> message.date > lastSync })
+        realm.insertOrUpdate(messages)
+        messageCursor.close()
+
+
         // Sync conversations
         val conversationCursor = cursorToConversation.getConversationsCursor()
         val conversations = conversationCursor
@@ -112,16 +123,6 @@ class SyncRepositoryImpl @Inject constructor(
         conversationCursor.close()
 
 
-        // Sync messages
-        val messageCursor = cursorToMessage.getMessagesCursor()
-        val messageColumns = CursorToMessage.MessageColumns(messageCursor)
-        val messages = messageCursor.mapWhile(
-                { cursor -> cursorToMessage.map(Pair(cursor, messageColumns)) },
-                { message -> message.date > lastSync })
-        realm.insertOrUpdate(messages)
-        messageCursor.close()
-
-
         // Sync recipients
         val recipientCursor = cursorToRecipient.getRecipientCursor()
         val recipients = recipientCursor.map { cursor -> cursorToRecipient.map(cursor) }
@@ -167,11 +168,11 @@ class SyncRepositoryImpl @Inject constructor(
         // Map the cursor to a message
         return cursor.asFlowable()
                 .map { cursorToMessage.map(Pair(it, columnsMap)) }
-                .map { message ->
-                    existingId?.let { message.id = it }
-                    message
+                .doOnNext { message -> existingId?.let { message.id = it } }
+                .doOnNext { message ->
+                    val threadId = message.conversation?.id
+                    message.conversation = threadId?.let { messageRepo.getOrCreateConversation(threadId) }
                 }
-                .doOnNext { message -> messageRepo.getOrCreateConversation(message.threadId) }
                 .doOnNext { message -> message.insertOrUpdate() }
     }