flux with kotlin (Abema Dev Con 2016)

Transcript

1. Flux with Kotlin Abema Dev Con 2016

2. About me @_a_akira AAkira a0akira Akira Aratani

3. About FRESH! • 生放送配信プラットフォーム • FRESH! ≒ AbemaTV

4. Released map M1 2012-04-12 M11 2015-03-19 M14 2015-10-01 1.0-beta4 2015-12-22

   M13 2015-09-16 1.0 2016-02-16 1.0-RC 2016-02-04 2016-01-21 Release 2015-04 開発開始 kotlin FRESH

5. Architecture

6. https://facebook.github.io/flux/docs/overview.html Data in a Flux application flows in a single

   direction About Flux

7. About Flux

8. About Flux Observer pattern

9. https://speakerdeck.com/ogaclejapan/flux-de-relax

10. Architecture of FRESH IUUQTHJUIVCDPNBOESPJE"OESPJE$MFBO"SDIJUFDUVSF $MFBO"SDIJUFDUVSF

11. Architecture of FRESH

12. View Action Dispatcher Store DB WEB view(flux) data Repository

13. View Action Dispatcher Store Repository DB WEB view(flux) data OKHttp(Retrofit)

    SQLBrite Observable<T> Subject Observable<T> Observable<T>

14. IUUQRJJUBDPNTBUPSVGVKJXBSBJUFNTDCGEBFDCBC

15. How to implement with kotlin

16. Using libraries • Dagger2 • RxJava(RxKotlin) • RxAndroid • RxBinding

    • SQLBrite • RxPreferences • RxLifecycle => DI => Reactive Extensions => RxJava bindings for Android => binding APIs for Android's UI widgets => introduces reactive stream to SQLite => introduces reactive stream to Preferences => handles Android LifeCycles

17. Archive List BSDIJWF BSDIJWF BSDIJWF BSDIJWF BSDIJWF

18. View Dispatcher Store DB WEB view(flux) data Repository Action

19. Action class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val programRepository: ProgramRepository)

    { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh(screenId) }) }) } }

20. Action class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val programRepository: ProgramRepository)

    { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh(screenId) }) }) } } *OKFDU%JTQBUDIFSBOE3FQPTJUPSZ

21. Action class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val programRepository: ProgramRepository)

    { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh(screenId) }) }) } } $BMM3FQPTJUPSZ

22. View Action Dispatcher Store DB WEB view(flux) data Repository

23. Repository class ProgramRepository(private val programService: ProgramService) { fun getProgramsArchive(count: Int?

    = null, offset: Int? = null) = programService.getPrograms(count = count, offset = offset, status = "archive", orderBy = "endAt") .subscribeOn(Schedulers.io()) .map { it.data.nullSafeMap(::toProgram) } }

24. Repository class ProgramRepository(private val programService: ProgramService) { fun getProgramsArchive(count: Int?

    = null, offset: Int? = null) = programService.getPrograms(count = count, offset = offset, status = "archive", orderBy = "endAt") .subscribeOn(Schedulers.io()) .map { it.data.nullSafeMap(::toProgram) } } *OKFDU4FSWJDF

25. Repository class ProgramRepository(private val programService: ProgramService) { fun getProgramsArchive(count: Int?

    = null, offset: Int? = null) = programService.getPrograms(count = count, offset = offset, status = "archive", orderBy = "endAt") .subscribeOn(Schedulers.io()) .map { it.data.nullSafeMap(::toProgram) } } $BMM4FSWJDF

26. View Action Dispatcher Store view(flux) data Repository DB WEB

27. WEB (Service) interface ProgramService { @GET("/programs") fun getPrograms(@Query("channelId") channelId: String?

    = null, @Query("count") count: Int?, @Query("orderBy") orderBy: String? = null, @Query("status") status: String? = null, ……, ……, ): Observable<ApiResponse<List<ApiProgram>>> }

28. WEB (Service) 3FUVSOPCTFSWBCMF 3FUSPpU 0,)UUQ interface ProgramService { @GET("/programs") fun

    getPrograms(@Query("channelId") channelId: String? = null, @Query("count") count: Int?, @Query("orderBy") orderBy: String? = null, @Query("status") status: String? = null, ……, ……, ): Observable<ApiResponse<List<ApiProgram>>> }

29. View Action Store DB WEB view(flux) data Repository Dispatcher

