OLD | NEW |
1 // Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file |
2 // for details. All rights reserved. Use of this source code is governed by a | 2 // for details. All rights reserved. Use of this source code is governed by a |
3 // BSD-style license that can be found in the LICENSE file. | 3 // BSD-style license that can be found in the LICENSE file. |
4 | 4 |
5 library isolate.example.http_server; | 5 library isolate.example.http_server; |
6 | 6 |
7 import "dart:async"; | 7 import "dart:async"; |
8 import "dart:io"; | 8 import "dart:io"; |
9 import "dart:isolate"; | 9 import "dart:isolate"; |
10 | 10 |
11 import 'package:isolate/isolate_runner.dart'; | 11 import 'package:isolate/isolate_runner.dart'; |
12 import "package:isolate/ports.dart"; | 12 import "package:isolate/ports.dart"; |
13 import "package:isolate/runner.dart"; | 13 import "package:isolate/runner.dart"; |
14 | 14 |
15 typedef Future RemoteStop(); | 15 typedef Future RemoteStop(); |
16 | 16 |
17 Future<RemoteStop> runHttpServer( | 17 Future<RemoteStop> runHttpServer( |
18 Runner runner, ServerSocket socket, HttpListener listener) { | 18 Runner runner, int port, HttpListener listener) async { |
19 return runner.run(_startHttpServer, new List(2)..[0] = socket.reference | 19 var stopPort = await runner.run(_startHttpServer, [port, listener]); |
20 ..[1] = listener) | 20 |
21 .then((SendPort stopPort) => () => _sendStop(stopPort)); | 21 return () => _sendStop(stopPort); |
22 } | 22 } |
23 | 23 |
24 Future _sendStop(SendPort stopPort) { | 24 Future _sendStop(SendPort stopPort) => singleResponseFuture(stopPort.send); |
25 return singleResponseFuture(stopPort.send); | |
26 } | |
27 | 25 |
28 Future<SendPort> _startHttpServer(List args) { | 26 Future<SendPort> _startHttpServer(List args) async { |
29 ServerSocketReference ref = args[0]; | 27 int port = args[0]; |
30 HttpListener listener = args[1]; | 28 HttpListener listener = args[1]; |
31 return ref.create().then((socket) { | 29 |
32 return listener.start(new HttpServer.listenOn(socket)); | 30 var server = |
33 }).then((_) { | 31 await HttpServer.bind(InternetAddress.ANY_IP_V6, port, shared: true); |
34 return singleCallbackPort((SendPort resultPort) { | 32 await listener.start(server); |
35 sendFutureResult(new Future.sync(listener.stop), resultPort); | 33 |
36 }); | 34 return singleCallbackPort((SendPort resultPort) { |
| 35 sendFutureResult(new Future.sync(listener.stop), resultPort); |
37 }); | 36 }); |
38 } | 37 } |
39 | 38 |
40 /// An [HttpRequest] handler setup. Gets called when with the server, and | 39 /// An [HttpRequest] handler setup. Gets called when with the server, and |
41 /// is told when to stop listening. | 40 /// is told when to stop listening. |
42 /// | 41 /// |
43 /// These callbacks allow the listener to set up handlers for HTTP requests. | 42 /// These callbacks allow the listener to set up handlers for HTTP requests. |
44 /// The object should be sendable to an equivalent isolate. | 43 /// The object should be sendable to an equivalent isolate. |
45 abstract class HttpListener { | 44 abstract class HttpListener { |
46 Future start(HttpServer server); | 45 Future start(HttpServer server); |
47 Future stop(); | 46 Future stop(); |
48 } | 47 } |
49 | 48 |
50 /// An [HttpListener] that sets itself up as an echo server. | 49 /// An [HttpListener] that sets itself up as an echo server. |
51 /// | 50 /// |
52 /// Returns the message content plus an ID describing the isolate that | 51 /// Returns the message content plus an ID describing the isolate that |
53 /// handled the request. | 52 /// handled the request. |
54 class EchoHttpListener implements HttpListener { | 53 class EchoHttpListener implements HttpListener { |
| 54 static const _delay = const Duration(seconds: 2); |
| 55 static final _id = Isolate.current.hashCode; |
| 56 final SendPort _counter; |
| 57 |
55 StreamSubscription _subscription; | 58 StreamSubscription _subscription; |
56 static int _id = new Object().hashCode; | |
57 SendPort _counter; | |
58 | 59 |
59 EchoHttpListener(this._counter); | 60 EchoHttpListener(this._counter); |
60 | 61 |
61 start(HttpServer server) { | 62 Future start(HttpServer server) async { |
62 print("Starting isolate $_id"); | 63 print("Starting isolate $_id"); |
63 _subscription = server.listen((HttpRequest request) { | 64 _subscription = server.listen((HttpRequest request) async { |
64 request.response.addStream(request).then((_) { | 65 await request.response.addStream(request); |
65 _counter.send(null); | 66 print("Request to $hashCode"); |
66 print("Request to $_id"); | 67 request.response.write("#$_id\n"); |
67 request.response.write("#$_id\n"); | 68 var watch = new Stopwatch()..start(); |
68 var t0 = new DateTime.now().add(new Duration(seconds: 2)); | 69 while (watch.elapsed < _delay); |
69 while (new DateTime.now().isBefore(t0)); | 70 print("Response from $_id"); |
70 print("Response from $_id"); | 71 await request.response.close(); |
71 request.response.close(); | 72 _counter.send(null); |
72 }); | |
73 }); | 73 }); |
74 } | 74 } |
75 | 75 |
76 stop() { | 76 Future stop() async { |
77 print("Stopping isolate $_id"); | 77 print("Stopping isolate $_id"); |
78 _subscription.cancel(); | 78 await _subscription.cancel(); |
79 _subscription = null; | 79 _subscription = null; |
80 } | 80 } |
81 } | 81 } |
82 | 82 |
83 main(args) { | 83 main(List<String> args) async { |
84 int port = 0; | 84 int port = 0; |
85 if (args.length > 0) { | 85 if (args.length > 0) { |
86 port = int.parse(args[0]); | 86 port = int.parse(args[0]); |
87 } | 87 } |
88 RawReceivePort counter = new RawReceivePort(); | 88 |
| 89 var counter = new ReceivePort(); |
89 HttpListener listener = new EchoHttpListener(counter.sendPort); | 90 HttpListener listener = new EchoHttpListener(counter.sendPort); |
90 ServerSocket | 91 |
91 .bind(InternetAddress.ANY_IP_V6, port) | 92 // Used to ensure the requested port is available or to find an available |
92 .then((ServerSocket socket) { | 93 // port if `0` is provided. |
93 port = socket.port; | 94 ServerSocket socket = |
94 return Future.wait(new Iterable.generate(5, (_) => IsolateRunner.spawn()), | 95 await ServerSocket.bind(InternetAddress.ANY_IP_V6, port, shared: true); |
95 cleanUp: (isolate) { isolate.close(); }) | 96 |
96 .then((List<IsolateRunner> isolates) { | 97 port = socket.port; |
97 return Future.wait(isolates.map((IsolateRunner isolate) { | 98 var isolates = await Future.wait( |
98 return runHttpServer(isolate, socket, listener); | 99 new Iterable.generate(5, (_) => IsolateRunner.spawn()), |
99 }), cleanUp: (server) { server.stop(); }); | 100 cleanUp: (isolate) { |
100 }) | 101 isolate.close(); |
101 .then((stoppers) { | |
102 socket.close(); | |
103 int count = 25; | |
104 counter.handler = (_) { | |
105 count--; | |
106 if (count == 0) { | |
107 stoppers.forEach((f) => f()); | |
108 counter.close(); | |
109 } | |
110 }; | |
111 print("Server listening on port $port for 25 requests"); | |
112 print("Test with:"); | |
113 print(" ab -c10 -n 25 http://localhost:$port/"); | |
114 }); | |
115 }); | 102 }); |
| 103 |
| 104 List<RemoteStop> stoppers = await Future.wait(isolates |
| 105 .map((IsolateRunner isolate) { |
| 106 return runHttpServer(isolate, socket.port, listener); |
| 107 }), cleanUp: (server) { |
| 108 server.stop(); |
| 109 }); |
| 110 |
| 111 await socket.close(); |
| 112 int count = 25; |
| 113 |
| 114 print("Server listening on port $port for $count requests"); |
| 115 print("Test with:"); |
| 116 print(" ab -l -c10 -n $count http://localhost:$port/"); |
| 117 |
| 118 await for (var event in counter) { |
| 119 count--; |
| 120 if (count == 0) { |
| 121 print('Shutting down'); |
| 122 for (var stopper in stoppers) { |
| 123 await stopper(); |
| 124 } |
| 125 counter.close(); |
| 126 } |
| 127 } |
| 128 print('Finished'); |
116 } | 129 } |
OLD | NEW |