Skip to content

Commit dfb7b0a

Browse files
author
Kaushik Gopal
committed
fix: merge conflict
1 parent c8f7d79 commit dfb7b0a

14 files changed

Lines changed: 409 additions & 22 deletions

File tree

‎README.md‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@ I've also been giving talks about Learning Rx using many of the examples listed
2626
14. [Pagination with Rx (using Subjects)](#14-pagination-with-rx-using-subjects)
2727
15. [Orchestrating Observable: make parallel network calls, then combine the result into a single data point (using flatmap & zip)](#15-orchestrating-observable-make-parallel-network-calls-then-combine-the-result-into-a-single-data-point-using-flatmap--zip)
2828
16. [Simple Timeout example (using timeout)](#16-simple-timeout-example-using-timeout)
29+
17. [Setup and teardown resources (using `using`)](#17-setup-and-teardown-resources-using-using)
30+
18. [Multicast playground](#18-multicast-playground)
2931

3032
## Description
3133

@@ -161,7 +163,7 @@ Cases demonstrated here:
161163
4. run a task constantly every 3s, but after running it 5 times, terminate automatically
162164
5. run a task A, pause for sometime, then execute Task B, then terminate
163165

164-
### 11. RxBus : event bus using RxJava (using RxRelay (never terminating Subjects) and debouncedBuffer)
166+
### 11. RxBus : event bus using RxJava (using RxRelay (never terminating Subjects) and debouncedBuffer)
165167

166168
There are accompanying blog posts that do a much better job of explaining the details on this demo:
167169

@@ -222,6 +224,20 @@ This is a simple example demonstrating the use of the `.timeout` operator. Butto
222224

223225
Notice how we can provide a custom Observable that indicates how to react under a timeout Exception.
224226

227+
### 17. Setup and teardown resources (using `using`)
228+
229+
The [operator `using`](http://reactivex.io/documentation/operators/using.html) is relatively less known and notoriously hard to Google. It's a beautiful API that helps to setup a (costly) resource, use it and then dispose off in a clean way.
230+
231+
The nice thing about this operator is that it provides a mechansim to use potentially costly resources in a tightly scoped manner. using -> setup, use and dispose. Think DB connections (like Realm instances), socket connections, thread locks etc.
232+
233+
### 18. Multicast Playground
234+
235+
Multicasting in Rx is like a dark art. Not too many folks know how to pull it off without concern. This example condiers two subscribers (in the forms of buttons) and allows you to add/remove subscribers at different points of time and see how the different operators behave under those circumstances.
236+
237+
The source observale is a timer (`interval`) observable and the reason this was chosen was to intentionally pick a non-terminating observable, so you can test/confirm if your multicast experiment will leak.
238+
239+
_I also gave a talk about [Multicasting in detail at 360|Andev](https://speakerdeck.com/kaushikgopal/rx-by-example-volume-3-the-multicast-edition). If you have the inclination and time, I highly suggest watching that talk first (specifically the Multicast operator permutation segment) and then messing around with the example here._
240+
225241
## Rx 2.x
226242

227243
All the examples here have been migrated to use RxJava 2.X.

‎app/build.gradle‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ dependencies {
2525

2626
compile 'com.github.kaushikgopal:CoreTextUtils:c703fa12b6'
2727
compile "com.jakewharton:butterknife:${butterKnifeVersion}"
28-
annotationProcessor "com.jakewharton:butterknife-compiler:${butterKnifeVersion}"
28+
kapt "com.jakewharton:butterknife-compiler:${butterKnifeVersion}"
2929
compile 'com.jakewharton.timber:timber:4.5.1'
3030
compile "com.squareup.retrofit2:retrofit:${retrofitVersion}"
3131
compile "com.squareup.retrofit2:converter-gson:${retrofitVersion}"
@@ -45,6 +45,7 @@ dependencies {
4545
// explicitly depend on RxJava's latest version for bug fixes and new features.
4646
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
4747

48+
compile 'com.jakewharton.rx:replaying-share-kotlin:2.0.0'
4849
compile "com.github.akarnokd:rxjava2-extensions:0.16.0"
4950
compile 'com.jakewharton.rxrelay2:rxrelay:2.0.0'
5051
compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
@@ -66,7 +67,6 @@ android {
6667
targetSdkVersion sdkVersion
6768
versionCode 2
6869
versionName "1.2"
69-
7070
multiDexEnabled true
7171
}
7272
buildTypes {

‎app/src/main/java/com/morihacky/android/rxjava/MyApp.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
package com.morihacky.android.rxjava;
22

33
import android.support.multidex.MultiDexApplication;
4-
54
import com.morihacky.android.rxjava.volley.MyVolley;
65
import com.squareup.leakcanary.LeakCanary;
76
import com.squareup.leakcanary.RefWatcher;

‎app/src/main/java/com/morihacky/android/rxjava/fragments/MainFragment.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,16 @@ void demoNetworkDetector() {
116116
clickedOn(new NetworkDetectorFragment());
117117
}
118118

119+
@OnClick(R.id.btn_demo_using)
120+
void demoUsing() {
121+
clickedOn(new UsingFragment());
122+
}
123+
124+
@OnClick(R.id.btn_demo_multicastPlayground)
125+
void demoMulticastPlayground() {
126+
clickedOn(new MulticastPlaygroundFragment());
127+
}
128+
119129
private void clickedOn(@NonNull Fragment fragment) {
120130
final String tag = fragment.getClass().toString();
121131
getActivity()

‎app/src/main/kotlin/com/morihacky/android/rxjava/ext/RxExt.kt‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package com.morihacky.android.rxjava.ext
2+
3+
import io.reactivex.disposables.CompositeDisposable
4+
import io.reactivex.disposables.Disposable
5+
6+
operator fun CompositeDisposable.plus(disposable: Disposable): CompositeDisposable {
7+
add(disposable)
8+
return this
9+
}
10+
11+

‎app/src/main/kotlin/com/morihacky/android/rxjava/fragments/MulticastPlaygroundFragment.kt‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
package com.morihacky.android.rxjava.fragments
2+
3+
import android.content.Context
4+
import android.os.Bundle
5+
import android.os.Handler
6+
import android.os.Looper
7+
import android.view.LayoutInflater
8+
import android.view.View
9+
import android.view.ViewGroup
10+
import android.widget.*
11+
import butterknife.BindView
12+
import butterknife.ButterKnife
13+
import butterknife.OnClick
14+
import com.jakewharton.rx.replayingShare
15+
import com.morihacky.android.rxjava.R
16+
import io.reactivex.Observable
17+
import io.reactivex.disposables.Disposable
18+
import java.util.concurrent.TimeUnit
19+
20+
class MulticastPlaygroundFragment : BaseFragment() {
21+
22+
@BindView(R.id.list_threading_log) lateinit var logList: ListView
23+
@BindView(R.id.dropdown) lateinit var pickOperatorDD: Spinner
24+
@BindView(R.id.msg_text) lateinit var messageText: TextView
25+
26+
private lateinit var sharedObservable: Observable<Long>
27+
private lateinit var adapter: LogAdapter
28+
29+
private var logs: MutableList<String> = ArrayList()
30+
private var disposable1: Disposable? = null
31+
private var disposable2: Disposable? = null
32+
33+
override fun onCreateView(inflater: LayoutInflater?,
34+
container: ViewGroup?,
35+
savedInstanceState: Bundle?): View? {
36+
val layout = inflater!!.inflate(R.layout.fragment_multicast_playground, container, false)
37+
ButterKnife.bind(this, layout)
38+
39+
_setupLogger()
40+
_setupDropdown()
41+
42+
return layout
43+
}
44+
45+
@OnClick(R.id.btn_1)
46+
fun onBtn1Click() {
47+
48+
disposable1?.let {
49+
it.dispose()
50+
_log("subscriber 1 disposed")
51+
disposable1 = null
52+
return
53+
}
54+
55+
disposable1 =
56+
sharedObservable
57+
.doOnSubscribe { _log("subscriber 1 (subscribed)") }
58+
.subscribe({ long -> _log("subscriber 1: onNext $long") })
59+
60+
}
61+
62+
@OnClick(R.id.btn_2)
63+
fun onBtn2Click() {
64+
disposable2?.let {
65+
it.dispose()
66+
_log("subscriber 2 disposed")
67+
disposable2 = null
68+
return
69+
}
70+
71+
disposable2 =
72+
sharedObservable
73+
.doOnSubscribe { _log("subscriber 2 (subscribed)") }
74+
.subscribe({ long -> _log("subscriber 2: onNext $long") })
75+
}
76+
77+
@OnClick(R.id.btn_3)
78+
fun onBtn3Click() {
79+
logs = ArrayList<String>()
80+
adapter.clear()
81+
}
82+
83+
// -----------------------------------------------------------------------------------
84+
// Method that help wiring up the example (irrelevant to RxJava)
85+
86+
private fun _log(logMsg: String) {
87+
88+
if (_isCurrentlyOnMainThread()) {
89+
logs.add(0, logMsg + " (main thread) ")
90+
adapter.clear()
91+
adapter.addAll(logs)
92+
} else {
93+
logs.add(0, logMsg + " (NOT main thread) ")
94+
95+
// You can only do below stuff on main thread.
96+
Handler(Looper.getMainLooper()).post {
97+
adapter.clear()
98+
adapter.addAll(logs)
99+
}
100+
}
101+
}
102+
103+
private fun _setupLogger() {
104+
logs = ArrayList<String>()
105+
adapter = LogAdapter(activity, ArrayList<String>())
106+
logList.adapter = adapter
107+
}
108+
109+
private fun _setupDropdown() {
110+
pickOperatorDD.adapter = ArrayAdapter<String>(context,
111+
android.R.layout.simple_spinner_dropdown_item,
112+
arrayOf(".publish().refCount()",
113+
".publish().autoConnect(2)",
114+
".replay(1).autoConnect(2)",
115+
".replay(1).refCount()",
116+
".replayingShare()"))
117+
118+
119+
pickOperatorDD.onItemSelectedListener = object : AdapterView.OnItemSelectedListener {
120+
121+
override fun onItemSelected(p0: AdapterView<*>?, p1: View?, index: Int, p3: Long) {
122+
123+
val sourceObservable = Observable.interval(0L, 3, TimeUnit.SECONDS)
124+
.doOnSubscribe { _log("observer (subscribed)") }
125+
.doOnDispose { _log("observer (disposed)") }
126+
.doOnTerminate { _log("observer (terminated)") }
127+
128+
sharedObservable =
129+
when (index) {
130+
0 -> {
131+
messageText.setText(R.string.msg_demo_multicast_publishRefCount)
132+
sourceObservable.publish().refCount()
133+
}
134+
1 -> {
135+
messageText.setText(R.string.msg_demo_multicast_publishAutoConnect)
136+
sourceObservable.publish().autoConnect(2)
137+
}
138+
2 -> {
139+
messageText.setText(R.string.msg_demo_multicast_replayAutoConnect)
140+
sourceObservable.replay(1).autoConnect(2)
141+
}
142+
3 -> {
143+
messageText.setText(R.string.msg_demo_multicast_replayRefCount)
144+
sourceObservable.replay(1).refCount()
145+
}
146+
4 -> {
147+
messageText.setText(R.string.msg_demo_multicast_replayingShare)
148+
sourceObservable.replayingShare()
149+
}
150+
else -> throw RuntimeException("got to pick an op yo!")
151+
}
152+
}
153+
154+
override fun onNothingSelected(p0: AdapterView<*>?) {}
155+
}
156+
}
157+
158+
private fun _isCurrentlyOnMainThread(): Boolean {
159+
return Looper.myLooper() == Looper.getMainLooper()
160+
}
161+
162+
private inner class LogAdapter(context: Context, logs: List<String>) :
163+
ArrayAdapter<String>(context, R.layout.item_log, R.id.item_log, logs)
164+
165+
}
166+

app/src/main/java/com/morihacky/android/rxjava/fragments/PlaygroundFragment.kt renamed to app/src/main/kotlin/com/morihacky/android/rxjava/fragments/PlaygroundFragment.kt

Lines changed: 10 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -16,25 +16,21 @@ class PlaygroundFragment : BaseFragment() {
1616
private var _logsList: ListView? = null
1717
private var _adapter: LogAdapter? = null
1818

19-
private var _attempt = 0
2019
private var _logs: MutableList<String> = ArrayList()
2120

2221
override fun onCreateView(inflater: LayoutInflater?,
2322
container: ViewGroup?,
2423
savedInstanceState: Bundle?): View? {
25-
return inflater!!.inflate(R.layout.fragment_concurrency_schedulers, container, false)
26-
}
27-
28-
override fun onActivityCreated(savedInstanceState: Bundle?) {
29-
super.onActivityCreated(savedInstanceState)
24+
val view = inflater?.inflate(R.layout.fragment_concurrency_schedulers, container, false)
3025

31-
_logsList = activity.findViewById(R.id.list_threading_log) as ListView
26+
_logsList = view?.findViewById(R.id.list_threading_log) as ListView
27+
_setupLogger()
3228

33-
activity.findViewById(R.id.btn_start_operation).setOnClickListener { _ ->
29+
view.findViewById(R.id.btn_start_operation).setOnClickListener { _ ->
3430
_log("Button clicked")
3531
}
3632

37-
_setupLogger()
33+
return view
3834
}
3935

4036
// -----------------------------------------------------------------------------------
@@ -44,23 +40,23 @@ class PlaygroundFragment : BaseFragment() {
4440

4541
if (_isCurrentlyOnMainThread()) {
4642
_logs.add(0, logMsg + " (main thread) ")
47-
_adapter!!.clear()
48-
_adapter!!.addAll(_logs)
43+
_adapter?.clear()
44+
_adapter?.addAll(_logs)
4945
} else {
5046
_logs.add(0, logMsg + " (NOT main thread) ")
5147

5248
// You can only do below stuff on main thread.
5349
Handler(Looper.getMainLooper()).post {
54-
_adapter!!.clear()
55-
_adapter!!.addAll(_logs)
50+
_adapter?.clear()
51+
_adapter?.addAll(_logs)
5652
}
5753
}
5854
}
5955

6056
private fun _setupLogger() {
6157
_logs = ArrayList<String>()
6258
_adapter = LogAdapter(activity, ArrayList<String>())
63-
_logsList!!.adapter = _adapter
59+
_logsList?.adapter = _adapter
6460
}
6561

6662
private fun _isCurrentlyOnMainThread(): Boolean {

0 commit comments

Comments
 (0)