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 

    26 package jdk.incubator.http.internal.common;

    27 

    28 import jdk.incubator.http.internal.common.SequentialScheduler.CompleteRestartableTask;

    29 

    30 import java.util.Objects;

    31 import java.util.concurrent.Flow.Subscription;

    32 import java.util.concurrent.atomic.AtomicReference;

    33 

    34 /**

    35  * Acts as a subscription receiving calls to {@code request} and {@code cancel}

    36  * methods until the replacing subscription is set.

    37  *

    38  * <p> After the replacing subscription is set, it gets updated with the result

    39  * of calls happened before that and starts receiving calls to its

    40  * {@code request} and {@code cancel} methods.

    41  *

    42  * <p> This subscription ensures that {@code request} and {@code cancel} methods

    43  * of the replacing subscription are called sequentially.

    44  */

    45 public final class TemporarySubscription implements Subscription {

    46 

    47     private final AtomicReference<Subscription> subscription = new AtomicReference<>();

    48     private final Demand demand = new Demand();

    49     private volatile boolean cancelled;

    50     private volatile long illegalValue = 1;

    51 

    52     private final SequentialScheduler scheduler = new SequentialScheduler(new UpdateTask());

    53 

    54     @Override

    55     public void request(long n) {

    56         if (n <= 0) {

    57             // Any non-positive request would do, no need to remember them

    58             // all or any one in particular.

    59             // tl;dr racy, but don't care

    60             illegalValue = n;

    61         } else {

    62             demand.increase(n);

    63         }

    64         scheduler.runOrSchedule();

    65     }

    66 

    67     @Override

    68     public void cancel() {

    69         cancelled = true;

    70         scheduler.runOrSchedule();

    71     }

    72 

    73     public void replaceWith(Subscription permanentSubscription) {

    74         Objects.requireNonNull(permanentSubscription);

    75         if (permanentSubscription == this) {

    76             // Otherwise it would be an unpleasant bug to chase

    77             throw new IllegalStateException("Self replacement");

    78         }

    79         if (!subscription.compareAndSet(null, permanentSubscription)) {

    80             throw new IllegalStateException("Already replaced");

    81         }

    82         scheduler.runOrSchedule();

    83     }

    84 

    85     private final class UpdateTask extends CompleteRestartableTask {

    86 

    87         @Override

    88         public void run() {

    89             Subscription dst = TemporarySubscription.this.subscription.get();

    90             if (dst == null) {

    91                 return;

    92             }

    93             /* As long as the result is effectively the same, it does not matter

    94                how requests are accumulated and what goes first: request or

    95                cancel. See rules 3.5, 3.6, 3.7 and 3.9 from the reactive-streams

    96                specification. */

    97             long illegalValue = TemporarySubscription.this.illegalValue;

    98             if (illegalValue <= 0) {

    99                 dst.request(illegalValue);

   100                 scheduler.stop();

   101             } else if (cancelled) {

   102                 dst.cancel();

   103                 scheduler.stop();

   104             } else {

   105                 long accumulatedValue = demand.decreaseAndGet(Long.MAX_VALUE);

   106                 if (accumulatedValue > 0) {

   107                     dst.request(accumulatedValue);

   108                 }

   109             }

   110         }

   111     }

   112 }