30. Dispatcher class ArchiveDispatcher(provider: DispatcherProvider) { val archiveDao = provider.archiveDao() }

31. Action / Dispatcher class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val

    programRepository: ProgramRepository) { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh() }) }) } }

32. Action / Dispatcher class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val

    programRepository: ProgramRepository) { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh() }) }) } } 6QEBUFEBUBCBTF

33. View Action Dispatcher DB WEB view(flux) data Repository Store

34. Store class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun programs() =

    dispatcher.archiveDao.selectAll().subscribeOn(Schedulers.io()) }

35. Store *OKFDU%JTQBUDIFS class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun programs()

    = dispatcher.archiveDao.selectAll().subscribeOn(Schedulers.io()) }

36. Store 0CTFSWF42- 42-#SJUF class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun

    programs() = dispatcher.archiveDao.selectAll().subscribeOn(Schedulers.io()) }

37. Action Dispatcher Store DB WEB view(flux) data Repository View

38. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() }

39. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() } *OKFDU"DUJPOBOE4UPSF

40. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() } $BMMSFGSFTIBDUJPO

41. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() } 0CTFSWFTUPSF

42. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() } 3Y"OESPJE

43. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() } 3Y-JGFDZDMF

44. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() }

45. View class ArchiveFragment { @Inject lateinit var archiveAction: ArchiveAction @Inject

    lateinit var archiveStore: ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.programs() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { adapter.clear() it.forEach { … } // add to adapter adapter.notifyDataSetChanged() } savedInstanceState ?: archiveAction.refresh() } Why don’t you implement onError

46. Action / Dispatcher class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val

    programRepository: ProgramRepository) { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh() }) }) } } 4FOEFSSPS open class CommonDispatcher(val provider: DispatcherProvider) { fun errors() = provider.errorDispatcher.errorObservable fun onError(action: Any, cause: Throwable, extras: Any? = null, recoverAction: (() -> Unit)? = null) = provider.errorDispatcher.onError(action, cause, extras, recoverAction) }

47. Action / Dispatcher class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val

    programRepository: ProgramRepository) { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh() }) }) } } open class CommonDispatcher(val provider: DispatcherProvider) { fun errors() = provider.errorDispatcher.errorObservable fun onError(action: Any, cause: Throwable, extras: Any? = null, recoverAction: (() -> Unit)? = null) = provider.errorDispatcher.onError(action, cause, extras, recoverAction) } $PNNPO%JTQBUDIFS

48. Action / Dispatcher class ArchiveAction(private val dispatcher: ArchiveDispatcher, private val

    programRepository: ProgramRepository) { fun refresh() { programRepository.getProgramsArchive(count = 20, offset = 0) .subscribe({ dispatcher.archiveDao.replaceAll(it) }, { dispatcher.onError(this, it, recoverAction = { refresh() }) }) } } open class CommonDispatcher(val provider: DispatcherProvider) { fun errors() = provider.errorDispatcher.errorObservable fun onError(action: Any, cause: Throwable, extras: Any? = null, recoverAction: (() -> Unit)? = null) = provider.errorDispatcher.onError(action, cause, extras, recoverAction) } *OMJOF'VODUJPOTCZ,PUMJO

49. View class ArchiveFragment { @Inject lateinit var archiveStore: ArchiveStore override

    fun onResume() { super.onResume() onAirStore.errors() .filter { it.action is ArchiveAction } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { if (it.kind == ErrorEvent.Kind.NETWORK) { showSnackBar(){ it.recoverAction?.invoke() } } } } }

50. View class ArchiveFragment { @Inject lateinit var archiveStore: ArchiveStore override

    fun onResume() { super.onResume() onAirStore.errors() .filter { it.action is ArchiveAction } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { if (it.kind == ErrorEvent.Kind.NETWORK) { showSnackBar(){ it.recoverAction?.invoke() } } } } } 4VCTDSJCFJOPO3FTVNF

51. View class ArchiveFragment { @Inject lateinit var archiveStore: ArchiveStore override

    fun onResume() { super.onResume() onAirStore.errors() .filter { it.action is ArchiveAction } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { if (it.kind == ErrorEvent.Kind.NETWORK) { showSnackBar(){ it.recoverAction?.invoke() } } } } } 'JMUFSPOMZBSDIJWFBDUJPO %*

