Add RxStore and StoryViewerPage forward navigation.

app/src/main/java/org/thoughtcrime/securesms/database/model/DisplayRecord.java

@@ -20,6 +20,7 @@ import android.content.Context;
 import android.text.SpannableString;
 
 import androidx.annotation.NonNull;
+import androidx.annotation.VisibleForTesting;
 
 import org.thoughtcrime.securesms.database.MmsSmsColumns;
 import org.thoughtcrime.securesms.database.SmsDatabase;
@@ -80,6 +81,11 @@ public abstract class DisplayRecord {
            !MmsSmsColumns.Types.isIdentityDefault(type);
   }
 
+  @VisibleForTesting
+  public long getType() {
+    return type;
+  }
+
   public boolean isSent() {
     return MmsSmsColumns.Types.isSentType(type);
   }

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/page/StoryPost.kt

@@ -10,7 +10,7 @@ import org.thoughtcrime.securesms.util.MediaUtil
 /**
  * Each story is made up of a collection of posts
  */
-class StoryPost(
+data class StoryPost(
   val id: Long,
   val sender: Recipient,
   val group: Recipient?,
@@ -20,7 +20,8 @@ class StoryPost(
   val dateInMilliseconds: Long,
   val content: Content,
   val conversationMessage: ConversationMessage,
-  val allowsReplies: Boolean
+  val allowsReplies: Boolean,
+  val hasSelfViewed: Boolean
 ) {
   sealed class Content(val uri: Uri?) {
     class AttachmentContent(val attachment: Attachment) : Content(attachment.uri) {

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/page/StoryViewerPageFragment.kt

@@ -22,6 +22,7 @@ import androidx.core.view.doOnNextLayout
 import androidx.fragment.app.DialogFragment
 import androidx.fragment.app.Fragment
 import androidx.fragment.app.viewModels
+import androidx.lifecycle.LiveDataReactiveStreams
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Observable
 import org.signal.core.util.DimensionUnit
@@ -252,42 +253,44 @@ class StoryViewerPageFragment :
       }
     }
 
-    viewModel.state.observe(viewLifecycleOwner) { state ->
-      if (state.posts.isNotEmpty() && state.selectedPostIndex < state.posts.size) {
-        val post = state.posts[state.selectedPostIndex]
+    LiveDataReactiveStreams
+      .fromPublisher(viewModel.state.observeOn(AndroidSchedulers.mainThread()))
+      .observe(viewLifecycleOwner) { state ->
+        if (state.posts.isNotEmpty() && state.selectedPostIndex < state.posts.size) {
+          val post = state.posts[state.selectedPostIndex]
 
-        presentViewsAndReplies(post)
-        presentSenderAvatar(senderAvatar, post)
-        presentGroupAvatar(groupAvatar, post)
-        presentFrom(from, post)
-        presentDate(date, post)
-        presentDistributionList(distributionList, post)
-        presentCaption(caption, largeCaption, largeCaptionOverlay, post)
-        presentBlur(blurContainer, post)
+          presentViewsAndReplies(post)
+          presentSenderAvatar(senderAvatar, post)
+          presentGroupAvatar(groupAvatar, post)
+          presentFrom(from, post)
+          presentDate(date, post)
+          presentDistributionList(distributionList, post)
+          presentCaption(caption, largeCaption, largeCaptionOverlay, post)
+          presentBlur(blurContainer, post)
 
-        val durations: Map<Int, Long> = state.posts
-          .mapIndexed { index, storyPost ->
-            index to when {
-              storyPost.content.isVideo() -> -1L
-              storyPost.content is StoryPost.Content.TextContent -> calculateDurationForText(storyPost.content)
-              else -> DEFAULT_DURATION
+          val durations: Map<Int, Long> = state.posts
+            .mapIndexed { index, storyPost ->
+              index to when {
+                storyPost.content.isVideo() -> -1L
+                storyPost.content is StoryPost.Content.TextContent -> calculateDurationForText(storyPost.content)
+                else -> DEFAULT_DURATION
+              }
             }
+            .toMap()
+
+          if (progressBar.segmentCount != state.posts.size || progressBar.segmentDurations != durations) {
+            progressBar.segmentCount = state.posts.size
+            progressBar.segmentDurations = durations
           }
-          .toMap()
 
-        if (progressBar.segmentCount != state.posts.size || progressBar.segmentDurations != durations) {
-          progressBar.segmentCount = state.posts.size
-          progressBar.segmentDurations = durations
+          presentStory(post, state.selectedPostIndex)
+          presentSlate(post)
+
+          viewModel.setAreSegmentsInitialized(true)
+        } else if (state.selectedPostIndex >= state.posts.size) {
+          callback.onFinishedPosts(storyRecipientId)
         }
-
-        presentStory(post, state.selectedPostIndex)
-        presentSlate(post)
-
-        viewModel.setAreSegmentsInitialized(true)
-      } else if (state.selectedPostIndex >= state.posts.size) {
-        callback.onFinishedPosts(storyRecipientId)
       }
-    }
 
     viewModel.storyViewerPlaybackState.observe(viewLifecycleOwner) { state ->
       if (state.isPaused) {

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/page/StoryViewerPageRepository.kt

@@ -24,7 +24,10 @@ import org.thoughtcrime.securesms.recipients.Recipient
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.thoughtcrime.securesms.util.Base64
 
-class StoryViewerPageRepository(context: Context) {
+/**
+ * Open for testing.
+ */
+open class StoryViewerPageRepository(context: Context) {
 
   private val context = context.applicationContext
 
@@ -77,7 +80,8 @@ class StoryViewerPageRepository(context: Context) {
           dateInMilliseconds = record.dateSent,
           content = getContent(record as MmsMessageRecord),
           conversationMessage = ConversationMessage.ConversationMessageFactory.createWithUnresolvedData(context, record),
-          allowsReplies = record.storyType.isStoryWithReplies
+          allowsReplies = record.storyType.isStoryWithReplies,
+          hasSelfViewed = if (record.isOutgoing) true else record.viewedReceiptCount > 0
         )
 
         emitter.onNext(story)

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/page/StoryViewerPageViewModel.kt

@@ -4,6 +4,7 @@ import androidx.lifecycle.LiveData
 import androidx.lifecycle.ViewModel
 import androidx.lifecycle.ViewModelProvider
 import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Observable
 import io.reactivex.rxjava3.disposables.CompositeDisposable
 import io.reactivex.rxjava3.kotlin.plusAssign
@@ -11,6 +12,7 @@ import io.reactivex.rxjava3.subjects.PublishSubject
 import io.reactivex.rxjava3.subjects.Subject
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.thoughtcrime.securesms.util.livedata.Store
+import org.thoughtcrime.securesms.util.rx.RxStore
 import java.util.Optional
 import kotlin.math.max
 import kotlin.math.min
@@ -24,7 +26,7 @@ class StoryViewerPageViewModel(
   private val repository: StoryViewerPageRepository
 ) : ViewModel() {
 
-  private val store = Store(StoryViewerPageState())
+  private val store = RxStore(StoryViewerPageState())
   private val disposables = CompositeDisposable()
   private val storyViewerDialogSubject: Subject<Optional<StoryViewerDialog>> = PublishSubject.create()
 
@@ -34,7 +36,7 @@ class StoryViewerPageViewModel(
 
   val groupDirectReplyObservable: Observable<Optional<StoryViewerDialog>> = storyViewerDialogSubject
 
-  val state: LiveData<StoryViewerPageState> = store.stateLiveData
+  val state: Flowable<StoryViewerPageState> = store.stateFlowable
 
   fun getStateSnapshot(): StoryViewerPageState = store.state
 
@@ -50,7 +52,8 @@ class StoryViewerPageViewModel(
           val initialIndex = posts.indexOfFirst { it.id == initialStoryId }
           initialIndex.takeIf { it > -1 } ?: state.selectedPostIndex
         } else if (state.posts.isEmpty()) {
-          val initialIndex = posts.indexOfFirst { !it.conversationMessage.messageRecord.isOutgoing && it.conversationMessage.messageRecord.viewedReceiptCount == 0 }
+          val initialPost = getNextUnreadPost(posts)
+          val initialIndex = initialPost?.let { posts.indexOf(it) } ?: -1
           initialIndex.takeIf { it > -1 } ?: state.selectedPostIndex
         } else {
           state.selectedPostIndex
@@ -89,11 +92,24 @@ class StoryViewerPageViewModel(
   }
 
   fun goToNextPost() {
+    if (store.state.posts.isEmpty()) {
+      return
+    }
+
     val postIndex = store.state.selectedPostIndex
-    setSelectedPostIndex(postIndex + 1)
+    val nextUnreadPost: StoryPost? = getNextUnreadPost(store.state.posts.drop(postIndex + 1))
+    if (nextUnreadPost == null) {
+      setSelectedPostIndex(postIndex + 1)
+    } else {
+      setSelectedPostIndex(store.state.posts.indexOf(nextUnreadPost))
+    }
   }
 
   fun goToPreviousPost() {
+    if (store.state.posts.isEmpty()) {
+      return
+    }
+
     val postIndex = store.state.selectedPostIndex
     setSelectedPostIndex(max(0, postIndex - 1))
   }
@@ -194,6 +210,10 @@ class StoryViewerPageViewModel(
     }
   }
 
+  private fun getNextUnreadPost(list: List<StoryPost>): StoryPost? {
+    return list.firstOrNull { !it.hasSelfViewed }
+  }
+
   fun getPostAt(index: Int): StoryPost? {
     return store.state.posts.getOrNull(index)
   }

app/src/main/java/org/thoughtcrime/securesms/util/rx/RxStore.kt

@@ -0,0 +1,30 @@
+package org.thoughtcrime.securesms.util.rx
+
+import io.reactivex.rxjava3.core.Flowable
+import io.reactivex.rxjava3.processors.BehaviorProcessor
+import io.reactivex.rxjava3.schedulers.Schedulers
+import io.reactivex.rxjava3.subjects.PublishSubject
+
+/**
+ * Rx replacement for Store.
+ * Actions are run on the computation thread.
+ */
+class RxStore<T : Any>(defaultValue: T) {
+
+  private val behaviorProcessor = BehaviorProcessor.createDefault(defaultValue)
+  private val actionSubject = PublishSubject.create<(T) -> T>().toSerialized()
+
+  val state: T get() = behaviorProcessor.value!!
+  val stateFlowable: Flowable<T> = behaviorProcessor
+
+  init {
+    actionSubject
+      .observeOn(Schedulers.computation())
+      .scan(defaultValue) { v, f -> f(v) }
+      .subscribe { behaviorProcessor.onNext(it) }
+  }
+
+  fun update(transformer: (T) -> T) {
+    actionSubject.onNext(transformer)
+  }
+}

app/src/test/java/org/thoughtcrime/securesms/stories/viewer/page/StoryViewerPageViewModelTest.kt

@@ -0,0 +1,129 @@
+package org.thoughtcrime.securesms.stories.viewer.page
+
+import android.app.Application
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.plugins.RxJavaPlugins
+import io.reactivex.rxjava3.schedulers.TestScheduler
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.whenever
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+
+@RunWith(RobolectricTestRunner::class)
+@Config(application = Application::class)
+class StoryViewerPageViewModelTest {
+
+  private val repository: StoryViewerPageRepository = mock()
+
+  private val testScheduler = TestScheduler()
+
+  @Before
+  fun setUp() {
+    RxJavaPlugins.setInitComputationSchedulerHandler { testScheduler }
+    RxJavaPlugins.setComputationSchedulerHandler { testScheduler }
+  }
+
+  @After
+  fun tearDown() {
+    RxJavaPlugins.reset()
+  }
+
+  @Test
+  fun `Given no initial story and 3 records all viewed, when I initialize, then I expect storyIndex to be 0`() {
+    // GIVEN
+    val storyPosts = createStoryPosts(3) { true }
+    whenever(repository.getStoryPostsFor(any())).thenReturn(Observable.just(storyPosts))
+
+    // WHEN
+    val testSubject = createTestSubject()
+
+    // THEN
+    testScheduler.triggerActions()
+    val testSubscriber = testSubject.state.test()
+
+    testSubscriber.assertValueAt(0) { it.selectedPostIndex == 0 }
+  }
+
+  @Test
+  fun `Given no initial story and 3 records all not viewed, when I initialize, then I expect storyIndex to be 0`() {
+    // GIVEN
+    val storyPosts = createStoryPosts(3) { false }
+    whenever(repository.getStoryPostsFor(any())).thenReturn(Observable.just(storyPosts))
+
+    // WHEN
+    val testSubject = createTestSubject()
+
+    // THEN
+    testScheduler.triggerActions()
+    val testSubscriber = testSubject.state.test()
+
+    testSubscriber.assertValueAt(0) { it.selectedPostIndex == 0 }
+  }
+
+  @Test
+  fun `Given no initial story and 3 records with 2nd is not viewed, when I initialize, then I expect storyIndex to be 1`() {
+    // GIVEN
+    val storyPosts = createStoryPosts(3) { it % 2 != 0 }
+    whenever(repository.getStoryPostsFor(any())).thenReturn(Observable.just(storyPosts))
+
+    // WHEN
+    val testSubject = createTestSubject()
+
+    // THEN
+    testScheduler.triggerActions()
+    val testSubscriber = testSubject.state.test()
+
+    testSubscriber.assertValueAt(0) { it.selectedPostIndex == 1 }
+  }
+
+  @Test
+  fun `Given no initial story and 3 records with 1st and 3rd not viewed, when I goToNext, then I expect storyIndex to be 2`() {
+    // GIVEN
+    val storyPosts = createStoryPosts(3) { it % 2 == 0 }
+    whenever(repository.getStoryPostsFor(any())).thenReturn(Observable.just(storyPosts))
+
+    // WHEN
+    val testSubject = createTestSubject()
+    testScheduler.triggerActions()
+    testSubject.goToNextPost()
+    testScheduler.triggerActions()
+
+    // THEN
+    val testSubscriber = testSubject.state.test()
+
+    testSubscriber.assertValueAt(0) { it.selectedPostIndex == 2 }
+  }
+
+  private fun createTestSubject(): StoryViewerPageViewModel {
+    return StoryViewerPageViewModel(
+      RecipientId.from(1),
+      -1L,
+      repository
+    )
+  }
+
+  private fun createStoryPosts(count: Int, isViewed: (Int) -> Boolean = { false }): List<StoryPost> {
+    return (1..count).map {
+      StoryPost(
+        id = it.toLong(),
+        sender = Recipient.UNKNOWN,
+        group = null,
+        distributionList = null,
+        viewCount = 0,
+        replyCount = 0,
+        dateInMilliseconds = it.toLong(),
+        content = StoryPost.Content.TextContent(mock(), it.toLong(), false, 0),
+        conversationMessage = mock(),
+        allowsReplies = true,
+        hasSelfViewed = isViewed(it)
+      )
+    }
+  }
+}

app/src/test/java/org/thoughtcrime/securesms/util/rx/RxStoreTest.kt

@@ -0,0 +1,70 @@
+package org.thoughtcrime.securesms.util.rx
+
+import io.reactivex.rxjava3.plugins.RxJavaPlugins
+import io.reactivex.rxjava3.schedulers.TestScheduler
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+
+class RxStoreTest {
+
+  private val testScheduler = TestScheduler()
+
+  @Before
+  fun setUp() {
+    RxJavaPlugins.setInitComputationSchedulerHandler { testScheduler }
+    RxJavaPlugins.setComputationSchedulerHandler { testScheduler }
+  }
+
+  @After
+  fun tearDown() {
+    RxJavaPlugins.reset()
+  }
+
+  @Test
+  fun `Given an initial state, when I observe, then I expect my initial state`() {
+    // GIVEN
+    val testSubject = RxStore(1)
+
+    // WHEN
+    val subscriber = testSubject.stateFlowable.test()
+    testScheduler.triggerActions()
+
+    // THEN
+    subscriber.assertValueAt(0, 1)
+    subscriber.assertNotComplete()
+  }
+
+  @Test
+  fun `Given immediate observation, when I update, then I expect both states`() {
+    // GIVEN
+    val testSubject = RxStore(1)
+
+    // WHEN
+    val subscriber = testSubject.stateFlowable.test()
+    testSubject.update { 2 }
+
+    testScheduler.triggerActions()
+
+    // THEN
+    subscriber.assertValueAt(0, 1)
+    subscriber.assertValueAt(1, 2)
+    subscriber.assertNotComplete()
+  }
+
+  @Test
+  fun `Given late observation after several updates, when I observe, then I expect latest state`() {
+    // GIVEN
+    val testSubject = RxStore(1)
+    testSubject.update { 2 }
+
+    // WHEN
+    testScheduler.triggerActions()
+    val subscriber = testSubject.stateFlowable.test()
+    testScheduler.triggerActions()
+
+    // THEN
+    subscriber.assertValueAt(0, 2)
+    subscriber.assertNotComplete()
+  }
+}