WIP lazy parse events in sync

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/crosssigning/UpdateTrustWorker.kt

@@ -130,7 +130,7 @@ internal class UpdateTrustWorker(context: Context,
                         myUserId -> myTrustResult
                         else     -> {
                             crossSigningService.checkOtherMSKTrusted(myCrossSigningInfo, infoEntry.value).also {
-                                Timber.d("## CrossSigning - user:${infoEntry.key} result:$it")
+                                Timber.d("## CrossSigning - user:${infoEntry.key} result:${it.isVerified()}")
                             }
                         }
                     }

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/di/MoshiProvider.kt

@@ -36,6 +36,9 @@ import org.matrix.android.sdk.internal.network.parsing.ForceToBooleanJsonAdapter
 import org.matrix.android.sdk.internal.network.parsing.RuntimeJsonAdapterFactory
 import org.matrix.android.sdk.internal.network.parsing.TlsVersionMoshiAdapter
 import org.matrix.android.sdk.internal.network.parsing.UriMoshiAdapter
+import org.matrix.android.sdk.internal.session.sync.model.RoomSyncEphemeralLazyJsonAdapter
+import org.matrix.android.sdk.internal.session.sync.model.RoomSyncTimeLineLazyJsonAdapter
+import org.matrix.android.sdk.internal.session.sync.model.RoomSyncTimeline
 
 object MoshiProvider {
 
@@ -44,6 +47,8 @@ object MoshiProvider {
             .add(ForceToBooleanJsonAdapter())
             .add(CipherSuiteMoshiAdapter())
             .add(TlsVersionMoshiAdapter())
+            .add(RoomSyncTimeLineLazyJsonAdapter(Moshi.Builder().build()))
+            .add(RoomSyncEphemeralLazyJsonAdapter(Moshi.Builder().build()))
             .add(RuntimeJsonAdapterFactory.of(MessageContent::class.java, "msgtype", MessageDefaultContent::class.java)
                     .registerSubtype(MessageTextContent::class.java, MessageType.MSGTYPE_TEXT)
                     .registerSubtype(MessageNoticeContent::class.java, MessageType.MSGTYPE_NOTICE)

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/notification/ProcessEventForPushTask.kt

@@ -48,11 +48,11 @@ internal class DefaultProcessEventForPushTask @Inject constructor(
         params.syncResponse.join.keys.forEach {
             defaultPushRuleService.dispatchRoomJoined(it)
         }
-        val newJoinEvents = params.syncResponse.join
+        val newJoinEvents = params.syncResponse.join.asSequence()
                 .mapNotNull { (key, value) ->
                     value.timeline?.events?.map { it.copy(roomId = key) }
-                }
-                .flatten()
+                }.flatten()
+
         val inviteEvents = params.syncResponse.invite
                 .mapNotNull { (key, value) ->
                     value.inviteState?.events?.map { it.copy(roomId = key) }
@@ -69,7 +69,7 @@ internal class DefaultProcessEventForPushTask @Inject constructor(
         }.filter {
             it.senderId != userId
         }
-        Timber.v("[PushRules] Found ${allEvents.size} out of ${(newJoinEvents + inviteEvents).size}" +
+        Timber.v("[PushRules] Found out of}" +
                 " to check for push rules with ${params.rules.size} rules")
         allEvents.forEach { event ->
             fulfilledBingRule(event, params.rules)?.let {

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/RoomSyncHandler.kt

@@ -61,7 +61,6 @@ import org.matrix.android.sdk.internal.session.room.typing.TypingEventContent
 import org.matrix.android.sdk.internal.session.sync.model.InvitedRoomSync
 import org.matrix.android.sdk.internal.session.sync.model.RoomSync
 import org.matrix.android.sdk.internal.session.sync.model.RoomSyncAccountData
-import org.matrix.android.sdk.internal.session.sync.model.RoomSyncEphemeral
 import org.matrix.android.sdk.internal.session.sync.model.RoomsSyncResponse
 import timber.log.Timber
 import javax.inject.Inject
@@ -105,7 +104,7 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
         }
         val syncLocalTimeStampMillis = System.currentTimeMillis()
         val rooms = when (handlingStrategy) {
-            is HandlingStrategy.JOINED  ->
+            is HandlingStrategy.JOINED ->
                 handlingStrategy.data.mapWithProgress(reporter, R.string.initial_sync_start_importing_account_joined_rooms, 0.6f) {
                     handleJoinedRoom(realm, it.key, it.value, isInitialSync, insertType, syncLocalTimeStampMillis)
                 }
@@ -114,7 +113,7 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
                     handleInvitedRoom(realm, it.key, it.value, insertType, syncLocalTimeStampMillis)
                 }
 
-            is HandlingStrategy.LEFT    -> {
+            is HandlingStrategy.LEFT -> {
                 handlingStrategy.data.mapWithProgress(reporter, R.string.initial_sync_start_importing_account_left_rooms, 0.3f) {
                     handleLeftRoom(realm, it.key, it.value, insertType, syncLocalTimeStampMillis)
                 }
@@ -132,8 +131,9 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
         Timber.v("Handle join sync for room $roomId")
 
         var ephemeralResult: EphemeralResult? = null
-        if (roomSync.ephemeral?.events?.isNotEmpty() == true) {
-            ephemeralResult = handleEphemeral(realm, roomId, roomSync.ephemeral, isInitialSync)
+        roomSync.ephemeral?.events?.let {
+            ephemeralResult = handleEphemeral(realm, roomId, it, isInitialSync)
+            roomSync.ephemeral.release()
         }
 
         if (roomSync.accountData?.events?.isNotEmpty() == true) {
@@ -164,12 +164,12 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
                 roomMemberEventHandler.handle(realm, roomId, event)
             }
         }
-        if (roomSync.timeline?.events?.isNotEmpty() == true) {
+        roomSync.timeline?.events?.let {
             val chunkEntity = handleTimelineEvents(
                     realm,
                     roomId,
                     roomEntity,
-                    roomSync.timeline.events,
+                    it,
                     roomSync.timeline.prevToken,
                     roomSync.timeline.limited,
                     insertType,
@@ -184,6 +184,8 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
             it.type == EventType.STATE_ROOM_MEMBER
         } != null
 
+        roomSync.timeline?.release()
+
         roomTypingUsersHandler.handle(realm, roomId, ephemeralResult)
         roomChangeMembershipStateDataSource.setMembershipFromSync(roomId, Membership.JOIN)
         roomSummaryUpdater.update(
@@ -274,7 +276,7 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
     private fun handleTimelineEvents(realm: Realm,
                                      roomId: String,
                                      roomEntity: RoomEntity,
-                                     eventList: List<Event>,
+                                     eventList: Sequence<Event>,
                                      prevToken: String? = null,
                                      isLimited: Boolean = true,
                                      insertType: EventInsertType,
@@ -290,7 +292,7 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
         lastChunk?.isLastForward = false
         chunkEntity.isLastForward = true
 
-        val eventIds = ArrayList<String>(eventList.size)
+        val eventIds = ArrayList<String>()
         val roomMemberContentsByUser = HashMap<String, RoomMemberContent?>()
 
         for (event in eventList) {
@@ -375,10 +377,10 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
 
     private fun handleEphemeral(realm: Realm,
                                 roomId: String,
-                                ephemeral: RoomSyncEphemeral,
+                                ephemeral: Sequence<Event>,
                                 isInitialSync: Boolean): EphemeralResult {
         var result = EphemeralResult()
-        for (event in ephemeral.events) {
+        for (event in ephemeral) {
             when (event.type) {
                 EventType.RECEIPT -> {
                     @Suppress("UNCHECKED_CAST")
@@ -386,7 +388,7 @@ internal class RoomSyncHandler @Inject constructor(private val readReceiptHandle
                         readReceiptHandler.handle(realm, roomId, readReceiptContent, isInitialSync)
                     }
                 }
-                EventType.TYPING  -> {
+                EventType.TYPING -> {
                     event.content.toModel<TypingEventContent>()?.let { typingEventContent ->
                         result = result.copy(typingUserIds = typingEventContent.typingUserIds)
                     }

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/model/LazyEventParser.kt

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2020 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.matrix.android.sdk.internal.session.sync.model
+
+import com.squareup.moshi.JsonAdapter
+import com.squareup.moshi.JsonReader
+import com.squareup.moshi.Moshi
+import org.matrix.android.sdk.api.session.events.model.Event
+
+class LazyEventParser(moshi: Moshi) {
+    private val personAdapter: JsonAdapter<Event> = moshi.adapter(Event::class.java)
+
+    fun parse(reader: JsonReader): Sequence<Event> {
+        return if (reader.peek() == JsonReader.Token.BEGIN_ARRAY) {
+            sequence {
+                reader.readArray {
+                    yield(personAdapter.fromJson(reader)!!)
+                }
+            }
+        } else emptySequence()
+    }
+}

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/model/RoomSyncEphemeral.kt

@@ -21,10 +21,13 @@ import com.squareup.moshi.JsonClass
 import org.matrix.android.sdk.api.session.events.model.Event
 
 // RoomSyncEphemeral represents the ephemeral events in the room that aren't recorded in the timeline or state of the room (e.g. typing).
-@JsonClass(generateAdapter = true)
-internal data class RoomSyncEphemeral(
+// @JsonClass(generateAdapter = true)
+internal interface RoomSyncEphemeral {
         /**
          * List of ephemeral events (array of Event).
          */
-        @Json(name = "events") val events: List<Event> = emptyList()
-)
+        @Json(name = "events") val events: Sequence<Event>
+
+
+        fun release()
+}

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/model/RoomSyncEphemeralLazyJsonAdapter.kt

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2020 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.matrix.android.sdk.internal.session.sync.model
+
+import com.squareup.moshi.FromJson
+import com.squareup.moshi.JsonAdapter
+import com.squareup.moshi.JsonReader
+import com.squareup.moshi.JsonWriter
+import com.squareup.moshi.Moshi
+import com.squareup.moshi.ToJson
+import okio.buffer
+import okio.source
+import org.matrix.android.sdk.api.session.events.model.Event
+import java.io.ByteArrayInputStream
+import java.lang.NullPointerException
+
+internal class RoomSyncEphemeralLazyJsonAdapter(private val moshi: Moshi) : JsonAdapter<RoomSyncEphemeral>() {
+
+    companion object {
+        val NAMES = JsonReader.Options.of("events")
+    }
+
+
+    private val lazyEventParser = LazyEventParser(moshi)
+
+    @FromJson
+    override fun fromJson(reader: JsonReader): RoomSyncEphemeral? {
+         var lazySource: String? = null
+        reader.beginObject()
+        while (reader.hasNext()) {
+            when (reader.selectName(RoomSyncTimeLineLazyJsonAdapter.NAMES)) {
+                0 -> {
+                    lazySource = reader.nextSource().readUtf8()
+                    //events = lazyEventParser.parse(JsonReader.of(ByteArrayInputStream(readUtf8.toByteArray()).source().buffer()))
+                }
+                -1 -> {
+                    // Unknown name, skip it.
+                    reader.skipName()
+                    reader.skipValue()
+                }
+            }
+        }
+        reader.endObject()
+
+        return object : RoomSyncEphemeral {
+
+            var src = lazySource
+            override val events: Sequence<Event>
+                get() =  src?.let {
+                    lazyEventParser.parse(JsonReader.of(ByteArrayInputStream(it.toByteArray()).source().buffer()))
+                } ?: emptySequence()
+
+            override fun release() {
+                src  = null
+            }
+        }
+    }
+
+    @ToJson
+    override fun toJson(writer: JsonWriter, value: RoomSyncEphemeral?) {
+        if (value == null) {
+            throw NullPointerException("value was null! Wrap in .nullSafe() to write nullable values.")
+        }
+        val eventAdapter: JsonAdapter<Event> = moshi.adapter(Event::class.java)
+        writer.beginObject()
+        writer.name("events")
+        writer.beginArray()
+        value.events.forEach { event ->
+            eventAdapter.toJson(writer, event)
+        }
+        writer.endArray()
+        writer.endObject()
+    }
+}

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/model/RoomSyncTimeLineLazyJsonAdapter.kt

@@ -0,0 +1,129 @@
+/*
+ * Copyright (c) 2020 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.matrix.android.sdk.internal.session.sync.model
+
+import com.squareup.moshi.FromJson
+import com.squareup.moshi.JsonAdapter
+import com.squareup.moshi.JsonReader
+import com.squareup.moshi.JsonWriter
+import com.squareup.moshi.Moshi
+import com.squareup.moshi.ToJson
+import okio.buffer
+import okio.source
+import org.matrix.android.sdk.api.session.events.model.Event
+import java.io.ByteArrayInputStream
+
+inline fun JsonReader.readArray(body: () -> Unit) {
+    beginArray()
+    while (hasNext()) {
+        body()
+    }
+    endArray()
+}
+
+internal class RoomSyncTimeLineLazyJsonAdapter(private val moshi: Moshi) : JsonAdapter<RoomSyncTimeline>() {
+
+
+    companion object {
+        val NAMES = JsonReader.Options.of("events", "limited", "prev_batch")
+    }
+
+    private val lazyEventParser = LazyEventParser(moshi)
+
+    @FromJson
+    override fun fromJson(reader: JsonReader): RoomSyncTimeline? {
+        var limited: Boolean? = false
+        var prevToken: String? = null
+        var lazyRawString: String? = null
+        var mask0 = -1
+        reader.beginObject()
+        while (reader.hasNext()) {
+            when (reader.selectName(NAMES)) {
+                0 -> {
+                    lazyRawString = reader.nextSource().readUtf8()
+                    //events = lazyEventParser.parse(JsonReader.of(ByteArrayInputStream(readUtf8.toByteArray()).source().buffer()))
+                }
+                1 -> {
+                    limited = reader.nextBoolean()
+                    // $mask = $mask and (1 shl 1).inv()
+                    mask0 = mask0 and 0xfffffffd.toInt()
+                }
+                2 -> {
+                    prevToken = reader.nextString()
+                    // $mask = $mask and (1 shl 2).inv()
+                    mask0 = mask0 and 0xfffffffb.toInt()
+                }
+                -1 -> {
+                    // Unknown name, skip it.
+                    reader.skipName()
+                    reader.skipValue()
+                }
+            }
+        }
+        reader.endObject()
+
+        return object : RoomSyncTimeline {
+
+            var lazySource = lazyRawString
+
+            override val events: Sequence<Event>
+                get() = lazySource?.let {
+                    lazyEventParser.parse(JsonReader.of(ByteArrayInputStream(it.toByteArray()).source().buffer()))
+                } ?: emptySequence()
+
+            override val limited: Boolean
+                get() = limited ?: false
+
+            override val prevToken: String?
+                get() = prevToken
+
+            override fun release() {
+                lazySource = null
+            }
+        }
+
+//        RoomSyncTimeline(
+//                events = events ?: emptySequence(),
+//                limited = limited ?: false,
+//                prevToken = prevToken
+//        )
+    }
+
+//    private val listOfEventAdapter: JsonAdapter<List<Event>> =
+//            moshi.adapter(Types.newParameterizedType(List::class.java, Event::class.java), emptySet(),
+//                    "events")
+
+    @ToJson
+    override fun toJson(writer: JsonWriter, value: RoomSyncTimeline?) {
+        if (value == null) {
+            throw NullPointerException("value was null! Wrap in .nullSafe() to write nullable values.")
+        }
+        val eventAdapter: JsonAdapter<Event> = moshi.adapter(Event::class.java)
+        writer.beginObject()
+        writer.name("events")
+        writer.beginArray()
+        value.events.forEach { event ->
+            eventAdapter.toJson(writer, event)
+        }
+        writer.endArray()
+        writer.name("limited")
+        writer.value(value.limited)
+        writer.name("prev_batch")
+        writer.value(value.prevToken)
+        writer.endObject()
+    }
+}

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/model/RoomSyncTimeline.kt

@@ -21,21 +21,23 @@ import com.squareup.moshi.JsonClass
 import org.matrix.android.sdk.api.session.events.model.Event
 
 // RoomSyncTimeline represents the timeline of messages and state changes for a room during server sync v2.
-@JsonClass(generateAdapter = true)
-internal data class RoomSyncTimeline(
+// @JsonClass(generateAdapter = true)
+internal interface RoomSyncTimeline {
 
         /**
          * List of events (array of Event).
          */
-        @Json(name = "events") val events: List<Event> = emptyList(),
+        val events: Sequence<Event>
 
         /**
          * Boolean which tells whether there are more events on the server
          */
-        @Json(name = "limited") val limited: Boolean = false,
+        val limited: Boolean
 
         /**
          * If the batch was limited then this is a token that can be supplied to the server to retrieve more events
          */
-        @Json(name = "prev_batch") val prevToken: String? = null
-)
+        val prevToken: String?
+
+        fun release()
+}

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/worker/WorkerParamsFactory.kt

@@ -21,6 +21,8 @@ import com.squareup.moshi.Moshi
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.internal.di.MoshiProvider
 import org.matrix.android.sdk.internal.network.parsing.CheckNumberType
+import org.matrix.android.sdk.internal.session.sync.model.RoomSyncTimeLineLazyJsonAdapter
+import org.matrix.android.sdk.internal.session.sync.model.RoomSyncTimeline
 
 internal object WorkerParamsFactory {