Export of my JShell session for the Exploring reactive programming in Java session at ConFoo 2019.

confoo-reactive.jsh

Thread  = new Thread(() -> System.out.println("hello confoo"))
Thread t = new Thread(() -> System.out.println("hello confoo"))
t.start()
ExecutorService e = Executors.newSingleThreadExecutor()
Future<String> f = e.submit(() -> "hello confoo")
f
f.get()
ExecutorService e = ForkJoinPool.commonPool()
Future<String> f = e.submit(() -> "hello confoo")
f.get()
CompletableFuture<String> cf = new CompletableFuture<String>()
cf.complete("done")
cf.get()
CompletableFuture<String> cf = new CompletableFuture<String>()
cf.get()
cf.completeExceptionally(new IllegalStateException())
cf.get()
CompletableFuture.supplyAsync(() -> "hello")
.thenApplyAsync(x -> x + " confoo")
.thenAcceptAsync(System.out::println)
CompletableFuture.supplyAsync(() -> "hello")
.thenApplyAsync(x -> x + " confoo")
.exceptionallyAsync(t -> t.getMessage())
.thenAcceptAsync(System.out::println)
CompletableFuture.failedFuture(new IllegalStateException())
.thenApplyAsync(x -> x + " confoo")
.exceptionallyAsync(t -> t.getMessage())
.thenAcceptAsync(System.out::println)
CompletableFuture<String> cf = new CompletableFuture<String>()
cf.completeOnTimeout("timed out", 5, TimeUnit.SECONDS)
cf.get()
CompletableFuture<String> cf = new CompletableFuture<String>()
cf.orTimeout(5, TimeUnit.SECONDS)
cf.get()
CompletableFuture<String> cf = new CompletableFuture<String>()
CompletableFuture<String> copy = cf.copy()
cf.complete("done")
cf
copy
copy.get()
CompletableFuture<String> cf = new CompletableFuture<String>()
CompletableFuture<String> copy = cf.copy()
copy.complete("done")
copy
cf
class SimpleSubscriber implements Subscriber<String> {

public void onSubscribe(Subscription sub) {}

public void onNext(String item) {}

public void onError(Throwable t) {}

public void onComplete() {}

}
class SimpleSubscriber implements Subscriber<String> {

private Subscription sub;

public void onSubscribe(Subscription sub) {
this.sub=sub;
sub.request(1);
}

public void onNext(String item) {
System.out.println("received: " + item;
}

public void onError(Throwable t) {}

public void onComplete() {}

}
class SimpleSubscriber implements Subscriber<String> {

private Subscription sub;

public void onSubscribe(Subscription sub) {
this.sub=sub;
sub.request(1);
}

public void onNext(String item) {
System.out.println("received: " + item;
sub.request(1);
}

public void onError(Throwable t) {
t.printStackTrace();
}

public void onComplete() {
System.out.println("done");
}

}
class SimpleSubscriber implements Subscriber<String> {

private Subscription sub;

public void onSubscribe(Subscription sub) {
this.sub=sub;
sub.request(1);
}

public void onNext(String item) {
System.out.println("received: " + item);
sub.request(1);
}

public void onError(Throwable t) {
t.printStackTrace();
}

public void onComplete() {
System.out.println("done");
}

}
SimpleSubscriber sub = new SimpleSubscriber()
SubmissionPublisher<String> pub = new SubmissionPublisher<String>()
pub.subscribe(sub)
pub.getSubscribers()
pub.submit("hello confoo")
pub.close()
HttpHandler handler = he -> {
String body = "hello confoo";
he.sendResponseHeaders(200, body.length());

try (OutputStream os = he.getResponseBody()){
os.write(body.getBytes());
}
}
/l handler
HttpServer hs = HttpServer.create(new InetSocketAddress(8000), 0)
hs.createContext("/hello", handler)
hs.start()
HttpClient client = HttpClient.newHttpClient()
client.version()
HttpRequest request = HttpRequest.newBuilder().uri(URI.create("http://localhost:8000/hello")).GET().build()
HttpResponse<String> response = client.send(request, BodyHandlers.ofString())
response.statusCode()
response.body()
CompletableFuture<HttpResponse<String>> response = client.sendAsync(request, BodyHandlers.ofString())
response.get().body()