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 package jdk.incubator.http;
25
26 import jdk.incubator.http.internal.websocket.RawChannel;
27 import org.testng.annotations.Test;
28
29 import java.io.IOException;
30 import java.io.InputStream;
31 import java.io.OutputStream;
32 import java.io.UncheckedIOException;
33 import java.net.ServerSocket;
34 import java.net.Socket;
35 import java.net.URI;
36 import java.nio.ByteBuffer;
37 import java.nio.channels.SelectionKey;
38 import java.util.Random;
39 import java.util.concurrent.CountDownLatch;
40 import java.util.concurrent.TimeUnit;
41 import java.util.concurrent.atomic.AtomicInteger;
42 import java.util.concurrent.atomic.AtomicLong;
43
44 import static jdk.incubator.http.HttpResponse.BodyHandler.discard;
45 import static org.testng.Assert.assertEquals;
46
66
67 /*
68 * This one works similarly by providing means to ensure a read from the
69 * channel will stall at least once (no more data available on the socket).
70 */
71 private final CountDownLatch readStall = new CountDownLatch(1);
72 private final CountDownLatch initialReadStall = new CountDownLatch(1);
73
74 private final AtomicInteger writeHandles = new AtomicInteger();
75 private final AtomicInteger readHandles = new AtomicInteger();
76
77 private final CountDownLatch exit = new CountDownLatch(1);
78
79 @Test
80 public void test() throws Exception {
81 try (ServerSocket server = new ServerSocket(0)) {
82 int port = server.getLocalPort();
83 new TestServer(server).start();
84
85 final RawChannel chan = channelOf(port);
86 initialWriteStall.await();
87
88 // It's very important not to forget the initial bytes, possibly
89 // left from the HTTP thingy
90 int initialBytes = chan.initialByteBuffer().remaining();
91 print("RawChannel has %s initial bytes", initialBytes);
92 clientRead.addAndGet(initialBytes);
93
94 // tell the server we have read the initial bytes, so
95 // that it makes sure there is something for us to
96 // read next in case the initialBytes have already drained the
97 // channel dry.
98 initialReadStall.countDown();
99
100 chan.registerEvent(new RawChannel.RawEvent() {
101
102 private final ByteBuffer reusableBuffer = ByteBuffer.allocate(32768);
103
104 @Override
105 public int interestOps() {
168 readStall.countDown();
169 break;
170 }
171 int r = read.remaining();
172 total += r;
173 clientRead.addAndGet(r);
174 }
175 print("OP_READ read %s bytes (%s total)", total, clientRead.get());
176 }
177 });
178 exit.await(); // All done, we need to compare results:
179 assertEquals(clientRead.get(), serverWritten.get());
180 assertEquals(serverRead.get(), clientWritten.get());
181 }
182 }
183
184 private static RawChannel channelOf(int port) throws Exception {
185 URI uri = URI.create("http://127.0.0.1:" + port + "/");
186 print("raw channel to %s", uri.toString());
187 HttpRequest req = HttpRequest.newBuilder(uri).build();
188 HttpResponse<?> r = HttpClient.newHttpClient().send(req, discard(null));
189 r.body();
190 return ((HttpResponseImpl) r).rawChannel();
191 }
192
193 private class TestServer extends Thread { // Powered by Slowpokes
194
195 private final ServerSocket server;
196
197 TestServer(ServerSocket server) throws IOException {
198 this.server = server;
199 }
200
201 @Override
202 public void run() {
203 try (Socket s = server.accept()) {
204 InputStream is = s.getInputStream();
205 OutputStream os = s.getOutputStream();
206
207 processHttp(is, os);
208
209 Thread reader = new Thread(() -> {
210 try {
|
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 package jdk.incubator.http;
25
26 import jdk.incubator.http.internal.websocket.RawChannel;
27 import jdk.incubator.http.internal.websocket.WebSocketRequest;
28 import org.testng.annotations.Test;
29
30 import java.io.IOException;
31 import java.io.InputStream;
32 import java.io.OutputStream;
33 import java.io.UncheckedIOException;
34 import java.net.ServerSocket;
35 import java.net.Socket;
36 import java.net.URI;
37 import java.nio.ByteBuffer;
38 import java.nio.channels.SelectionKey;
39 import java.util.Random;
40 import java.util.concurrent.CountDownLatch;
41 import java.util.concurrent.TimeUnit;
42 import java.util.concurrent.atomic.AtomicInteger;
43 import java.util.concurrent.atomic.AtomicLong;
44
45 import static jdk.incubator.http.HttpResponse.BodyHandler.discard;
46 import static org.testng.Assert.assertEquals;
47
67
68 /*
69 * This one works similarly by providing means to ensure a read from the
70 * channel will stall at least once (no more data available on the socket).
71 */
72 private final CountDownLatch readStall = new CountDownLatch(1);
73 private final CountDownLatch initialReadStall = new CountDownLatch(1);
74
75 private final AtomicInteger writeHandles = new AtomicInteger();
76 private final AtomicInteger readHandles = new AtomicInteger();
77
78 private final CountDownLatch exit = new CountDownLatch(1);
79
80 @Test
81 public void test() throws Exception {
82 try (ServerSocket server = new ServerSocket(0)) {
83 int port = server.getLocalPort();
84 new TestServer(server).start();
85
86 final RawChannel chan = channelOf(port);
87 print("RawChannel is %s", String.valueOf(chan));
88 initialWriteStall.await();
89
90 // It's very important not to forget the initial bytes, possibly
91 // left from the HTTP thingy
92 int initialBytes = chan.initialByteBuffer().remaining();
93 print("RawChannel has %s initial bytes", initialBytes);
94 clientRead.addAndGet(initialBytes);
95
96 // tell the server we have read the initial bytes, so
97 // that it makes sure there is something for us to
98 // read next in case the initialBytes have already drained the
99 // channel dry.
100 initialReadStall.countDown();
101
102 chan.registerEvent(new RawChannel.RawEvent() {
103
104 private final ByteBuffer reusableBuffer = ByteBuffer.allocate(32768);
105
106 @Override
107 public int interestOps() {
170 readStall.countDown();
171 break;
172 }
173 int r = read.remaining();
174 total += r;
175 clientRead.addAndGet(r);
176 }
177 print("OP_READ read %s bytes (%s total)", total, clientRead.get());
178 }
179 });
180 exit.await(); // All done, we need to compare results:
181 assertEquals(clientRead.get(), serverWritten.get());
182 assertEquals(serverRead.get(), clientWritten.get());
183 }
184 }
185
186 private static RawChannel channelOf(int port) throws Exception {
187 URI uri = URI.create("http://127.0.0.1:" + port + "/");
188 print("raw channel to %s", uri.toString());
189 HttpRequest req = HttpRequest.newBuilder(uri).build();
190 // Switch on isWebSocket flag to prevent the connection from
191 // being returned to the pool.
192 ((WebSocketRequest)req).isWebSocket(true);
193 HttpClient client = HttpClient.newHttpClient();
194 try {
195 HttpResponse<?> r = client.send(req, discard(null));
196 r.body();
197 return ((HttpResponseImpl) r).rawChannel();
198 } finally {
199 // Need to hold onto the client until the RawChannel is
200 // created. This would not be needed if we had created
201 // a WebSocket, but here we are fiddling directly
202 // with the internals of HttpResponseImpl!
203 java.lang.ref.Reference.reachabilityFence(client);
204 }
205 }
206
207 private class TestServer extends Thread { // Powered by Slowpokes
208
209 private final ServerSocket server;
210
211 TestServer(ServerSocket server) throws IOException {
212 this.server = server;
213 }
214
215 @Override
216 public void run() {
217 try (Socket s = server.accept()) {
218 InputStream is = s.getInputStream();
219 OutputStream os = s.getOutputStream();
220
221 processHttp(is, os);
222
223 Thread reader = new Thread(() -> {
224 try {
|