<dependencies>
<dependency>
<groupId>io.reactivex.rxjava3</groupId>
<artifactId>rxjava</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
import io.reactivex.rxjava3.core.Observable;
public class MyObservable {
public static void main(String[] args) {
Observable<String> observable = Observable.just("Hello", "World");
observable.subscribe(System.out::println);
}
}
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.subjects.PublishSubject;
public class MyWebSocketService {
private WebSocketClient webSocketClient;
private PublishSubject<String> incomingSubject;
public MyWebSocketService() {
webSocketClient = new WebSocketClient();
incomingSubject = PublishSubject.create();
}
public void connect(String url) {
webSocketClient.connect(url);
webSocketClient.setOnMessageReceivedListener(msg -> incomingSubject.onNext(msg));
}
public Observable<String> getIncomingObservable() {
return incomingSubject;
}
public void send(String message) {
webSocketClient.send(message);
}
}
import io.reactivex.rxjava3.core.Observable;
public class MyOperators {
public static void main(String[] args) {
Observable<Integer> numbers = Observable.range(1, 10);
numbers
.filter(n -> n % 2 == 0)
.map(n -> n * 2)
.subscribe(System.out::println);
}
}
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
public class MyErrorHandling {
public static void main(String[] args) {
Observable<String> observable = Observable.create(emitter -> {
try {
emitter.onNext("Success");
emitter.onError(new Exception("Error occurred"));
} catch (Exception e) {
emitter.onError(e);
}
});
Disposable disposable = observable.subscribe(
System.out::println,
error -> System.err.println("Error: " + error.getMessage()),
() -> System.out.println("Completed")
);
disposable.dispose();
}
}
