@Test public void givenMultipleThreadsWithMultipleMessagesEach_whenIProcess_thenIProperlyGroupByThreadAndRecipient() { List<RecipientId> recipients = Stream.range(1L, 4L).map(RecipientId::from).toList(); List<Long> threads = Stream.range(4L, 7L).toList(); int expected = recipients.size() * threads.size() + 1; List<MessageDatabase.MarkedMessageInfo> infoList = Stream.of(threads) .flatMap(threadId -> Stream.of(recipients) .map(recipientId -> createMarkedMessageInfo(threadId, recipientId))) .toList(); List<MessageDatabase.MarkedMessageInfo> duplicatedList = Util.concatenatedList(infoList, infoList); MarkReadReceiver.process(mockContext, duplicatedList); assertEquals("Should have 10 total jobs, including MultiDeviceReadUpdateJob", expected, jobs.size()); Set<Pair<Long, String>> threadRecipientPairs = new HashSet<>(); Stream.of(jobs).forEach(job -> { if (job instanceof MultiDeviceReadUpdateJob) { return; } Data data = job.serialize(); long threadId = data.getLong("thread"); String recipientId = data.getString("recipient"); long[] messageIds = data.getLongArray("message_ids"); assertEquals("Each job should contain two messages.", 2, messageIds.length); assertTrue("Each thread recipient pair should only exist once.", threadRecipientPairs.add(new Pair<>(threadId, recipientId))); }); assertEquals("Should have 9 total combinations.", 9, threadRecipientPairs.size()); }
public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) { if (markedReadMessages.isEmpty()) return; List<SyncMessageId> syncMessageIds = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getSyncMessageId) .toList(); List<ExpirationInfo> mmsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filter(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); List<ExpirationInfo> smsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filterNot(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); scheduleDeletion(context, smsExpirationInfo, mmsExpirationInfo); ApplicationDependencies.getJobManager().add(new MultiDeviceReadUpdateJob(syncMessageIds)); Map<Long, List<MarkedMessageInfo>> threadToInfo = Stream.of(markedReadMessages) .collect(Collectors.groupingBy(MarkedMessageInfo::getThreadId)); Stream.of(threadToInfo).forEach(threadToInfoEntry -> { Map<RecipientId, List<SyncMessageId>> idMapForThread = Stream.of(threadToInfoEntry.getValue()) .map(MarkedMessageInfo::getSyncMessageId) .collect(Collectors.groupingBy(SyncMessageId::getRecipientId)); Stream.of(idMapForThread).forEach(entry -> { List<Long> timestamps = Stream.of(entry.getValue()).map(SyncMessageId::getTimetamp).toList(); ApplicationDependencies.getJobManager().add(new SendReadReceiptJob(threadToInfoEntry.getKey(), entry.getKey(), timestamps)); }); }); }
MarkReadReceiver extends BroadcastReceiver { public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) { if (markedReadMessages.isEmpty()) return; List<SyncMessageId> syncMessageIds = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getSyncMessageId) .toList(); List<ExpirationInfo> mmsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filter(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); List<ExpirationInfo> smsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filterNot(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); scheduleDeletion(context, smsExpirationInfo, mmsExpirationInfo); ApplicationDependencies.getJobManager().add(new MultiDeviceReadUpdateJob(syncMessageIds)); Map<Long, List<MarkedMessageInfo>> threadToInfo = Stream.of(markedReadMessages) .collect(Collectors.groupingBy(MarkedMessageInfo::getThreadId)); Stream.of(threadToInfo).forEach(threadToInfoEntry -> { Map<RecipientId, List<SyncMessageId>> idMapForThread = Stream.of(threadToInfoEntry.getValue()) .map(MarkedMessageInfo::getSyncMessageId) .collect(Collectors.groupingBy(SyncMessageId::getRecipientId)); Stream.of(idMapForThread).forEach(entry -> { List<Long> timestamps = Stream.of(entry.getValue()).map(SyncMessageId::getTimetamp).toList(); ApplicationDependencies.getJobManager().add(new SendReadReceiptJob(threadToInfoEntry.getKey(), entry.getKey(), timestamps)); }); }); } }
MarkReadReceiver extends BroadcastReceiver { public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) { if (markedReadMessages.isEmpty()) return; List<SyncMessageId> syncMessageIds = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getSyncMessageId) .toList(); List<ExpirationInfo> mmsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filter(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); List<ExpirationInfo> smsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filterNot(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); scheduleDeletion(context, smsExpirationInfo, mmsExpirationInfo); ApplicationDependencies.getJobManager().add(new MultiDeviceReadUpdateJob(syncMessageIds)); Map<Long, List<MarkedMessageInfo>> threadToInfo = Stream.of(markedReadMessages) .collect(Collectors.groupingBy(MarkedMessageInfo::getThreadId)); Stream.of(threadToInfo).forEach(threadToInfoEntry -> { Map<RecipientId, List<SyncMessageId>> idMapForThread = Stream.of(threadToInfoEntry.getValue()) .map(MarkedMessageInfo::getSyncMessageId) .collect(Collectors.groupingBy(SyncMessageId::getRecipientId)); Stream.of(idMapForThread).forEach(entry -> { List<Long> timestamps = Stream.of(entry.getValue()).map(SyncMessageId::getTimetamp).toList(); ApplicationDependencies.getJobManager().add(new SendReadReceiptJob(threadToInfoEntry.getKey(), entry.getKey(), timestamps)); }); }); }  }
MarkReadReceiver extends BroadcastReceiver { public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) { if (markedReadMessages.isEmpty()) return; List<SyncMessageId> syncMessageIds = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getSyncMessageId) .toList(); List<ExpirationInfo> mmsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filter(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); List<ExpirationInfo> smsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filterNot(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); scheduleDeletion(context, smsExpirationInfo, mmsExpirationInfo); ApplicationDependencies.getJobManager().add(new MultiDeviceReadUpdateJob(syncMessageIds)); Map<Long, List<MarkedMessageInfo>> threadToInfo = Stream.of(markedReadMessages) .collect(Collectors.groupingBy(MarkedMessageInfo::getThreadId)); Stream.of(threadToInfo).forEach(threadToInfoEntry -> { Map<RecipientId, List<SyncMessageId>> idMapForThread = Stream.of(threadToInfoEntry.getValue()) .map(MarkedMessageInfo::getSyncMessageId) .collect(Collectors.groupingBy(SyncMessageId::getRecipientId)); Stream.of(idMapForThread).forEach(entry -> { List<Long> timestamps = Stream.of(entry.getValue()).map(SyncMessageId::getTimetamp).toList(); ApplicationDependencies.getJobManager().add(new SendReadReceiptJob(threadToInfoEntry.getKey(), entry.getKey(), timestamps)); }); }); }  @SuppressLint("StaticFieldLeak") @Override void onReceive(final Context context, Intent intent); static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages); }
MarkReadReceiver extends BroadcastReceiver { public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) { if (markedReadMessages.isEmpty()) return; List<SyncMessageId> syncMessageIds = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getSyncMessageId) .toList(); List<ExpirationInfo> mmsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filter(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); List<ExpirationInfo> smsExpirationInfo = Stream.of(markedReadMessages) .map(MarkedMessageInfo::getExpirationInfo) .filterNot(ExpirationInfo::isMms) .filter(info -> info.getExpiresIn() > 0 && info.getExpireStarted() <= 0) .toList(); scheduleDeletion(context, smsExpirationInfo, mmsExpirationInfo); ApplicationDependencies.getJobManager().add(new MultiDeviceReadUpdateJob(syncMessageIds)); Map<Long, List<MarkedMessageInfo>> threadToInfo = Stream.of(markedReadMessages) .collect(Collectors.groupingBy(MarkedMessageInfo::getThreadId)); Stream.of(threadToInfo).forEach(threadToInfoEntry -> { Map<RecipientId, List<SyncMessageId>> idMapForThread = Stream.of(threadToInfoEntry.getValue()) .map(MarkedMessageInfo::getSyncMessageId) .collect(Collectors.groupingBy(SyncMessageId::getRecipientId)); Stream.of(idMapForThread).forEach(entry -> { List<Long> timestamps = Stream.of(entry.getValue()).map(SyncMessageId::getTimetamp).toList(); ApplicationDependencies.getJobManager().add(new SendReadReceiptJob(threadToInfoEntry.getKey(), entry.getKey(), timestamps)); }); }); }  @SuppressLint("StaticFieldLeak") @Override void onReceive(final Context context, Intent intent); static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages); static final String CLEAR_ACTION; static final String THREAD_IDS_EXTRA; static final String NOTIFICATION_ID_EXTRA; }