Skip to content

Commit d73fb26

Browse files
author
Nickolay Savchenko
committed
Fix leaks on disconnect
1 parent de511b0 commit d73fb26

File tree

6 files changed

+57
-40
lines changed

6 files changed

+57
-40
lines changed

‎example-client/build.gradle‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,8 @@ android {
4646
dependencies {
4747
implementation 'androidx.appcompat:appcompat:1.0.2'
4848
implementation 'androidx.recyclerview:recyclerview:1.0.0'
49-
implementation 'org.java-websocket:Java-WebSocket:1.3.6'
49+
// implementation 'org.java-websocket:Java-WebSocket:1.3.6'
50+
implementation 'com.squareup.okhttp3:okhttp:3.12.1'
5051
implementation 'com.android.support:recyclerview-v7:28.0.0'
5152
// RxJava
5253
implementation 'io.reactivex.rxjava2:rxjava:2.2.5'

‎example-client/src/main/AndroidManifest.xml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,8 @@
1010
android:icon="@mipmap/ic_launcher"
1111
android:label="@string/app_name"
1212
android:supportsRtl="true"
13-
android:theme="@style/AppTheme">
13+
android:theme="@style/AppTheme"
14+
android:networkSecurityConfig="@xml/network_security_config">
1415
<activity android:name="ua.naiksoftware.stompclientexample.MainActivity">
1516
<intent-filter>
1617
<action android:name="android.intent.action.MAIN" />

‎example-client/src/main/java/ua/naiksoftware/stompclientexample/MainActivity.java‎

Lines changed: 12 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -51,16 +51,13 @@ protected void onCreate(Bundle savedInstanceState) {
5151
mAdapter.setHasStableIds(true);
5252
mRecyclerView.setAdapter(mAdapter);
5353
mRecyclerView.setLayoutManager(new LinearLayoutManager(this, LinearLayoutManager.VERTICAL, true));
54+
55+
mStompClient = Stomp.over(Stomp.ConnectionProvider.OKHTTP, "ws://" + ANDROID_EMULATOR_LOCALHOST
56+
+ ":" + RestClient.SERVER_PORT + "/example-endpoint/websocket");
5457
}
5558

5659
public void disconnectStomp(View view) {
5760
mStompClient.disconnect();
58-
59-
if (compositeDisposable != null) {
60-
compositeDisposable.dispose();
61-
62-
compositeDisposable = null;
63-
}
6461
}
6562

6663
public static final String LOGIN = "login";
@@ -73,11 +70,9 @@ public void connectStomp(View view) {
7370
headers.add(new StompHeader(LOGIN, "guest"));
7471
headers.add(new StompHeader(PASSCODE, "guest"));
7572

76-
mStompClient = Stomp.over(Stomp.ConnectionProvider.JWS, "ws://" + ANDROID_EMULATOR_LOCALHOST
77-
+ ":" + RestClient.SERVER_PORT + "/example-endpoint/websocket");
78-
7973
mStompClient.withClientHeartbeat(30000).withServerHeartbeat(30000);
8074

75+
clearSubscriptions();
8176
compositeDisposable = new CompositeDisposable();
8277

8378
Disposable dispLifecycle = mStompClient.lifecycle()
@@ -94,6 +89,7 @@ public void connectStomp(View view) {
9489
break;
9590
case CLOSED:
9691
toast("Stomp connection closed");
92+
clearSubscriptions();
9793
break;
9894
case FAILED_SERVER_HEARTBEAT:
9995
toast("Stomp failed server heartbeat");
@@ -158,14 +154,17 @@ protected CompletableTransformer applySchedulers() {
158154
.observeOn(AndroidSchedulers.mainThread());
159155
}
160156

161-
@Override
162-
protected void onDestroy() {
163-
mStompClient.disconnect();
164-
157+
private void clearSubscriptions() {
165158
if (compositeDisposable != null) {
166159
compositeDisposable.dispose();
160+
167161
compositeDisposable = null;
168162
}
163+
}
164+
165+
@Override
166+
protected void onDestroy() {
167+
mStompClient.disconnect();
169168

170169
if (mRestPingDisposable != null) mRestPingDisposable.dispose();
171170
super.onDestroy();

‎example-client/src/main/res/xml/network_security_config.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
<?xml version="1.0" encoding="utf-8"?>
2+
<network-security-config>
3+
<domain-config cleartextTrafficPermitted="true">
4+
<domain includeSubdomains="true">10.0.2.2</domain>
5+
</domain-config>
6+
</network-security-config>

‎lib/src/main/AndroidManifest.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
<manifest package="com.github.forresthopkinsa">
1+
<manifest package="ua.naiksoftware.stomp">
22

33
<application />
44

‎lib/src/main/java/ua/naiksoftware/stomp/StompClient.java‎

Lines changed: 34 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package ua.naiksoftware.stomp;
22

3+
import android.annotation.SuppressLint;
34
import android.support.annotation.NonNull;
45
import android.support.annotation.Nullable;
56
import android.util.Log;
@@ -55,7 +56,6 @@ public class StompClient {
5556

5657
public StompClient(ConnectionProvider connectionProvider) {
5758
mConnectionProvider = connectionProvider;
58-
mMessageStream = PublishSubject.create();
5959
mStreamMap = new ConcurrentHashMap<>();
6060
mConnectionStream = BehaviorSubject.createDefault(false);
6161
pathMatcher = new SimplePathMatcher();
@@ -123,14 +123,11 @@ public void connect(@Nullable List<StompHeader> _headers) {
123123

124124
case CLOSED:
125125
Log.d(TAG, "Socket closed");
126-
setConnected(false);
127-
isConnecting = false;
126+
disconnect();
128127
break;
129128

130129
case ERROR:
131130
Log.d(TAG, "Socket closed with error");
132-
setConnected(false);
133-
isConnecting = false;
134131
break;
135132

136133
case FAILED_SERVER_HEARTBEAT:
@@ -143,14 +140,21 @@ public void connect(@Nullable List<StompHeader> _headers) {
143140
isConnecting = true;
144141
mMessagesDisposable = mConnectionProvider.messages()
145142
.map(StompMessage::from)
146-
.doOnNext(this::callSubscribers)
143+
.doOnNext(getMessageStream()::onNext)
147144
.filter(msg -> msg.getStompCommand().equals(StompCommand.CONNECTED))
148145
.subscribe(stompMessage -> {
149146
setConnected(true);
150147
isConnecting = false;
151148
});
152149
}
153150

151+
private PublishSubject<StompMessage> getMessageStream() {
152+
if (mMessageStream == null || mMessageStream.hasComplete()) {
153+
mMessageStream = PublishSubject.create();
154+
}
155+
return mMessageStream;
156+
}
157+
154158
private void setConnected(boolean connected) {
155159
mConnected = connected;
156160
mConnectionStream.onNext(mConnected);
@@ -159,6 +163,7 @@ private void setConnected(boolean connected) {
159163
/**
160164
* Disconnect from server, and then reconnect with the last-used headers
161165
*/
166+
@SuppressLint("CheckResult")
162167
public void reconnect() {
163168
disconnectCompletable()
164169
.subscribe(() -> connect(mHeaders),
@@ -186,33 +191,33 @@ public Completable send(@NonNull StompMessage stompMessage) {
186191
.startWith(connectionComplete);
187192
}
188193

189-
private void callSubscribers(StompMessage stompMessage) {
190-
mMessageStream.onNext(stompMessage);
191-
}
192-
193194
public Flowable<LifecycleEvent> lifecycle() {
194195
return mConnectionProvider.lifecycle().toFlowable(BackpressureStrategy.BUFFER);
195196
}
196197

197198
public void disconnect() {
198-
disconnectCompletable().subscribe(() -> {}, e -> Log.e(tag, "Disconnect error", e));
199-
if(mStreamMap != null && !mStreamMap.isEmpty()){
200-
mStreamMap.clear();
201-
}
202-
if(mTopics != null && !mTopics.isEmpty()){
203-
mTopics.clear();
199+
if (!mConnected) {
200+
Log.d(TAG, "Skip disconnect, already not connected");
201+
return;
204202
}
203+
disconnectCompletable().subscribe(() -> {}, e -> Log.e(tag, "Disconnect error", e));
205204
}
206205

207206
public Completable disconnectCompletable() {
208-
if (mLifecycleDisposable != null) {
209-
mLifecycleDisposable.dispose();
210-
}
211-
if (mMessagesDisposable != null) {
212-
mMessagesDisposable.dispose();
213-
}
214207
return mConnectionProvider.disconnect()
215-
.doOnComplete(() -> setConnected(false));
208+
.doOnComplete(() -> {
209+
setConnected(false);
210+
isConnecting = false;
211+
if (mLifecycleDisposable != null) {
212+
mLifecycleDisposable.dispose();
213+
}
214+
if (mMessagesDisposable != null) {
215+
mMessagesDisposable.dispose();
216+
}
217+
if (mMessageStream != null && !mMessageStream.hasComplete()) {
218+
mMessageStream.onComplete();
219+
}
220+
});
216221
}
217222

218223
public Flowable<StompMessage> topic(String destinationPath) {
@@ -224,7 +229,7 @@ public Flowable<StompMessage> topic(@NonNull String destPath, List<StompHeader>
224229
return Flowable.error(new IllegalArgumentException("Topic path cannot be null"));
225230
else if (!mStreamMap.containsKey(destPath))
226231
mStreamMap.put(destPath,
227-
mMessageStream
232+
getMessageStream()
228233
.filter(msg -> pathMatcher.matches(destPath, msg))
229234
.toFlowable(BackpressureStrategy.BUFFER)
230235
.doOnSubscribe(disposable -> subscribePath(destPath, headerList).subscribe())
@@ -294,6 +299,11 @@ private Completable unsubscribePath(String dest) {
294299

295300
Log.d(TAG, "Unsubscribe path: " + dest + " id: " + topicId);
296301

302+
if (!mConnected) {
303+
Log.d(TAG, "Not connected, skip sending Unsubscribe frame to " + dest);
304+
return Completable.complete();
305+
}
306+
297307
return send(new StompMessage(StompCommand.UNSUBSCRIBE,
298308
Collections.singletonList(new StompHeader(StompHeader.ID, topicId)), null));
299309
}

0 commit comments

Comments
 (0)