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. 8 * 9 * This code is distributed in the hope that it will be useful, but WITHOUT 10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 12 * version 2 for more details (a copy is included in the LICENSE file that 13 * accompanied this code). 14 * 15 * You should have received a copy of the GNU General Public License version 16 * 2 along with this work; if not, write to the Free Software Foundation, 17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 18 * 19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 20 * or visit www.oracle.com if you need additional information or have any 21 * questions. 22 */ 23 24 import java.nio.ByteBuffer; 25 import java.util.List; 26 import java.util.concurrent.CompletionStage; 27 import java.util.concurrent.CountDownLatch; 28 import java.util.concurrent.ExecutorService; 29 import java.util.concurrent.Executors; 30 import java.util.concurrent.Flow.Subscription; 31 import java.util.concurrent.SubmissionPublisher; 32 import java.util.function.IntSupplier; 33 import java.util.stream.IntStream; 34 import jdk.incubator.http.HttpResponse.BodySubscriber; 35 import org.testng.annotations.DataProvider; 36 import org.testng.annotations.Test; 37 import static java.lang.Long.MAX_VALUE; 38 import static java.lang.Long.MIN_VALUE; 39 import static java.lang.System.out; 40 import static java.nio.ByteBuffer.wrap; 41 import static java.util.concurrent.TimeUnit.SECONDS; 42 import static jdk.incubator.http.HttpResponse.BodySubscriber.buffering; 43 import static org.testng.Assert.*; 44 45 /* 46 * @test 47 * @summary Direct test for HttpResponse.BodySubscriber.buffering() cancellation 48 * @run testng/othervm BufferingSubscriberCancelTest 49 */ 50 51 public class BufferingSubscriberCancelTest { 52 53 @DataProvider(name = "bufferSizes") 54 public Object[][] bufferSizes() { 55 return new Object[][]{ 56 // bufferSize should be irrelevant 57 {1}, {100}, {511}, {512}, {513}, {1024}, {2047}, {2048} 58 }; 59 } 60 61 @Test(dataProvider = "bufferSizes") 62 public void cancelWithoutAnyItemsPublished(int bufferSize) throws Exception { 63 ExecutorService executor = Executors.newFixedThreadPool(1); 64 SubmissionPublisher<List<ByteBuffer>> publisher = 65 new SubmissionPublisher<>(executor, 1); 66 67 CountDownLatch gate = new CountDownLatch(1); // single onSubscribe 68 ExposingSubscriber exposingSubscriber = new ExposingSubscriber(gate); 69 BodySubscriber subscriber = buffering(exposingSubscriber, bufferSize); 70 publisher.subscribe(subscriber); 71 gate.await(30, SECONDS); 72 assertEqualsWithRetry(publisher::getNumberOfSubscribers, 1); 73 exposingSubscriber.subscription.cancel(); 74 assertEqualsWithRetry(publisher::getNumberOfSubscribers, 0); 75 76 // further cancels/requests should be a no-op 77 Subscription s = exposingSubscriber.subscription; 78 s.cancel(); s.request(1); 79 s.cancel(); s.request(100); s.cancel(); 80 s.cancel(); s.request(MAX_VALUE); s.cancel(); s.cancel(); 81 s.cancel(); s.cancel(); s.cancel(); s.cancel(); 82 s.request(MAX_VALUE); s.request(MAX_VALUE); s.request(MAX_VALUE); 83 s.request(-1); s.request(-100); s.request(MIN_VALUE); 84 assertEqualsWithRetry(publisher::getNumberOfSubscribers, 0); 85 executor.shutdown(); 86 } 87 88 @DataProvider(name = "sizeAndItems") 89 public Object[][] sizeAndItems() { 90 return new Object[][] { 91 // bufferSize and item bytes must be equal to count onNext calls 92 // bufferSize items 93 { 1, List.of(wrap(new byte[] { 1 })) }, 94 { 2, List.of(wrap(new byte[] { 1, 2 })) }, 95 { 3, List.of(wrap(new byte[] { 1, 2, 3})) }, 96 { 4, List.of(wrap(new byte[] { 1, 2 , 3, 4})) }, 97 { 5, List.of(wrap(new byte[] { 1, 2 , 3, 4, 5})) }, 98 { 6, List.of(wrap(new byte[] { 1, 2 , 3, 4, 5, 6})) }, 99 { 7, List.of(wrap(new byte[] { 1, 2 , 3, 4, 5, 6, 7})) }, 100 { 8, List.of(wrap(new byte[] { 1, 2 , 3, 4, 5, 6, 7, 8})) }, 101 { 9, List.of(wrap(new byte[] { 1, 2 , 3, 4, 5, 6, 7, 8, 9})) }, 102 { 10, List.of(wrap(new byte[] { 1, 2 , 3, 4, 5, 6, 7, 8, 9, 10})) }, 103 }; 104 } 105 106 @Test(dataProvider = "sizeAndItems") 107 public void cancelWithItemsPublished(int bufferSize, List<ByteBuffer> items) 108 throws Exception 109 { 110 ExecutorService executor = Executors.newFixedThreadPool(1); 111 SubmissionPublisher<List<ByteBuffer>> publisher = 112 new SubmissionPublisher<>(executor, 24); 113 114 final int ITERATION_TIMES = 10; 115 // onSubscribe + onNext ITERATION_TIMES 116 CountDownLatch gate = new CountDownLatch(1 + ITERATION_TIMES); 117 ExposingSubscriber exposingSubscriber = new ExposingSubscriber(gate); 118 BodySubscriber subscriber = buffering(exposingSubscriber, bufferSize); 119 publisher.subscribe(subscriber); 120 121 assertEqualsWithRetry(publisher::getNumberOfSubscribers, 1); 122 IntStream.range(0, ITERATION_TIMES).forEach(x -> publisher.submit(items)); 123 gate.await(30, SECONDS); 124 exposingSubscriber.subscription.cancel(); 125 IntStream.range(0, ITERATION_TIMES+1).forEach(x -> publisher.submit(items)); 126 127 assertEqualsWithRetry(publisher::getNumberOfSubscribers, 0); 128 assertEquals(exposingSubscriber.onNextInvocations, ITERATION_TIMES); 129 executor.shutdown(); 130 } 131 132 // same as above but with more racy conditions, do not wait on the gate 133 @Test(dataProvider = "sizeAndItems") 134 public void cancelWithItemsPublishedNoWait(int bufferSize, List<ByteBuffer> items) 135 throws Exception 136 { 137 ExecutorService executor = Executors.newFixedThreadPool(1); 138 SubmissionPublisher<List<ByteBuffer>> publisher = 139 new SubmissionPublisher<>(executor, 24); 140 141 final int ITERATION_TIMES = 10; 142 // any callback will so, since onSub is guaranteed to be before onNext 143 CountDownLatch gate = new CountDownLatch(1); 144 ExposingSubscriber exposingSubscriber = new ExposingSubscriber(gate); 145 BodySubscriber subscriber = buffering(exposingSubscriber, bufferSize); 146 publisher.subscribe(subscriber); 147 148 IntStream.range(0, ITERATION_TIMES).forEach(x -> publisher.submit(items)); 149 gate.await(30, SECONDS); 150 exposingSubscriber.subscription.cancel(); 151 IntStream.range(0, ITERATION_TIMES+1).forEach(x -> publisher.submit(items)); 152 153 int onNextInvocations = exposingSubscriber.onNextInvocations; 154 assertTrue(onNextInvocations <= ITERATION_TIMES, 155 "Expected <= " + ITERATION_TIMES + ", got " + onNextInvocations); 156 executor.shutdown(); 157 } 158 159 static class ExposingSubscriber implements BodySubscriber<Void> { 160 final CountDownLatch gate; 161 volatile Subscription subscription; 162 volatile int onNextInvocations; 163 164 ExposingSubscriber(CountDownLatch gate) { 165 this.gate = gate; 166 } 167 168 @Override 169 public void onSubscribe(Subscription subscription) { 170 //out.println("onSubscribe " + subscription); 171 this.subscription = subscription; 172 gate.countDown(); 173 subscription.request(MAX_VALUE); // forever 174 } 175 176 @Override 177 public void onNext(List<ByteBuffer> item) { 178 //out.println("onNext " + item); 179 onNextInvocations++; 180 gate.countDown(); 181 } 182 183 @Override 184 public void onError(Throwable throwable) { 185 out.println("onError " + throwable); 186 } 187 188 @Override 189 public void onComplete() { 190 out.println("onComplete "); 191 } 192 193 @Override 194 public CompletionStage<Void> getBody() { 195 throw new UnsupportedOperationException("getBody is unsupported"); 196 } 197 } 198 199 // There is a race between cancellation and subscriber callbacks, the 200 // following mechanism retries a number of times to allow for this race. The 201 // only requirement is that the expected result is actually observed. 202 203 static final int TEST_RECHECK_TIMES = 30; 204 205 static void assertEqualsWithRetry(IntSupplier actualSupplier, int expected) 206 throws Exception 207 { 208 int actual = expected + 1; // anything other than expected 209 for (int i=0; i< TEST_RECHECK_TIMES; i++) { 210 actual = actualSupplier.getAsInt(); 211 if (actual == expected) 212 return; 213 Thread.sleep(100); 214 } 215 assertEquals(actual, expected); // will fail with the usual testng message 216 } 217 }