Skip to content

Commit 0b18b86

Browse files
NaikSoftwareNaikSoftware
committed
Refactor
1 parent f5d191e commit 0b18b86

File tree

6 files changed

+69
-64
lines changed

6 files changed

+69
-64
lines changed

‎build.gradle‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ buildscript {
66
maven {
77
url "https://maven.google.com"
88
}
9+
google()
910
}
1011
dependencies {
1112
classpath 'com.android.tools.build:gradle:3.0.1'
@@ -20,6 +21,7 @@ allprojects {
2021
repositories {
2122
jcenter()
2223
maven { url "https://jitpack.io" }
24+
google()
2325
}
2426
}
2527

‎example-client/build.gradle‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,8 @@
11
apply plugin: 'com.android.application'
2-
apply plugin: 'me.tatarka.retrolambda'
32

43
android {
54
compileSdkVersion 25
6-
buildToolsVersion "25.0.2"
5+
buildToolsVersion '26.0.2'
76

87
defaultConfig {
98
applicationId "ua.naiksoftware.stompclientexample"
@@ -29,7 +28,7 @@ dependencies {
2928
compile fileTree(dir: 'libs', include: ['*.jar'])
3029
testCompile 'junit:junit:4.12'
3130
compile 'com.android.support:appcompat-v7:25.3.1'
32-
compile 'org.java-websocket:java-websocket:1.3.2'
31+
compile 'org.java-websocket:Java-WebSocket:1.3.6'
3332
compile 'com.android.support:recyclerview-v7:25.3.1'
3433
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
3534
compile 'com.squareup.retrofit2:converter-gson:2.3.0'

‎lib/build.gradle‎

Lines changed: 5 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -28,15 +28,12 @@ android {
2828

2929

3030
dependencies {
31-
compile fileTree(include: ['*.jar'], dir: 'libs')
32-
testCompile 'junit:junit:4.12'
33-
compile 'io.reactivex:rxjava:1.3.0'
34-
compile 'net.sourceforge.streamsupport:streamsupport:1.5.5'
35-
compile 'net.sourceforge.streamsupport:streamsupport-cfuture:1.5.5'
31+
implementation "io.reactivex.rxjava2:rxjava:2.1.8"
3632
// Supported transports
37-
compile 'org.java-websocket:Java-WebSocket:1.3.6'
38-
compile 'com.squareup.okhttp3:okhttp:3.8.1'
39-
implementation 'com.android.support:support-annotations:24.2.0'
33+
compileOnly 'org.java-websocket:Java-WebSocket:1.3.6'
34+
compileOnly 'com.squareup.okhttp3:okhttp:3.9.1'
35+
36+
implementation 'com.android.support:support-annotations:27.1.0'
4037
}
4138

4239
task sourcesJar(type: Jar) {

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

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,13 @@
66

77
import java.util.concurrent.TimeUnit;
88

9-
import rx.Completable;
10-
import rx.Observable;
11-
import rx.subjects.BehaviorSubject;
12-
import rx.subjects.PublishSubject;
9+
import io.reactivex.BackpressureStrategy;
10+
import io.reactivex.Completable;
11+
import io.reactivex.CompletableSource;
12+
import io.reactivex.Flowable;
13+
import io.reactivex.Observable;
14+
import io.reactivex.subjects.BehaviorSubject;
15+
import io.reactivex.subjects.PublishSubject;
1316

1417
/**
1518
* Created by forresthopkinsa on 8/8/2017.
@@ -30,7 +33,7 @@ abstract class AbstractConnectionProvider implements ConnectionProvider {
3033
AbstractConnectionProvider() {
3134
mLifecycleStream = PublishSubject.create();
3235
mMessagesStream = PublishSubject.create();
33-
mConnectionStream = BehaviorSubject.create(false);
36+
mConnectionStream = BehaviorSubject.create();
3437
}
3538

3639
@NonNull
@@ -51,25 +54,23 @@ public Observable<String> messages() {
5154

5255
@Override
5356
public Completable disconnect() {
54-
Observable<Boolean> ex = Observable.error(new IllegalStateException("Attempted to disconnect when already disconnected"));
57+
CompletableSource ex = Completable.error(new IllegalStateException("Attempted to disconnect when already disconnected"));
5558

5659
Completable block = mConnectionStream
57-
.first(isConnected -> isConnected)
58-
.timeout(1, TimeUnit.SECONDS, ex)
59-
.toCompletable();
60+
.filter(connected -> connected).firstOrError().toCompletable()
61+
.timeout(1, TimeUnit.SECONDS, ex);
6062

6163
return Completable
6264
.fromAction(this::rawDisconnect)
6365
.startWith(block);
6466
}
6567

6668
private Completable initSocket() {
67-
Observable<Boolean> ex = Observable.error(new IllegalStateException("Attempted to connect when already connected"));
69+
CompletableSource ex = Completable.error(new IllegalStateException("Attempted to connect when already connected"));
6870

6971
Completable block = mConnectionStream
70-
.first(isConnected -> !isConnected)
71-
.timeout(1, TimeUnit.SECONDS, ex)
72-
.toCompletable();
72+
.filter(connected -> !connected).firstOrError().toCompletable()
73+
.timeout(1, TimeUnit.SECONDS, ex);
7374

7475
return Completable
7576
.fromAction(this::createWebSocketConnection)
@@ -140,7 +141,12 @@ void emitMessage(String stompMessage) {
140141

141142
@NonNull
142143
@Override
143-
public Observable<LifecycleEvent> getLifecycleReceiver() {
144+
public Observable<LifecycleEvent> lifecycle() {
144145
return mLifecycleStream;
145146
}
147+
148+
@Override
149+
public Flowable<Boolean> connected() {
150+
return mConnectionStream.toFlowable(BackpressureStrategy.LATEST);
151+
}
146152
}

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

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

3-
import rx.Completable;
4-
import rx.Observable;
3+
import io.reactivex.Completable;
4+
import io.reactivex.Flowable;
5+
import io.reactivex.Observable;
56

67
/**
78
* Created by naik on 05.05.16.
@@ -23,7 +24,7 @@ public interface ConnectionProvider {
2324
/**
2425
* Subscribe this for receive #LifecycleEvent events
2526
*/
26-
Observable<LifecycleEvent> getLifecycleReceiver();
27+
Observable<LifecycleEvent> lifecycle();
2728

2829
/**
2930
* Disconnects from server. This is basically a Callable.
@@ -32,4 +33,6 @@ public interface ConnectionProvider {
3233
Completable disconnect();
3334

3435
Completable setHeartbeat(int ms);
36+
37+
Flowable<Boolean> connected();
3538
}

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

Lines changed: 34 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,12 @@
1010
import java.util.UUID;
1111
import java.util.concurrent.ConcurrentHashMap;
1212

13-
import java8.util.StringJoiner;
14-
import java8.util.concurrent.CompletableFuture;
15-
import rx.Completable;
16-
import rx.Observable;
17-
import rx.Subscription;
18-
import rx.schedulers.Schedulers;
19-
import rx.subjects.PublishSubject;
13+
import io.reactivex.BackpressureStrategy;
14+
import io.reactivex.Completable;
15+
import io.reactivex.CompletableSource;
16+
import io.reactivex.Flowable;
17+
import io.reactivex.disposables.Disposable;
18+
import io.reactivex.subjects.PublishSubject;
2019
import ua.naiksoftware.stomp.ConnectionProvider;
2120
import ua.naiksoftware.stomp.LifecycleEvent;
2221
import ua.naiksoftware.stomp.StompHeader;
@@ -39,28 +38,20 @@ public class StompClient {
3938
private boolean legacyWhitespace;
4039

4140
private PublishSubject<StompMessage> mMessageStream;
42-
private CompletableFuture<Boolean> mConnectionFuture;
43-
private Completable mConnectionComplete;
44-
private ConcurrentHashMap<String, Observable<StompMessage>> mStreamMap;
41+
private ConcurrentHashMap<String, Flowable<StompMessage>> mStreamMap;
4542
private Parser parser;
46-
private Subscription lifecycleSub;
47-
private Subscription messagesSubscription;
43+
private Disposable mLifecycleDisposable;
44+
private Disposable mMessagesDisposable;
4845
private List<StompHeader> mHeaders;
4946
private int heartbeat;
5047

5148
public StompClient(ConnectionProvider connectionProvider) {
5249
mConnectionProvider = connectionProvider;
5350
mMessageStream = PublishSubject.create();
5451
mStreamMap = new ConcurrentHashMap<>();
55-
resetStatus();
5652
parser = Parser.NONE;
5753
}
5854

59-
private void resetStatus() {
60-
mConnectionFuture = new CompletableFuture<>();
61-
mConnectionComplete = Completable.fromFuture(mConnectionFuture).subscribeOn(Schedulers.newThread());
62-
}
63-
6455
public enum Parser {
6556
NONE,
6657
RABBITMQ
@@ -109,7 +100,7 @@ public void connect(@Nullable List<StompHeader> _headers) {
109100
mHeaders = _headers;
110101

111102
if (mConnected) return;
112-
lifecycleSub = lifecycle()
103+
mLifecycleDisposable = mConnectionProvider.lifecycle()
113104
.subscribe(lifecycleEvent -> {
114105
switch (lifecycleEvent.getType()) {
115106
case OPENED:
@@ -134,14 +125,14 @@ public void connect(@Nullable List<StompHeader> _headers) {
134125
});
135126

136127
isConnecting = true;
137-
messagesSubscription = mConnectionProvider.messages()
128+
mMessagesDisposable = mConnectionProvider.messages()
138129
.map(StompMessage::from)
139130
.doOnNext(this::callSubscribers)
140131
.filter(msg -> msg.getStompCommand().equals(StompCommand.CONNECTED))
141132
.subscribe(stompMessage -> {
142133
mConnected = true;
143134
isConnecting = false;
144-
mConnectionFuture.complete(true);
135+
// mConnectionFuture.complete(true);
145136
});
146137
}
147138

@@ -166,37 +157,42 @@ public Completable send(String destination, String data) {
166157

167158
public Completable send(@NonNull StompMessage stompMessage) {
168159
Completable completable = mConnectionProvider.send(stompMessage.compile(legacyWhitespace));
169-
return completable.startWith(mConnectionComplete);
160+
CompletableSource connectionComplete = mConnectionProvider.connected()
161+
.filter(isConnected -> isConnected)
162+
.firstOrError().toCompletable();
163+
return completable
164+
.startWith(mConnectionProvider.disconnect())
165+
.startWith(connectionComplete);
170166
}
171167

172168
private void callSubscribers(StompMessage stompMessage) {
173169
mMessageStream.onNext(stompMessage);
174170
}
175171

176-
public Observable<LifecycleEvent> lifecycle() {
177-
return mConnectionProvider.getLifecycleReceiver();
172+
public Flowable<LifecycleEvent> lifecycle() {
173+
return mConnectionProvider.lifecycle().toFlowable(BackpressureStrategy.BUFFER);
178174
}
179175

180176
public void disconnect() {
181-
resetStatus();
182-
lifecycleSub.unsubscribe();
183-
messagesSubscription.unsubscribe();
177+
mLifecycleDisposable.dispose();
178+
mMessagesDisposable.dispose();
184179
mConnectionProvider.disconnect().subscribe(() -> mConnected = false);
185180
}
186181

187-
public Observable<StompMessage> topic(String destinationPath) {
182+
public Flowable<StompMessage> topic(String destinationPath) {
188183
return topic(destinationPath, null);
189184
}
190185

191-
public Observable<StompMessage> topic(@NonNull String destPath, List<StompHeader> headerList) {
186+
public Flowable<StompMessage> topic(@NonNull String destPath, List<StompHeader> headerList) {
192187
if (destPath == null)
193-
return Observable.error(new IllegalArgumentException("Topic path cannot be null"));
188+
return Flowable.error(new IllegalArgumentException("Topic path cannot be null"));
194189
else if (!mStreamMap.containsKey(destPath))
195190
mStreamMap.put(destPath,
196191
mMessageStream
197192
.filter(msg -> matches(destPath, msg))
198-
.doOnSubscribe(() -> subscribePath(destPath, headerList).subscribe())
199-
.doOnUnsubscribe(() -> unsubscribePath(destPath).subscribe())
193+
.toFlowable(BackpressureStrategy.BUFFER)
194+
.doOnSubscribe(disposable -> subscribePath(destPath, headerList).subscribe())
195+
.doFinally(() -> unsubscribePath(destPath).subscribe())
200196
.share()
201197
);
202198
return mStreamMap.get(destPath);
@@ -250,10 +246,12 @@ private boolean matches(String path, StompMessage msg) {
250246
}
251247
}
252248
// at this point, 'transformed' looks like ["lorem", "ipsum", "[^.]+", "sit"]
253-
StringJoiner sj = new StringJoiner("\\.");
254-
for (String s : transformed)
255-
sj.add(s);
256-
String join = sj.toString();
249+
StringBuilder sb = new StringBuilder();
250+
for (String s : transformed) {
251+
if (sb.length() > 0) sb.append("\\.");
252+
sb.append(s);
253+
}
254+
String join = sb.toString();
257255
// join = "lorem\.ipsum\.[^.]+\.sit"
258256

259257
ret = dest.matches(join);

0 commit comments

Comments
 (0)