jax-rs and cdi bike the (reactive) bridge

@delabassee @josepaumard#Devoxx

JAX-RS and CDI

Bike the (Reactive) BridgeDavid Delabassée (@delabassee) - Oracle

José Paumard (@josepaumard) - Consultant

`

Copyright © 2017, Oracle and/or its affiliates. All rights reserved. | 2

@delabassee


@JosePaumard

@JosePaumard

https://github.com/JosePaumard

https://www.slideshare.net/jpaumard

https://www.youtube.com/user/JPaumard

Copyright © 2017, Oracle and/or its affiliates. All rights reserved. |

Safe Harbor Statement

The following is intended to outline our general product direction. It is intended for information purposes only, and may not be incorporated into any contract. It is not a commitment to deliver any material, code, or functionality, and should not be relied upon in making purchasing decisions. The development, release, and timing of any features or functionality described for Oracle’s products remains at the sole discretion of Oracle.

5


reactive

adjective | re·ac·tive | \ rē-ˈak-tiv \

1 :of, relating to, or marked by reaction or reactance2 a :readily responsive to a stimulus

https://www.merriam-webster.com/dictionary/reactive

https://www.merriam-webster.com/dictionary/reactive


JAX-RS

8


JAX-RS 2.1

• JSR 370– Java EE 8

• Async– New Reactive Client API

– New method for pausing resquest processing, etc.

• Server-Sent Event support

• JSON-P & JSON-B support

• …

Java API for RESTful Web Services

9


// JAX-RS 2.0

Client client = ClientBuilder.newClient();

WebTarget target = client.target("http://weath.er/api").queryParam("city", "Paris");

Forecast forecast = target.request().get(Forecast.class);

// …

client.close();

JAX-RS Client API

10


JAX-RS Client API

• Fluent API– Client Builder Client Web Target Request building Response

javax.ws.rs.client.Client interface

11

List<Forecast> forecast = ClientBuilder.newClient()

.target("http://weath.er/cities")

.request()

.accept("application/json")

.header("Foo","bar")

.get(new GenericType<List<Forecast>>() {});


JAX-RS Client API

• Synchronous invoker

• Asynchronous invoker

JAX-RS 2.0 Invokers

12

String city = client.target("http://locati.on/api")

.queryParam("city", "Paris")

.request()

.get(String.class);

Future<String> fCity = client.target("http://locati.on/api")


.request()

.async()

.get(String.class);


JAX-RS Client APIAsynchronous invocation

13



.request()

.async()

.get(String.class);

String city = fCity.get();



14



.request()

.async()

.get(String.class);

try {

String city = fCity.get(5, TimeUnit.SECONDS);

} catch(TimeoutException timeout) {

// …

}



15

// Set ClientProperties.CONNECT_TIMEOUT & READ_TIMEOUT



.request()

.async()

.get(String.class);

while ( !fCity.isDone() ) {

// response hasn't been received yet

}

String city = fCity.get();


JAX-RS Client API

• InvocationCallback Interface– javax.ws.rs.client.InvocationCallback<RESPONSE>

• Container will receive async processing events from an invocation– completed(RESPONSE response)

– failed(Throwable throwable)

Asynchronous invocation

16


JAX-RS Client API

WebTarget myResource = client.target("http://examp.le/api/read");

Future<Customer> future = myResource.request(MediaType.TEXT_PLAIN)

.async()

.get(new InvocationCallback<Customer>() {

@Override

public void completed (Customer customer) {

// do something with the customer

}

@Override

public void failed (Throwable throwable) {

// Oops!

}

});

…

InvocationCallback

17


The Travel Service

18


The Travel Service

• Customer details: 150 ms• Recommended destinations: 250 ms• Price calculation for a customer and destination: 170 ms (each)• Weather forecast for a destination: 330 ms (each)

Synchronous

19

5 400 ms


The Travel ServiceAsynchronous

20

730 ms


The Travel Service

21


The Travel Service

22

destination.path("recommended").request().header("Rx-User", "Async").async().get(new InvocationCallback<List<Destination>>() {

@Overridepublic void completed(final List<Destination> recommended) {

final CountDownLatch innerLatch = new CountDownLatch(recommended.size());final Map<String, Forecast> forecasts =

Collections.synchronizedMap(new HashMap<>());for (final Destination dest : recommended) {

forecast.resolveTemplate("dest", dest.getDestination()).request().async().get(new InvocationCallback<Forecast>() {

@Overridepublic void completed(final Forecast forecast) {

forecasts.put(dest.getDestination(), forecast);innerLatch.countDown();

}


The Travel Service

23

// cont.@Overridepublic void failed(final Throwable throwable) {

innerLatch.countDown();}

});}

try {if (!innerLatch.await(10, TimeUnit.SECONDS)) { // timeout }

} catch (final InterruptedException e) { // Ooops, interrupted! }

// Continue with processing…}

@Overridepublic void failed(final Throwable throwable) { // Recommendation error }

});// ...


JAX-RS Client APINew JAX-RS Reactive Invoker

24

// JAX-RS 2.0Response response = client.target(recommandationService)

.request()

.get();

Future<Response> futureResponse = client.target(recommandationService).request().async().get();

// JAX-RS 2.1CompletionStage<Response> completionStageResp = client.target(recommandationService)

.request()

.rx()

.get();


CompletionStage API

• A model for a Task– That performs an action and may return a value

– That can be triggered by another task

– That may trigger another task

– That can be executed in a different thread

• A CompletionStage is an element of an asynchronous chain

25


Leveraging the CompletionStage API

26

CS1 CS21 thenApply(Function<T, U> function);

CS31

CS41

CSA

thenCombine(CompletionStage<U> other, BiFunction<T, U> combiner);

thenCompose(Function<T, CompletionStage<U>> function)


CompletionStage Pipeline

27

CS1 CS21

CS22

CS31 CS41

CSA


The Travel Service

28

CompletionStage<JsonObject> queryForecastCS = client.target("forecast").queryParam("format", "json").request().rx().get(JsonObject.class);

Function<JsonObject, Forecast> unmarshallForecast = jsonObject -> JsonBuilder.create().fromJson(jsonObject.toString(), Forecast.class);

Function<Destination, CompletionStage<Void>> populateWithForecast = destination ->

queryForecastCS.thenApply(unmarshallForecast).thenAccept(forecast -> destination.setForecast(forecast));

Function<Destination, CompletionStage<Void>> populateWithQuotation = destination ->

queryQuotationCS.thenApply(unmarshallQuotation).thenAccept(quotation -> destination.setQuotation(quotation));


The Travel Service

29

Function<Destination, CompletableFuture<Void>> populateDestination =destination ->

CompletableFuture.allOf(populateWithForecast.apply(destination).toCompletableFuture(),populateWithQuotation.apply(destination).toCompletableFuture()

).toCompletableFuture();


The Travel Service

30

Function<Destination, CompletableFuture<Void>> populateDestination =destination ->

CompletableFuture.allOf(populateWithForecast.apply(destination).toCompletableFuture(),populateWithQuotation.apply(destination).toCompletableFuture()

).toCompletableFuture();

Function<List<Destination>, CompletionStage<Void>> populateDestinations =destinations ->

CompletableFuture.allOf(destinations.stream().map(populateDestination)

.toArray(CompletableFuture[]::new));


The Travel Service

31

CompletionStage<Void> destinationsUpdated =destinationsCS.thenCompose(populateDestinations);

CompletionStage<List<Destination>> updatedDestinationsCS =destinationsUpdated.thenCompose(v -> destinationsCS);

• Then we can update some interface on the completion of updatedDestination and / or return it


The Travel Service

32

@GETpublic void populateDestinations(@Suspended final AsyncResponse asyncResponse) {

CompletionStage<List<Destination>> destinationCS = client.target("destination").queryParam("format", "json").request().rx().get(/* some JSONB code */);/* some more technical CS code */

CompletionStage<List<Destination>> updatedDestinationsCS = destinationCS.thenCompose(populateDestinations)

.thenCompose(v -> destinationsCS);

asyncResponse.resume(updatedDestinationsCS.toCompletableFuture().get());}


The Travel Service

33

@GETpublic CompletionStage<List<Destination>> populateDestinations() {

CompletionStage<List<Destination>> destinationCS = client.target("destination").queryParam("format", "json").request().rx().get(/* some JSONB code */);/* some more technical CS code */

CompletionStage<List<Destination>> updatedDestinationsCS = destinationCS.thenCompose(populateDestinations)

.thenCompose(v -> destinationsCS);

return updatedDestinationsCS;}


Exception Handling

34

CS1 CS21

CS22

CS31 CS41

CS32


Exception Handling

Returns a new CompletionStage

• That completes when the CS completes

• Either with the same result (normal completion)

• Or with the transformed exception

35

exceptionaly()

stage.exceptionaly( // Functionexception -> doSomethingNotTooStupidWith(exception));


Exception Handling


• That completes when the CS completes

• Calls the BiFunction with a null as result or exception

36

handle()

stage.handle( // BiFunction(result, exception) -> doSomethingWith(result, exception));


Exception Handling


• With the same result or exception as this stage

• That executes the given action when this stage completes

37

whenComplete()

stage.whenComplete( // BiConsumer + async version(result, exception) -> doSomethingWith(result, exception));


Exception Handling

38

CompletionStage<Void> quotation = client.target("quotation").request().rx().get(JsonObject.class).thenApply(unmarshallQuotation).exceptionnaly(throwable -> null).thenAccept(destination::setQuotation);


Exception Handling

39

CompletionStage<Void> quotation = client.target("quotation").request().rx().get(JsonObject.class).thenApply(unmarshallQuotation).handle(((quotation, throwable) -> {

if (throwable == null) {destination.setQuotation(quotation);

} else {// try to do something smart with the exception

}}


JAX-RS Reactive Extensions

• Supported on all HTTP Methods of the Client API– DELETE– GET– HEAD– OPTIONS– POST– PUT– TRACE

40


JAX-RS Reactive Extensions

• Implementations MUST support an invoker for CompletionStage

• Implementations MAY support other reactive APIs

• Jersey– CompletionStageRxInvoker (Default)– RxListenableFutureInvoker – Guava

41

https://github.com/jersey/jersey/tree/master/ext/rx

client.register(RxFlowableInvokerProvider.class);

client.target(...)...

.rx(RxFlowableInvoker.class)

.get();

– RxObservableInvoker – RxJava– RxFlowableInvoker – RxJava2

https://github.com/jersey/jersey/tree/master/ext/rx


CDI

42


CDI 2.0

• JSR 365– Java EE 8

• Java SE focus– Modular specification

– CDI Container bootstraping

• Obersevers Ordering

• Asynchronous Events

• …

Context and Dependency Injection

43


Asynchronous Events

44

// Producer@InjectEvent<Payload> event;

public void aCriticalBusinessMethod() {

CompletionStage<Payload> cs = event.fireAsync(new Payload());}

// Consumerpublic void anOberser(@ObservesAsync Payload event) {

// do something with the payload}


Asynchronous Events

45

// Producer@InjectEvent<Payload> event;

public void aCriticalBusinessMethod() {

CompletionStage<Payload> cs = event.fireAsync(new Payload(), executor);

}


Wrap-up

47


Java EE 8 – Modernization & Simplification

48

CDI 2.0

JSON-B 1.0 (*)

Security 1.0 (*)

Bean Validation 2.0

JSF 2.3

Servlet 4.0

JSON-P 1.1

JAX-RS 2.1 Reactive Client API, Server-Sent Events, …

HTTP/2, Server Push, …

Java <-> JSON binding

Updates to JSON standards, JSON Collectors, …

Async Event, Observers ordering, SE support, …

Embrace Java SE 8, new constraints, …

Improved CDI, WebSocket, SE 8 integration, …

Portable Identity Store, Authentication & Security Context


Thanks!

jax-rs and cdi bike the (reactive) bridge

Education