ストリームを流すタイミングを調整

app/src/main/kotlin/com/cyder/android/syncpod/repository/ChatDataRepository.kt

@@ -16,49 +16,13 @@ import javax.inject.Inject
 class ChatDataRepository @Inject constructor(
         private val syncPodWsApi: SyncPodWsApi
 ) : ChatRepository {
-    private var chatStream: Subject<Chat> = BehaviorSubject.create()
-    private var pastChatStream: Subject<List<Chat>> = BehaviorSubject.createDefault(emptyList())
-
-    init {
-        initSubjects()
-    }
-
-    private fun startObserve() {
-        syncPodWsApi.chatResponse
-                .subscribe {
-                    it.data?.apply {
-                        this@ChatDataRepository.chatStream.onNext(this.chat.toModel())
-                    }
-                }
-        syncPodWsApi.pastChatsResponse
-                .subscribe {
-                    it.data?.apply {
-                        this@ChatDataRepository.pastChatStream.onNext(this.pastChat.toModel())
-                    }
-                }
-    }
-
-    private fun initSubjects() {
-        syncPodWsApi.isEntered
-                .filter { it }
-                .subscribe {
-                    chatStream = BehaviorSubject.create()
-                    pastChatStream = BehaviorSubject.createDefault(emptyList())
-                    startObserve()
-                }
-        syncPodWsApi.isEntered
-                .filter { !it }
-                .subscribe {
-                    chatStream.onComplete()
-                    pastChatStream.onComplete()
-                }
-    }
-
     override val observePastChat: Flowable<List<Chat>>
-        get() = pastChatStream.toFlowable(BackpressureStrategy.LATEST)
+        get() = syncPodWsApi.pastChatsResponse
+                .map { it.data?.pastChat?.toModel() }
 
     override val observeChat: Flowable<Chat>
-        get() = chatStream.toFlowable(BackpressureStrategy.LATEST)
+        get() = syncPodWsApi.chatResponse
+                .map { it.data?.chat?.toModel() }
 
     override fun getPastChats() {
         syncPodWsApi.requestPastChat()

app/src/main/kotlin/com/cyder/android/syncpod/websocket/SyncPodWsApiImpl.kt

@@ -10,6 +10,7 @@ import io.reactivex.BackpressureStrategy
 import io.reactivex.Completable
 import io.reactivex.Flowable
 import io.reactivex.subjects.BehaviorSubject
+import io.reactivex.subjects.PublishSubject
 import io.reactivex.subjects.Subject
 import javax.inject.Inject
 
@@ -22,14 +23,14 @@ class SyncPodWsApiImpl @Inject constructor(
 ) : SyncPodWsApi {
     private var subscription: Subscription? = null
 
-    private var nowPlayingEvent: Subject<Response> = BehaviorSubject.create()
-    private var startVideoEvent: Subject<Response> = BehaviorSubject.create()
-    private var playListEvent: Subject<Response> = BehaviorSubject.create()
-    private var addVideoEvent: Subject<Response> = BehaviorSubject.create()
-    private var chatEvent: Subject<Response> = BehaviorSubject.create()
+    private var nowPlayingEvent: Subject<Response> = PublishSubject.create()
+    private var startVideoEvent: Subject<Response> = PublishSubject.create()
+    private var playListEvent: Subject<Response> = PublishSubject.create()
+    private var addVideoEvent: Subject<Response> = PublishSubject.create()
+    private var chatEvent: Subject<Response> = PublishSubject.create()
     private val manageEnteredStream: Subject<Boolean> = BehaviorSubject.createDefault(false)
-    private var pastChatsEvent: Subject<Response> = BehaviorSubject.create()
-    private var errorEvent: Subject<Response> = BehaviorSubject.create()
+    private var pastChatsEvent: Subject<Response> = PublishSubject.create()
+    private var errorEvent: Subject<Response> = PublishSubject.create()
     override val nowPlayingResponse: Flowable<Response>
         get() = nowPlayingEvent.toFlowable(BackpressureStrategy.LATEST)
     override val startVideoResponse: Flowable<Response>
@@ -49,13 +50,13 @@ class SyncPodWsApiImpl @Inject constructor(
 
     override fun enterRoom(roomKey: String): Completable {
 
-        nowPlayingEvent = BehaviorSubject.create()
-        startVideoEvent = BehaviorSubject.create()
-        playListEvent = BehaviorSubject.create()
-        addVideoEvent = BehaviorSubject.create()
-        chatEvent = BehaviorSubject.create()
-        pastChatsEvent = BehaviorSubject.create()
-        errorEvent = BehaviorSubject.create()
+        nowPlayingEvent = PublishSubject.create()
+        startVideoEvent = PublishSubject.create()
+        playListEvent = PublishSubject.create()
+        addVideoEvent = PublishSubject.create()
+        chatEvent = PublishSubject.create()
+        pastChatsEvent = PublishSubject.create()
+        errorEvent = PublishSubject.create()
 
         val channel = Channel(CHANNEL_NAME, mapOf(ROOM_KEY to roomKey))
         consumer.disconnect()