Using Java CompletionStage In Asynchronous Programming

Using Java CompletionStage inAsynchronous ProgrammingDEV4798Douglas SurberOracle Database JDBC ArchitectDatabase Server TechnologiesOctober 25, 2018Copyright 2018, Oracle and/or its affiliates. All rights reserved.

Safe Harbor StatementThe following is intended to outline our general product direction. It is intended forinformation purposes only, and may not be incorporated into any contract. It is not acommitment to deliver any material, code, or functionality, and should not be relied uponin making purchasing decisions. The development, release, timing, and pricing of anyfeatures or functionality described for Oracle’s products may change and remains at thesole discretion of Oracle Corporation.Copyright 2018, Oracle and/or its affiliates. All rights reserved. 2

Introduction to CompletionStageWhat are java.util.concurrent.CompletionStageand java.util.concurrent.CompletableFuture?Copyright 2018, Oracle and/or its affiliates. All rights reserved. 3

CompletableFuture:The Promises of Java[DEV5375]Venkat SubramaniamWednesday, October 24Copyright 2018, Oracle and/or its affiliates. All rights reserved. 4

Hands-on Lab: The Asynchronous JavaDatabase Access Driver[HOL4799]Douglas SurberWednesday, October 24Copyright 2018, Oracle and/or its affiliates. All rights reserved. 5

java.util.concurrent.CompletionStagepublic interface CompletionStage T A stage of a possibly asynchronous computation, that performs an actionor computes a value when another CompletionStage completes. A stagecompletes upon termination of its computation, but this may in turn triggerother dependent stages. stage.thenApply(x - square(x)).thenAccept(x - System.out.print(x)).thenRun(() - System.out.println());Copyright 2018, Oracle and/or its affiliates. All rights reserved. 6

java.util.concurrent.CompletableFuturepublic class CompletableFuture T implements Future T , CompletionStage T Both a CompletionStage and a Future A Future that may be explicitly completed (setting its value and status),and may be used as a CompletionStage, supporting dependentfunctions and actions that trigger upon its completion. CompletableFuture future .;future.complete(value);future.get();Copyright 2018, Oracle and/or its affiliates. All rights reserved. 7

ExamplesupplyAsync(Supplier supplier), thenApply(Function function)CompletionStage task CompleteableFuture.supplyAsync(() - 10);CompletionStage squareTask task.thenApply( v - v * v );Copyright 2018, Oracle and/or its affiliates. All rights reserved. 8

