send read receipt using new pipeline

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationActivity.java

@@ -141,6 +141,7 @@ import org.thoughtcrime.securesms.loki.activities.HomeActivity;
 import org.thoughtcrime.securesms.loki.api.PublicChatInfoUpdateWorker;
 import org.thoughtcrime.securesms.loki.database.LokiThreadDatabase;
 import org.thoughtcrime.securesms.loki.database.LokiUserDatabase;
+import org.thoughtcrime.securesms.loki.protocol.SessionMetaProtocol;
 import org.thoughtcrime.securesms.loki.utilities.GeneralUtilitiesKt;
 import org.thoughtcrime.securesms.loki.utilities.MentionManagerUtilities;
 import org.thoughtcrime.securesms.loki.utilities.OpenGroupUtilities;
@@ -1661,7 +1662,7 @@ public class ConversationActivity extends PassphraseRequiredActionBarActivity
         Context                 context    = ConversationActivity.this;
         List<MarkedMessageInfo> messageIds = DatabaseFactory.getThreadDatabase(context).setRead(params[0], false);
 
-        if (!org.thoughtcrime.securesms.loki.protocol.SessionMetaProtocol.shouldSendReadReceipt(recipient.getAddress())) {
+        if (!SessionMetaProtocol.shouldSendReadReceipt(recipient.getAddress())) {
           for (MarkedMessageInfo messageInfo : messageIds) {
             MarkReadReceiver.scheduleDeletion(context, messageInfo.getExpirationInfo());
           }

app/src/main/java/org/thoughtcrime/securesms/notifications/MarkReadReceiver.java

@@ -11,13 +11,15 @@ import androidx.core.app.NotificationManagerCompat;
 import com.annimon.stream.Collectors;
 import com.annimon.stream.Stream;
 
+import org.session.libsession.messaging.messages.control.ReadReceipt;
+import org.session.libsession.messaging.sending_receiving.MessageSender;
+import org.session.libsession.utilities.TextSecurePreferences;
 import org.thoughtcrime.securesms.ApplicationContext;
 import org.session.libsession.messaging.threads.Address;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
 import org.thoughtcrime.securesms.database.MessagingDatabase.ExpirationInfo;
 import org.thoughtcrime.securesms.database.MessagingDatabase.MarkedMessageInfo;
 import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
-import org.thoughtcrime.securesms.jobs.SendReadReceiptJob;
 import org.session.libsignal.utilities.logging.Log;
 import org.thoughtcrime.securesms.loki.protocol.SessionMetaProtocol;
 import org.thoughtcrime.securesms.service.ExpiringMessageManager;
@@ -25,7 +27,6 @@ import org.thoughtcrime.securesms.service.ExpiringMessageManager;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 public class MarkReadReceiver extends BroadcastReceiver {
 
@@ -68,8 +69,7 @@ public class MarkReadReceiver extends BroadcastReceiver {
 
   public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) {
     if (markedReadMessages.isEmpty()) return;
-
-    List<SyncMessageId> syncMessageIds = new LinkedList<>();
+    if (!TextSecurePreferences.isReadReceiptsEnabled(context)) return;
 
     for (MarkedMessageInfo messageInfo : markedReadMessages) {
       scheduleDeletion(context, messageInfo.getExpirationInfo());
@@ -83,10 +83,9 @@ public class MarkReadReceiver extends BroadcastReceiver {
       List<Long> timestamps = Stream.of(addressMap.get(address)).map(SyncMessageId::getTimetamp).toList();
       // Loki - Check whether we want to send a read receipt to this user
       if (!SessionMetaProtocol.shouldSendReadReceipt(address)) { continue; }
-      // Loki - Take into account multi device
-      ApplicationContext.getInstance(context)
-              .getJobManager()
-              .add(new SendReadReceiptJob(address, timestamps));
+      ReadReceipt readReceipt = new ReadReceipt(timestamps);
+      readReceipt.setSentTimestamp(System.currentTimeMillis());
+      MessageSender.send(readReceipt, address);
     }
   }
 

libsession/src/main/java/org/session/libsession/messaging/messages/control/ReadReceipt.kt

@@ -5,7 +5,7 @@ import org.session.libsignal.service.internal.push.SignalServiceProtos
 
 class ReadReceipt() : ControlMessage() {
 
-    var timestamps: LongArray? = null
+    var timestamps: List<Long>? = null
 
     companion object {
         const val TAG = "ReadReceipt"
@@ -15,12 +15,12 @@ class ReadReceipt() : ControlMessage() {
             if (receiptProto.type != SignalServiceProtos.ReceiptMessage.Type.READ) return null
             val timestamps = receiptProto.timestampList
             if (timestamps.isEmpty()) return null
-            return ReadReceipt(timestamps = timestamps.toLongArray())
+            return ReadReceipt(timestamps = timestamps)
         }
     }
 
     //constructor
-    internal constructor(timestamps: LongArray?) : this() {
+    internal constructor(timestamps: List<Long>?) : this() {
         this.timestamps = timestamps
     }
 

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageReceiverHandler.kt

@@ -50,7 +50,7 @@ fun MessageReceiver.handle(message: Message, proto: SignalServiceProtos.Content,
 
 private fun MessageReceiver.handleReadReceipt(message: ReadReceipt) {
     val context = MessagingConfiguration.shared.context
-    SSKEnvironment.shared.readReceiptManager.processReadReceipts(context, message.sender!!, message.timestamps!!.asList(), message.receivedTimestamp!!)
+    SSKEnvironment.shared.readReceiptManager.processReadReceipts(context, message.sender!!, message.timestamps!!, message.receivedTimestamp!!)
 }
 
 private fun MessageReceiver.handleTypingIndicator(message: TypingIndicator) {