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 java.net.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 java.net.http.HttpResponse.BodySubscribers.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")
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")
cancelWithoutAnyItemsPublished(int bufferSize)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")
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")
cancelWithItemsPublished(int bufferSize, List<ByteBuffer> items)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")
cancelWithItemsPublishedNoWait(int bufferSize, List<ByteBuffer> items)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 
ExposingSubscriber(CountDownLatch gate)164         ExposingSubscriber(CountDownLatch gate) {
165             this.gate = gate;
166         }
167 
168         @Override
onSubscribe(Subscription subscription)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
onNext(List<ByteBuffer> item)177         public void onNext(List<ByteBuffer> item) {
178             //out.println("onNext " + item);
179             onNextInvocations++;
180             gate.countDown();
181         }
182 
183         @Override
onError(Throwable throwable)184         public void onError(Throwable throwable) {
185             out.println("onError " + throwable);
186         }
187 
188         @Override
onComplete()189         public void onComplete() {
190             out.println("onComplete ");
191         }
192 
193         @Override
getBody()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 
assertEqualsWithRetry(IntSupplier actualSupplier, int expected)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 }
218