Stream room messages.

app/src/main/java/chat/rocket/android/chatroom/presentation/ChatRoomPresenter.kt

@@ -6,8 +6,14 @@ import chat.rocket.android.server.infraestructure.RocketChatClientFactory
 import chat.rocket.android.util.launchUI
 import chat.rocket.common.model.BaseRoom
 import chat.rocket.common.util.ifNull
+import chat.rocket.core.internal.realtime.subscribeRoomMessages
+import chat.rocket.core.internal.realtime.unsubscibre
 import chat.rocket.core.internal.rest.messages
 import chat.rocket.core.internal.rest.sendMessage
+import chat.rocket.core.model.Message
+import kotlinx.coroutines.experimental.CommonPool
+import kotlinx.coroutines.experimental.launch
+import timber.log.Timber
 import javax.inject.Inject
 
 class ChatRoomPresenter @Inject constructor(private val view: ChatRoomView,
@@ -15,13 +21,18 @@ class ChatRoomPresenter @Inject constructor(private val view: ChatRoomView,
                                             private val serverInteractor: GetCurrentServerInteractor,
                                             factory: RocketChatClientFactory) {
     private val client = factory.create(serverInteractor.get()!!)
+    private val roomMessages = ArrayList<Message>()
+    private var subId: String? = null
 
     fun loadMessages(chatRoomId: String, chatRoomType: String, offset: Int = 0) {
         launchUI(strategy) {
             view.showLoading()
             try {
                 val messages = client.messages(chatRoomId, BaseRoom.RoomType.valueOf(chatRoomType), offset.toLong(), 30).result
-                view.showMessages(messages.toMutableList(), serverInteractor.get()!!)
+                synchronized(roomMessages) {
+                    roomMessages.addAll(messages)
+                }
+                view.showMessages(messages, serverInteractor.get()!!)
             } catch (ex: Exception) {
                 ex.message?.let {
                     view.showMessage(it)
@@ -38,7 +49,10 @@ class ChatRoomPresenter @Inject constructor(private val view: ChatRoomView,
         launchUI(strategy) {
             try {
                 val message = client.sendMessage(chatRoomId, text)
-                view.showSentMessage(message)
+                synchronized(roomMessages) {
+                    roomMessages.add(0, message)
+                }
+                view.showNewMessage(message)
             } catch (ex: Exception) {
                 ex.message?.let {
                     view.showMessage(it)
@@ -48,4 +62,50 @@ class ChatRoomPresenter @Inject constructor(private val view: ChatRoomView,
             }
         }
     }
+
+    fun subscribeMessages(roomId: String) {
+        launchUI(strategy) {
+            subId = client.subscribeRoomMessages(roomId) {
+                Timber.d("subscribe messages for $roomId: $it")
+            }
+            listenMessages(roomId)
+        }
+    }
+
+    fun unsubscribeMessages() {
+        launchUI(strategy) {
+            subId?.let { subscriptionId ->
+                client.unsubscibre(subscriptionId)
+            }
+        }
+    }
+
+    private suspend fun listenMessages(roomId: String) {
+        launch(CommonPool + strategy.jobs) {
+            for (message in client.messagesChannel) {
+                if (message.roomId != roomId) {
+                    Timber.d("Ignoring message for room ${message.roomId}, expecting $roomId")
+                }
+
+                updateMessage(message)
+            }
+        }
+    }
+
+    private fun updateMessage(streamedMessage: Message) {
+        launchUI(strategy) {
+            synchronized(roomMessages) {
+                val index = roomMessages.indexOfFirst { msg -> msg.id == streamedMessage.id }
+                if (index != -1) {
+                    Timber.d("Updatind message at $index")
+                    roomMessages[index] = streamedMessage
+                    view.dispatchUpdateMessage(index, streamedMessage)
+                } else {
+                    Timber.d("Adding new message")
+                    roomMessages.add(0, streamedMessage)
+                    view.showNewMessage(streamedMessage)
+                }
+            }
+        }
+    }
 }
\ No newline at end of file

app/src/main/java/chat/rocket/android/chatroom/presentation/ChatRoomView.kt

@@ -12,7 +12,7 @@ interface ChatRoomView : LoadingView, MessageView {
      * @param dataSet The data set to show.
      * @param serverUrl The server URL.
      */
-    fun showMessages(dataSet: MutableList<Message>, serverUrl: String)
+    fun showMessages(dataSet: List<Message>, serverUrl: String)
 
     /**
      * Send a message to a chat room.
@@ -23,8 +23,15 @@ interface ChatRoomView : LoadingView, MessageView {
 
     /**
      * Shows a (recent) message sent to a chat room.
-
+     *
      * @param message The (recent) message sent to a chat room.
      */
-    fun showSentMessage(message: Message)
+    fun showNewMessage(message: Message)
+
+    /**
+     * Dispatch a update to the recycler views adapter about a changed message.
+     *
+     * @param index The index of the changed message
+     */
+    fun dispatchUpdateMessage(index: Int, message: Message)
 }
\ No newline at end of file

app/src/main/java/chat/rocket/android/chatroom/ui/ChatRoomActivity.kt

@@ -37,11 +37,10 @@ class ChatRoomActivity : AppCompatActivity(), HasSupportFragmentInjector {
     private var isChatRoomReadOnly: Boolean = false
 
     override fun onCreate(savedInstanceState: Bundle?) {
+        AndroidInjection.inject(this)
         super.onCreate(savedInstanceState)
         setContentView(R.layout.activity_chat_room)
 
-        AndroidInjection.inject(this)
-
         chatRoomId = intent.getStringExtra(INTENT_CHAT_ROOM_ID)
         requireNotNull(chatRoomId) { "no chat_room_id provided in Intent extras" }
 

app/src/main/java/chat/rocket/android/chatroom/ui/ChatRoomAdapter.kt

@@ -19,13 +19,22 @@ import kotlinx.android.synthetic.main.avatar.view.*
 import kotlinx.android.synthetic.main.item_message.view.*
 
 class ChatRoomAdapter(private val context: Context,
-                      private var dataSet: MutableList<Message>,
                       private val serverUrl: String) : RecyclerView.Adapter<ChatRoomAdapter.ViewHolder>() {
 
+    init {
+        setHasStableIds(true)
+    }
+
+    val dataSet = ArrayList<Message>()
+
     override fun onCreateViewHolder(parent: ViewGroup, viewType: Int): ViewHolder = ViewHolder(parent.inflate(R.layout.item_message))
 
     override fun onBindViewHolder(holder: ViewHolder, position: Int) = holder.bind(dataSet[position])
 
+    override fun onBindViewHolder(holder: ViewHolder, position: Int, payloads: MutableList<Any>?) {
+        onBindViewHolder(holder, position)
+    }
+
     override fun getItemCount(): Int = dataSet.size
 
     override fun getItemViewType(position: Int): Int = position
@@ -41,6 +50,15 @@ class ChatRoomAdapter(private val context: Context,
         notifyItemInserted(0)
     }
 
+    fun updateItem(index: Int, message: Message) {
+        dataSet[index] = message
+        notifyItemChanged(index)
+    }
+
+    override fun getItemId(position: Int): Long {
+        return dataSet[position].id.hashCode().toLong()
+    }
+
     inner class ViewHolder(itemView: View) : RecyclerView.ViewHolder(itemView) {
 
         fun bind(message: Message) = with(itemView) {

app/src/main/java/chat/rocket/android/chatroom/ui/ChatRoomFragment.kt

@@ -2,6 +2,7 @@ package chat.rocket.android.chatroom.ui
 
 import android.os.Bundle
 import android.support.v4.app.Fragment
+import android.support.v7.widget.DefaultItemAnimator
 import android.support.v7.widget.LinearLayoutManager
 import android.support.v7.widget.RecyclerView
 import android.view.LayoutInflater
@@ -65,16 +66,23 @@ class ChatRoomFragment : Fragment(), ChatRoomView {
     override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
         super.onViewCreated(view, savedInstanceState)
         presenter.loadMessages(chatRoomId, chatRoomType)
+        presenter.subscribeMessages(chatRoomId)
         setupComposer()
     }
 
-    override fun showMessages(dataSet: MutableList<Message>, serverUrl: String) {
+    override fun onDestroyView() {
+        presenter.unsubscribeMessages()
+        super.onDestroyView()
+    }
+
+    override fun showMessages(dataSet: List<Message>, serverUrl: String) {
         activity?.apply {
             if (recycler_view.adapter == null) {
-                adapter = ChatRoomAdapter(this, dataSet, serverUrl)
+                adapter = ChatRoomAdapter(this, serverUrl)
                 recycler_view.adapter = adapter
                 val linearLayoutManager = LinearLayoutManager(context, LinearLayoutManager.VERTICAL, true)
                 recycler_view.layoutManager = linearLayoutManager
+                recycler_view.itemAnimator = DefaultItemAnimator()
                 if (dataSet.size >= 30) {
                     recycler_view.addOnScrollListener(object : EndlessRecyclerViewScrollListener(linearLayoutManager) {
                         override fun onLoadMore(page: Int, totalItemsCount: Int, recyclerView: RecyclerView?) {
@@ -82,9 +90,9 @@ class ChatRoomFragment : Fragment(), ChatRoomView {
                         }
                     })
                 }
-            } else {
-                adapter.addDataSet(dataSet)
             }
+
+            adapter.addDataSet(dataSet)
         }
     }
 
@@ -94,12 +102,16 @@ class ChatRoomFragment : Fragment(), ChatRoomView {
         }
     }
 
-    override fun showSentMessage(message: Message) {
+    override fun showNewMessage(message: Message) {
         text_message.textContent = ""
         adapter.addItem(message)
         recycler_view.smoothScrollToPosition(0)
     }
 
+    override fun dispatchUpdateMessage(index: Int, message: Message) {
+        adapter.updateItem(index, message)
+    }
+
     override fun showLoading() = view_loading.setVisibility(true)
 
     override fun hideLoading() = view_loading.setVisibility(false)

app/src/main/java/chat/rocket/android/chatrooms/presentation/ChatRoomsPresenter.kt

@@ -87,8 +87,12 @@ class ChatRoomsPresenter @Inject constructor(private val view: ChatRoomsView,
                     State.Authenticating -> Timber.d("Authenticating")
                     State.Connected -> {
                         Timber.d("Connected")
-                        client.subscribeSubscriptions()
-                        client.subscribeRooms()
+                        client.subscribeSubscriptions {
+                            Timber.d("subscriptions: $it")
+                        }
+                        client.subscribeRooms {
+                            Timber.d("rooms: $it")
+                        }
                     }
                 }
             }