supplyAsync(Supplier blic static U CompletableFuture U supplyAsync(Supplier U supplier)Returns a new CompletableFuture that is asynchronously completed by atask running in the ForkJoinPool.commonPool() with the value obtained bycalling the given Supplier.Type Parameters: U - the function's return typeParameters: supplier - a function returning the value to be used to completethe returned CompletableFutureReturns: the new CompletableFutureCopyright 2018, Oracle and/or its affiliates. All rights reserved. 9

thenApply(Function ? super T,? extends U fn)java.util.concurrent.CompletionStage U CompletionStage U thenApply(Function ? super T,? extends U fn)Returns a new CompletionStage that, when this stage completes normally, isexecuted with this stage's result as the argument to the supplied function.Type Parameters: U - the function's return typeParameters: fn - the function to use to compute the value of the returnedCompletionStageReturns: the new CompletionStageCopyright 2018, Oracle and/or its affiliates. All rights reserved. 10

ExampleCompletionStage task CompleteableFuture.supplyAsync( () - 10 );CompletionStage squareTask task.thenApply( v - v * v);*task() - 10task() - 10thenApply †squareTask* CompletionStageCopyright 2018, Oracle and/or its affiliates. All rights reserved. v - v * v†dependency11

Example Executiontasktask() - 10() - 10thenApplysquareTaskv - v * vthenApplysquareTaskCopyright 2018, Oracle and/or its affiliates. All rights reserved. v - v * v12

Introduction to AoJWhat are AoJ and ADBA?Copyright 2018, Oracle and/or its affiliates. All rights reserved. 13

Asynchronous Database Access (ADBA)Proposed Java Standard What: Java standard database access API that never blocks user threads Who: Developed by the JDBC Community, JDBC Expert Group and Oracle When: Targeted for a near future release, Java 14 perhaps Why: Async apps have better scalability– Fewer threads means less thread scheduling, less thread contention– Database access is slow so blocked threads leave resources idle for a long time pyright 2018, Oracle and/or its affiliates. All rights reserved. 14

ADBA ExampleSelect some items from a tablepublic CompletionStage List Item itemsForAnswer(DataSource ds, int answer) {String sql "select id, name, answer from tab where answer :target";try (Session session ds.getSession()) {return session. List Item rowOperation(sql).set("target", answer, AdbaType.NUMERIC).collect(Collectors.mapping(row - new nStage();}}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 15

ADBA over JDBC (AoJ)Open Source implementation of ADBA using any JDBC as a backendDataSource ds ); master/java/AoJCopyright 2018, Oracle and/or its affiliates. All rights reserved. 16

Using CompletionStageCopyright 2018, Oracle and/or its affiliates. All rights reserved. 17

ADBA ExampleSelect some items from a tablepublic CompletionStage List Item itemsForAnswer(DataSource ds, int answer) {String sql "select id, name, answer from tab where answer :target";try (Session session ds.getSession()) {return session. List Item rowOperation(sql).set("target", answer, AdbaType.NUMERIC).collect(Collectors.mapping(row - new nStage();}}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 18

submit()com.oracle.adbaoverjdbc.Operationpublic Submission T submit() {if (isImmutable()) {throw new IllegalStateException("TODO");}immutable();return group.submit(this);}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 19

submit(Operation n S submit(Operation S op) {memberTail etExecutor()));return Submission.submit(this::cancel, memberTail);}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 20

rationGroupfinal CompletionStage T attachCompletionHandler(CompletionStage T result) {return result.handle((r, t) - {Throwable ex unwrapException(t);checkAbort(ex);if (t null)return handleResult(r);elsethrow handleError(ex);});}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 21

handle(BiFunction ?, Throwable, ? fn)java.util.concurrent.CompletionStage U CompletionStage U handle(BiFunction ? super T, Throwable, ? extends U fn)Returns a new CompletionStage that, when this stage completes either normally orexceptionally, is executed with this stage's result and exception as arguments to thesupplied function. When this stage is complete, the given function is invoked with theresult (or null if none) and the exception (or null if none) of this stage as arguments, andthe function's result is used to complete the returned stage.Type Parameters: U - the function's return typeParameters: fn - the function to use to compute the value of the returnedCompletionStageReturns: the new CompletionStageCopyright 2018, Oracle and/or its affiliates. All rights reserved. 22

attachCompletionHandlermemberTail follows(.).handle( (r, t) - { . } )memberTailResult ofprevious opResult ofprevious opfollows(.)handlememberTailCopyright 2018, Oracle and/or its affiliates. All rights reserved. (r,t) - {.}23

(r, t) - { . }com.oracle.adbaoverjdbc.OperationGroupfinal CompletionStage T attachCompletionHandler(CompletionStage T result) {return result.handle((r, t) - {Throwable ex unwrapException(t);checkAbort(ex);if (t null)return handleResult(r);elsethrow handleError(ex);});}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 24

submit(Operation n S submit(Operation S op) {memberTail etExecutor()));return Submission.submit(this::cancel, memberTail);}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 25

follows(CompletionStage ? predecessor,Executor etionStage T follows(CompletionStage ? predecessor,Executor executor) {predecessor attachFutureParameters(predecessor);return predecessor.thenRunAsync(this::executeQuery, executor).thenCompose(this::moreRows);}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 26

thenRunAsync(Runnable ionStage Void thenRunAsync(Runnable action)Returns a new CompletionStage that, when this stage completes normally, executes thegiven action using this stage's default asynchronous execution facility.Parameters: action - the action to perform before completing the returnedCompletionStageReturns: the new CompletionStageCopyright 2018, Oracle and/or its affiliates. All rights reserved. 27

follows( . . . )predecessor.thenRunAsync(this::executeQuery, executor)predecessorResult ofprevious opResult ofprevious opthenRunAsyncexecuteQueryCopyright 2018, Oracle and/or its affiliates. All rights reserved. 28

follows(CompletionStage ? predecessor,Executor etionStage T follows(CompletionStage ? predecessor,Executor executor) {predecessor attachFutureParameters(predecessor);return predecessor.thenRunAsync(this::executeQuery, executor).thenCompose(this::moreRows);}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 29

thenCompose(Function ?, CompletionStage ,Executor executor)java.util.concurrent.CompletionStage U CompletionStage U thenCompose(Function ? super T,? extendsCompletionStage U fn)Returns a new CompletionStage that is completed with the same value as theCompletionStage returned by the given function.When this stage completes normally, thegiven function is invoked with this stage's result as the argument, returning anotherCompletionStage. When that stage completes normally, the CompletionStage returned bythis method is completed with the same value.Type Parameters: U - the type of the returned CompletionStage's resultParameters: fn - the function to use to compute another CompletionStageReturns: the new CompletionStageCopyright 2018, Oracle and/or its affiliates. All rights reserved. 30

follows( . . . rythenComposemoreRowsCopyright 2018, Oracle and/or its affiliates. All rights reserved. 31

submit(Operation op, Executor ult ofprevious wshandle(r,t) - {.}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 32

Executing RowOperationPrevious OperationResult ofprevious e(r,t) - {.}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 33

Executing RowOperationexecuteQueryResult ofprevious e(r,t) - {.}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 34

moreRows(Object x)com.oracle.adbaoverjdbc.RowOperationprotected CompletionStage T moreRows(Object x) {checkCanceled();if (rowsRemain) {return CompletableFuture.runAsync(this::handleFetchRows, getExecutor()).thenCompose(this::moreRows);}else {return CompletableFuture.supplyAsync(this::completeQuery, getExecutor());}}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 35

runAsync(Runnable action, Executor c static CompletableFuture Void runAsync(Runnable runnable,Executor executor)Returns a new CompletableFuture that is asynchronously completed by atask running in the given executor after it runs the given action.Parameters: runnable - the action to run before completing the returnedCompletableFutureexecutor - the executor to use for asynchronous executionReturns: the new CompletableFutureCopyright 2018, Oracle and/or its affiliates. All rights reserved. 36

moreRows( . . . wsCopyright 2018, Oracle and/or its affiliates. All rights reserved. 37

Executing RowOperationBefore moreRows(Object x)Result ofprevious e(r,t) - {.}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 38

Executing mpose(this::moreRows)runAsynchandleFetchRowsResult ofprevious reRowsmoreRowshandle(r,t) - {.}Copyright 2018, Oracle and/or its affiliates. All rights reserved. 39