1 /*

     2  * Copyright (c)  2017, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 package jdk.incubator.sql2;

    26 

    27 import java.time.Duration;

    28 import java.util.concurrent.CompletionStage;

    29 import java.util.concurrent.Flow;

    30 import java.util.function.Consumer;

    31 import java.util.logging.Logger;

    32 import java.util.stream.Collector;

    33 

    34 /**

    35  * A set of {@link Operation}s that share certain properties, are managed as a

    36  * unit, and are executed as a unit. The {@link Operation}s created by an

    37  * {@link OperationGroup} and submitted are the member {@link Operation}s of

    38  * that {@link OperationGroup}. An {@link OperationGroup} is not a transaction

    39  * and is not related to a transaction in any way.

    40  *

    41  * An {@link OperationGroup} conceptually has a collection of member

    42  * {@link Operation}s. When an {@link OperationGroup} is submitted it is placed

    43  * in the collection of the {@link OperationGroup} of which it is a member. The

    44  * member {@link OperationGroup} is executed according to the attributes of the

    45  * {@link OperationGroup} of which it is a member. The member {@link Operation}s

    46  * of an {@link OperationGroup} are executed according to the attributes of that

    47  * {@link OperationGroup}.

    48  *

    49  * How an {@link OperationGroup} is executed depends on its attributes.

    50  *

    51  * If an {@link OperationGroup} has a condition and the value of that condition

    52  * is {@link Boolean#TRUE} then execute the member {@link Operation}s as below.

    53  * If it is {@link Boolean#FALSE} then the {@link OperationGroup} is completed

    54  * with the value null. If the condition completed exceptionally then the

    55  * {@link OperationGroup} is completed exceptionally with a

    56  * {@link SqlSkippedException} that has that exception as its cause.

    57  *

    58  * If the {@link OperationGroup} is sequential the member {@link Operation}s are

    59  * executed in the order they were submitted. If it is parallel, they may be

    60  * executed in any order including simultaneously.

    61  *

    62  * If an {@link OperationGroup} is dependent and a member {@link Operation}

    63  * completes exceptionally the remaining member {@link Operation}s in the

    64  * collection are completed exceptionally with a {@link SqlSkippedException}

    65  * that has the initial {@link Exception} as its cause and the {@link OperationGroup}

    66  * is completed exceptionally with the initial {@link Exception}. A member

    67  * {@link Operation} in-flight may either complete normally or be completed

    68  * exceptionally but must complete one way or the other. [NOTE: Too strong?]

    69  *

    70  * If an {@link OperationGroup} is held additional member {@link Operation}s may

    71  * be submitted after the {@link OperationGroup} is submitted. If an

    72  * {@link OperationGroup} is not held, no additional member {@link Operation}s

    73  * may be submitted after the {@link OperationGroup} is submitted. If an

    74  * {@link OperationGroup} is held it will be completed only after it is released

    75  * or if conditional and the condition is not {@link Boolean#TRUE}. If a

    76  * {@link OperationGroup} is dependent, held, one of its member

    77  * {@link Operation}s completed exceptionally, and its queue is empty then the

    78  * {@link OperationGroup} is released.

    79  *

    80  * The result of this {@link OperationGroup} is the result of collecting the

    81  * results of its member {@link Operation}s. If the {@link OperationGroup} is 

    82  * dependent and one of its member {@link Operation}s completes exceptionally,

    83  * the {@link OperationGroup} is completed exceptionally.

    84  *

    85  * ISSUE: Currently no way to create a nested {@link OperationGroup}. That is a

    86  * intentional limitation but may be a simplification we can live with. Or not.

    87  *

    88  * @param <S> The type of the result of the member {@link Operation}s

    89  * @param <T> The type of the collected results the member {@link Operation}s

    90  */

    91 public interface OperationGroup<S, T> extends Operation<T> {

    92 

    93   /**

    94    * Mark this {@link OperationGroup} as parallel. If this method is not called

    95    * the {@link OperationGroup} is sequential. If an {@link OperationGroup} is

    96    * parallel, member {@link Operation}s may be executed in any order including

    97    * in parallel. If an {@link OperationGroup} is sequential, the default,

    98    * member {@link Operation}s are executed strictly in the order they are

    99    * submitted.

   100    *

   101    * Note: There is no covariant override of this method in {@link Connection}

   102    * as there is only a small likelihood of needing it.

   103    *

   104    * @return this {@link OperationGroup}

   105    * @throws IllegalStateException if this {@link OperationGroup} has been

   106    * submitted or any member {@link Operation}s have been created.

   107    */

   108   public OperationGroup<S, T> parallel();

   109 

   110   /**

   111    * Mark this {@link OperationGroup} as independent. If this method is not

   112    * called the {@link OperationGroup} is dependent, the default. If an

   113    * {@link OperationGroup} is independent then failure of one member

   114    * {@link Operation} does not affect the execution of other member

   115    * {@link Operation}s. If an {@link OperationGroup} is dependent then failure

   116    * of one member {@link Operation} will cause all member {@link Operation}s

   117    * remaining in the queue to be completed exceptionally with a

   118    * {@link SqlSkippedException} with the cause set to the original exception.

   119    *

   120    * The result of this {@link OperationGroup}'s execution is the result of collecting the

   121    * results of the member {@link Operation}s that complete normally. 

   122    *

   123    * Note: There is no covariant override of this method in {@link Connection}

   124    * as there is only a small likelihood of needing it.

   125    *

   126    * @return this {@link OperationGroup}

   127    * @throws IllegalStateException f this {@link OperationGroup} blocks errors

   128    * or if this {@link OperationGroup} has been submitted or any member

   129    * {@link Operation}s have been created

   130    */

   131   public OperationGroup<S, T> independent();

   132 

   133   /**

   134    * Define a condition that determines whether the member {@link Operation}s of

   135    * this {@link OperationGroup} are executed or not. If and when this

   136    * {@link OperationGroup} is executed then if the condition argument is

   137    * completed with {@link Boolean#TRUE} the member {@link Operation}s are

   138    * executed. If {@link Boolean#FALSE} or if it is completed exceptionally the

   139    * member {@link Operation}s are not executed but are removed from the queue.

   140    * After all member {@link Operation}s have been removed from the queue this

   141    * {@link OperationGroup} is completed with {@code null}.

   142    *

   143    * Note: There is no covariant override of this method in Connection as there

   144    * is only a small likelihood of needing it.

   145    *

   146    * ISSUE: Should the member Operations be skipped or otherwise completed

   147    * exceptionally?

   148    *

   149    * @param condition a {@link CompletionStage} the value of which determines

   150    * whether this {@link OperationGroup} is executed or not

   151    * @return this OperationGroup

   152    * @throws IllegalStateException iif this {@link OperationGroup} has been

   153    * submitted or any member {@link Operation}s have been created

   154    */

   155   public OperationGroup<S, T> conditional(CompletionStage<Boolean> condition);

   156 

   157   /**

   158    * Mark this {@link OperationGroup} as held. It can be executed but cannot be

   159    * completed. A {@link OperationGroup} that is held remains in the queue even

   160    * if all of its current member {@link Operation}s have completed. So long as

   161    * the {@link OperationGroup} is held new member {@link Operation}s can be

   162    * submitted. A {@link OperationGroup} that is held must be released before it

   163    * can be completed and removed from the queue.

   164    *

   165    * Note: There is no covariant override of this method in Connection as there

   166    * is only a small likelihood of needing it.

   167    *

   168    * ISSUE: Need a better name.

   169    *

   170    * @return a Submission

   171    * @throws IllegalStateException if this {@link OperationGroup} has been

   172    * submitted

   173    */

   174   public Submission<T> submitHoldingForMoreMembers();

   175 

   176   /**

   177    * Allow this {@link OperationGroup} to be completed and removed from the

   178    * queue once all of its member {@link Operation}s have been completed. After

   179    * this method is called no additional member {@link Operation}s can be

   180    * submitted. Once all member {@link Operation}s have been removed from the

   181    * queue this {@link OperationGroup} will be completed and removed from the

   182    * queue.

   183    *

   184    * Calling this method when this {@link OperationGroup} is not held is a

   185    * no-op.

   186    *

   187    * Note: There is no covariant override of this method in Connection as there

   188    * is only a small likelihood of needing it.

   189    *

   190    * ISSUE: Need a better name.

   191    *

   192    * @return this OperationGroup

   193    * @throws IllegalStateException if this {@link OperationGroup} has been

   194    * completed

   195    */

   196   public OperationGroup<S, T> releaseProhibitingMoreMembers();

   197 

   198   /**

   199    * Provides a {@link Collector} to reduce the results of the member

   200    * {@link Operation}s. The result of this {@link OperationGroup} is the result

   201    * of calling finisher on the final accumulated result.If the

   202    * {@link Collector} is {@link Collector.Characteristics#UNORDERED} the member

   203    * {@link Operation} results may be accumulated out of order.If the

   204    * {@link Collector} is {@link Collector.Characteristics#CONCURRENT} then the

   205    * member {@link Operation} results may be split into subsets that are reduced

   206    * separately and then combined. If this {@link OperationGroup} is sequential,

   207    * the characteristics of the {@link Collector} only affect how the results of

   208    * the member {@link Operation}s are collected; the member {@link Operation}s

   209    * are executed sequentially regardless. If this {@link OperationGroup} is

   210    * parallel the characteristics of the {@link Collector} may influence the

   211    * execution order of the member {@link Operation}s.

   212    *

   213    * The default value is

   214    * {@code Collector.of(()->null, (a,t)->{}, (l,r)->null, a->null)}.

   215    *

   216    * @param c the Collector. Not null.

   217    * @return This OperationGroup

   218    * @throws IllegalStateException if called more than once or if this

   219    * {@link OperationGroup} has been submitted

   220    */

   221   public OperationGroup<S, T> collect(Collector<S, ?, T> c);

   222   

   223   /**

   224    * Returns an Operation that is never skipped. Skipping stops with a catchOperation

   225    * and the subsequent Operation is executed normally. The value of a 

   226    * catchOperation is always null.

   227    * 

   228    * @return an unskippable Operation;

   229    */

   230   public Operation<S> catchOperation();

   231   

   232   /**

   233    * Creates and submits a catch Operation. Convenience method.

   234    *

   235    * @return this OperationGroup

   236    */

   237   public default OperationGroup<S, T> catchErrors() {

   238     catchOperation().submit();

   239     return this;

   240   }

   241 

   242   /**

   243    * Return a new {@link ArrayCountOperation}.

   244    * <p>

   245    * Usage Note: Frequently use of this method will require a type witness to

   246    * enable correct type inferencing.

   247    * <pre><code>

   248    *   conn.<b>&lt;List&lt;Integer&gt;&gt;</b>arrayCountOperation(sql)

   249    *     .set ...

   250    *     .collect ...

   251    *     .submit ...

   252    * </code></pre>

   253    *

   254    * @param <R> the result type of the returned {@link ArrayCountOperation}

   255    * @param sql SQL to be executed. Must return an update count.

   256    * @return a new {@link ArrayCountOperation} that is a member of this

   257    * {@link OperationGroup}

   258    */

   259   public <R extends S> ArrayCountOperation<R> arrayCountOperation(String sql);

   260 

   261   /**

   262    * Return a new {@link CountOperation}.

   263    *

   264    * @param <R> the result type of the returned {@link CountOperation}

   265    * @param sql SQL to be executed. Must return an update count.

   266    * @return an new {@link CountOperation} that is a member of this

   267    * {@link OperationGroup}

   268    *

   269    */

   270   public <R extends S> ParameterizedCountOperation<R> countOperation(String sql);

   271 

   272   /**

   273    * Return a new {@link Operation} for a SQL that doesn't return any result,

   274    * for example DDL. The result of this Operation is always null.

   275    *

   276    * @param sql SQL for the {@link Operation}.

   277    * @return a new {@link Operation} that is a member of this

   278    * {@link OperationGroup}

   279    */

   280   public Operation<S> operation(String sql);

   281 

   282   /**

   283    * Return a new {@link OutOperation}. The SQL must return a set of zero or

   284    * more out parameters or function results.

   285    *

   286    * @param <R> the result type of the returned {@link OutOperation}

   287    * @param sql SQL for the {@link Operation}. Must return zero or more out

   288    * parameters or function results.

   289    * @return a new {@link OutOperation} that is a member of this

   290    * {@link OperationGroup}

   291    */

   292   public <R extends S> OutOperation<R> outOperation(String sql);

   293 

   294   /**

   295    * Return a {@link ParameterizedRowOperation}.

   296    *

   297    * @param <R> the type of the result of the returned

   298    * {@link ParameterizedRowOperation}

   299    * @param sql SQL for the {@link Operation}. Must return a row sequence.

   300    * @return a new {@link ParameterizedRowOperation} that is a member of this

   301    * {@link OperationGroup}

   302    */

   303   public <R extends S> ParameterizedRowOperation<R> rowOperation(String sql);

   304 

   305   public <R extends S> RowProcessorOperation<R> rowProcessorOperation(String sql);

   306 

   307   /**

   308    * Return a {@link StaticMultiOperation}.

   309    *

   310    * @param <R> the type of the result of the returned

   311    * {@link StaticMultiOperation}

   312    * @param sql SQL for the {@link Operation}

   313    * @return a new {@link StaticMultiOperation} that is a member of this

   314    * {@link OperationGroup}

   315    */

   316   public <R extends S> StaticMultiOperation<R> staticMultiOperation(String sql);

   317 

   318   /**

   319    * Return a {@link DynamicMultiOperation}. Use this when the number and type

   320    * of the results is not knowable.

   321    *

   322    * @param <R> the type of the result of the returned

   323    * {@link DynamicMultiOperation}

   324    * @param sql SQL for the {@link Operation}

   325    * @return a new {@link DynamicMultiOperation} that is a member of this

   326    * {@link OperationGroup}

   327    */

   328   public <R extends S> DynamicMultiOperation<R> dynamicMultiOperation(String sql);

   329 

   330   /**

   331    * Return an {@link Operation} that ends the database transaction. The

   332    * transaction is ended with a commit unless the {@link Transaction} has been

   333    * {@link Transaction#setRollbackOnly} in which case the transaction is ended

   334    * with a rollback.

   335    *

   336    * The type argument {@link S} of the containing {@link OperationGroup} must

   337    * be a supertype of {@link TransactionOutcome}.

   338    *

   339    * @param trans the Transaction that determines whether the Operation does a

   340    * database commit or a database rollback.

   341    * @return an {@link Operation} that will end the database transaction.

   342    * @throws IllegalStateException if this {@link OperationGroup} has been

   343    * submitted and is not held or is parallel.

   344    */

   345   public Operation<TransactionOutcome> endTransactionOperation(Transaction trans);

   346 

   347   /**

   348    * Convenience method that creates and submits a endTransaction

   349    * {@link Operation} that commits by default but can be set to rollback by

   350    * calling {@link Transaction#setRollbackOnly}.

   351    *

   352    * @param trans the Transaction that determines whether the Operation is a

   353    * database commit or a database rollback.

   354    * @return this {@link OperationGroup}

   355    * @throws IllegalStateException if this {@link OperationGroup} has been

   356    * submitted and is not held or is parallel.

   357    */

   358   public default CompletionStage<TransactionOutcome> commitMaybeRollback(Transaction trans) {

   359     return this.endTransactionOperation(trans).submit().getCompletionStage();

   360   }

   361 

   362   /**

   363    * Return a {@link LocalOperation}.

   364    *

   365    * @param <R> value type of the returned local Operation

   366    * @return a LocalOperation

   367    * @throws IllegalStateException if this OperationGroup has been submitted and

   368    * is not held

   369    */

   370   public <R extends S> LocalOperation<R> localOperation();

   371 

   372   /**

   373    * Returns a Flow.Processor that subscribes to a sequence of Operations and

   374    * produces a sequence of corresponding Submissions.The Operations must be

   375    * members of this OperationGroup. Calling Subscription.onNext with any

   376    * Operation that is not a member of this OperationGroup, that is was not

   377    * created by calling one of the Operation factory methods on this

   378    * OperationGroup, will cause the Subscription to be canceled and call

   379    * Subscriber.onError with IllegalArgumentException. The method

   380    * Subscription.onNext will call submit on each Operation it is passed and

   381    * publish the resulting Submission. Since an Operation can only be submitted

   382    * once, submitting an Operation and calling onNext with that submitted

   383    * Operation will cause the Subscription to be canceled and Subscriber.onError

   384    * to be called with IllegalStateException. The Processor does not retain

   385    * Submissions to produce to a subsequently attached Subscriber.

   386    *

   387    * If there is no Subscriber to the Processor, the Processor will request

   388    * Operations as appropriate. If there is a Subscriber to the Processor, the

   389    * Processor will request Operations no faster than the Subscriber requests

   390    * Submissions.

   391    *

   392    * Each call to this method returns a new Flow.processor. The Submissions

   393    * published to each Processor are exactly those generated by calling submit

   394    * on the Operations passed as arguments to onNext on the same Processor.

   395    * Calling this method while there is an active Processor will throw

   396    * IllegalStateException.

   397    *

   398    * Note: If any Operation is submitted directly, that is by calling submit

   399    * rather than passing it to onNext, the Submission returned by the submit

   400    * call will not be published.

   401    *

   402    * @param <R>

   403    * @return a Flow.Processor that accepts Operations and generates Submissions

   404    * @throws IllegalStateException if there is an active Processor

   405    */

   406   public <R extends S> Flow.Processor<Operation<R>, Submission<R>> operationProcessor();

   407 

   408   /**

   409    * Supply a {@link Logger} for the implementation of this

   410    * {@link OperationGroup} to use to log significant events. Exactly what

   411    * events are logged, at what Level the events are logged and with what

   412    * parameters is implementation dependent. All member {@link Operation}s of

   413    * this {@link OperationGroup} will use the same {@link Logger} except a

   414    * member {@link OperationGroup} that is supplied with a different

   415    * {@link Logger} uses that {@link Logger}.

   416    *

   417    * Supplying a {@link Logger} configured with a

   418    * {@link java.util.logging.MemoryHandler} with the

   419    * {@link java.util.logging.MemoryHandler#pushLevel} set to

   420    * {@link java.util.logging.Level#WARNING} will result in no log output in

   421    * normal operation. In the event of an error the actions leading up to the

   422    * error will be logged.

   423    *

   424    * Implementation Note: Implementations are encouraged to log the creation of

   425    * this {@link OperationGroup} set to {@link java.util.logging.Level#INFO},

   426    * the creation of member {@link Operation}s at the

   427    * {@link java.util.logging.Level#CONFIG} level, and execution of member

   428    * {@link Operation}s at the {@link java.util.logging.Level#FINE} level.

   429    * Detailed information about the execution of member {@link Operation}s may

   430    * be logged at the {@link java.util.logging.Level#FINER} and

   431    * {@link java.util.logging.Level#FINEST} levels. Errors in the execution of

   432    * user code should be logged at the {@link java.util.logging.Level#WARNING}

   433    * Level. Errors in the implementation code should be logged at the

   434    * {@link java.util.logging.Level#SEVERE} Level.

   435    *

   436    * @param logger used by the implementation to log significant events

   437    * @return this {@link OperationGroup}

   438    */

   439   public OperationGroup<S, T> logger(Logger logger);

   440 

   441   // Covariant overrides

   442   @Override

   443   public OperationGroup<S, T> timeout(Duration minTime);

   444 

   445   @Override

   446   public OperationGroup<S, T> onError(Consumer<Throwable> handler);

   447 }