# HG changeset patch
# User dl
# Date 1512775297 28800
#      Fri Dec 08 15:21:37 2017 -0800
# Node ID be91c7f99ba00c4ff9e9a2927a0554a436bd5710
# Parent  d0e8542ef650aca2bded22522d9b5205fbe9cb28
8193174: SubmissionPublisher invokes the Subscriber's onComplete before all of its submitted items have been published
Reviewed-by: martin, psandoz, chegar

diff --git a/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java b/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java
--- a/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java
+++ b/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java
@@ -1252,18 +1252,20 @@
                         head = h += taken;
                         d = subtractDemand(taken);
                     }
-                    else if ((empty = (t == h)) && (c & COMPLETE) != 0) {
-                        closeOnComplete(s);          // end of stream
-                        break;
-                    }
                     else if ((d = demand) == 0L && (c & REQS) != 0)
                         weakCasCtl(c, c & ~REQS);    // exhausted demand
                     else if (d != 0L && (c & REQS) == 0)
                         weakCasCtl(c, c | REQS);     // new demand
-                    else if (t == (t = tail) && (empty || d == 0L)) {
-                        int bit = ((c & ACTIVE) != 0) ? ACTIVE : RUN;
-                        if (weakCasCtl(c, c & ~bit) && bit == RUN)
-                            break;                   // un-keep-alive or exit
+                    else if (t == (t = tail)) {      // stability check
+                        if ((empty = (t == h)) && (c & COMPLETE) != 0) {
+                            closeOnComplete(s);      // end of stream
+                            break;
+                        }
+                        else if (empty || d == 0L) {
+                            int bit = ((c & ACTIVE) != 0) ? ACTIVE : RUN;
+                            if (weakCasCtl(c, c & ~bit) && bit == RUN)
+                                break;               // un-keep-alive or exit
+                        }
                     }
                 }
             }