52. View class ArchiveFragment { @Inject lateinit var archiveStore: ArchiveStore override

    fun onResume() { super.onResume() onAirStore.errors() .filter { it.action is ArchiveAction } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { if (it.kind == ErrorEvent.Kind.NETWORK) { showSnackBar(){ it.recoverAction?.invoke() } } } } } 3FUSZSFDPWFS"DUJPOJGVTFSTDMJDLBTOBDLCBS

53. Everything is a stream

54. DB WEB view(flux) data Repository Action Dispatcher Store View

55. Like event bus class ArchiveDispatcher(provider: DispatcherProvider) { val statusObservable =

    SerializedSubject(BehaviorSubject.create<Status>()) } class ArchiveAction(private val dispatcher: ArchiveDispatcher) { fun updateStatus(status: Status) { dispatcher.statusObservable.onNext(status) } } class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun status() = dispatcher.statusObservable }

56. Like event bus class ArchiveDispatcher(provider: DispatcherProvider) { val statusObservable =

    SerializedSubject(BehaviorSubject.create<Status>()) } class ArchiveAction(private val dispatcher: ArchiveDispatcher) { fun updateStatus(status: Status) { dispatcher.statusObservable.onNext(status) } } class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun status() = dispatcher.statusObservable }

57. Like event bus class ArchiveDispatcher(provider: DispatcherProvider) { val statusObservable =

    SerializedSubject(BehaviorSubject.create<Status>()) } class ArchiveAction(private val dispatcher: ArchiveDispatcher) { fun updateStatus(status: Status) { dispatcher.statusObservable.onNext(status) } } class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun status() = dispatcher.statusObservable }

58. Like event bus class ArchiveDispatcher(provider: DispatcherProvider) { val statusObservable =

    SerializedSubject(BehaviorSubject.create<Status>()) } class ArchiveAction(private val dispatcher: ArchiveDispatcher) { fun updateStatus(status: Status) { dispatcher.statusObservable.onNext(status) } } class ArchiveStore(private val dispatcher: ArchiveDispatcher) { fun status() = dispatcher.statusObservable }

59. Like event bus class ArchiveFragment { @Inject lateinit var archiveStore:

    ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.status() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { when(it) { HOGE -> textView.text = "hoge" FOO -> textView.text = "foo" BAR -> textView.text = "bar" } } }

60. Like event bus class ArchiveFragment { @Inject lateinit var archiveStore:

    ArchiveStore override fun onViewCreated(view: View?, savedInstanceState: Bundle?) { super.onViewCreated(view, savedInstanceState) archiveStore.status() .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { when(it) { HOGE -> textView.text = "hoge" FOO -> textView.text = "foo" BAR -> textView.text = "bar" } } }

61. List MJTU MJTU MJTU MJTU MJTU refresh

62. Connect observables refreshButton.clicks() .mergeWith(recyclerView.pull()) .withLatestFrom(archiveStore.programs()) { void, programs -> programs

    } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { // add to adapter }

63. Connect observables 3Y#JOEJOH refreshButton.clicks() .mergeWith(recyclerView.pull()) .withLatestFrom(archiveStore.programs()) { void, programs ->

    programs } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { // add to adapter }

64. Connect observables .FSHFDVTUPNPCTFSWBCMF refreshButton.clicks() .mergeWith(recyclerView.pull()) .withLatestFrom(archiveStore.programs()) { void, programs ->

    programs } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { // add to adapter }

65. Connect observables $POOFDUJUFNTGSPNTUPSF refreshButton.clicks() .mergeWith(recyclerView.pull()) .withLatestFrom(archiveStore.programs()) { void, programs ->

    programs } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { // add to adapter }

66. Connect observables "EEUPJUFNTJOTVCTDSJCF refreshButton.clicks() .mergeWith(recyclerView.pull()) .withLatestFrom(archiveStore.programs()) { void, programs ->

    programs } .observeOn(AndroidSchedulers.mainThread()) .bindToLifecycle(this) .subscribe { // add to adapter }

67. • combineLatest, mergeWith, zipWith • map, switchMap, concatMap • scan,

    reduce • interval • filter, skip, distinctUntilChanged • etc… Other operators used in FRESH

68. Conclusion • Clarity ;-) • Beautiful :-) • Comfortable :-D

69. Flux with Kotlin @_a_akira