OLD | NEW |
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2012, 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 class _CloseToken { | 5 patch class ReceivePort { |
6 /// This token is sent from [IsolateSink]s to [IsolateStream]s to ask them to | 6 /* patch */ factory ReceivePort() = _ReceivePortImpl; |
7 /// close themselves. | 7 |
8 const _CloseToken(); | 8 /* patch */ factory ReceivePort.fromRawReceivePort(RawReceivePort rawPort) = |
| 9 _ReceivePortImpl.fromRawReceivePort; |
9 } | 10 } |
10 | 11 |
11 patch bool _isCloseToken(var object) { | 12 patch class RawReceivePort { |
12 // TODO(floitsch): can we compare against const _CloseToken()? | 13 /** |
13 return object is _CloseToken; | 14 * Opens a long-lived port for receiving messages. |
14 } | 15 * |
15 | 16 * A [RawReceivePort] is low level and does not work with [Zone]s. It |
16 patch class MessageBox { | 17 * can not be paused. The data-handler must be set before the first |
17 /* patch */ MessageBox.oneShot() : this._oneShot(new ReceivePort()); | 18 * event is received. |
18 MessageBox._oneShot(ReceivePort receivePort) | 19 */ |
19 : stream = new IsolateStream._fromOriginalReceivePortOneShot(receivePort), | 20 /* patch */ factory RawReceivePort([void handler(event)]) { |
20 sink = new _IsolateSink._fromPort(receivePort.toSendPort()); | 21 _RawReceivePortImpl result = new _RawReceivePortImpl(); |
21 | 22 result.handler = handler; |
22 /* patch */ MessageBox() : this._(new ReceivePort()); | 23 return result; |
23 MessageBox._(ReceivePort receivePort) | |
24 : stream = new IsolateStream._fromOriginalReceivePort(receivePort), | |
25 sink = new _IsolateSink._fromPort(receivePort.toSendPort()); | |
26 } | |
27 | |
28 class _IsolateSink implements IsolateSink { | |
29 bool _isClosed = false; | |
30 final SendPort _port; | |
31 _IsolateSink._fromPort(this._port); | |
32 | |
33 void add(dynamic message) { | |
34 _port.send(message); | |
35 } | |
36 | |
37 void addError(Object errorEvent) { | |
38 throw new UnimplementedError("addError on isolate streams"); | |
39 } | |
40 | |
41 void close() { | |
42 if (_isClosed) return; | |
43 add(const _CloseToken()); | |
44 _isClosed = true; | |
45 } | |
46 | |
47 bool operator==(var other) { | |
48 return other is IsolateSink && _port == other._port; | |
49 } | |
50 | |
51 int get hashCode => _port.hashCode + 499; | |
52 } | |
53 | |
54 patch IsolateSink streamSpawnFunction( | |
55 void topLevelFunction(), | |
56 [bool unhandledExceptionCallback(IsolateUnhandledException e)]) { | |
57 SendPort sendPort = spawnFunction(topLevelFunction, | |
58 unhandledExceptionCallback); | |
59 return new _IsolateSink._fromPort(sendPort); | |
60 } | |
61 | |
62 patch class ReceivePort { | |
63 /* patch */ factory ReceivePort() { | |
64 return new _ReceivePortImpl(); | |
65 } | 24 } |
66 } | 25 } |
67 | 26 |
68 class _ReceivePortImpl implements ReceivePort { | 27 class _ReceivePortImpl extends Stream implements ReceivePort { |
69 factory _ReceivePortImpl() native "ReceivePortImpl_factory"; | 28 _ReceivePortImpl() : this.fromRawReceivePort(new RawReceivePort()); |
70 | 29 |
71 receive(void onMessage(var message, SendPort replyTo)) { | 30 _ReceivePortImpl.fromRawReceivePort(this._rawPort) { |
72 _onMessage = onMessage; | 31 _controller = new StreamController(onCancel: close, sync: true); |
| 32 _rawPort.handler = _controller.add; |
73 } | 33 } |
74 | 34 |
| 35 SendPort get sendPort { |
| 36 return _rawPort.sendPort; |
| 37 } |
| 38 |
| 39 StreamSubscription listen(void onData(var message), |
| 40 { Function onError, |
| 41 void onDone(), |
| 42 bool cancelOnError }) { |
| 43 return _controller.stream.listen(onData, |
| 44 onError: onError, |
| 45 onDone: onDone, |
| 46 cancelOnError: cancelOnError); |
| 47 } |
| 48 |
| 49 close() { |
| 50 _rawPort.close(); |
| 51 _controller.close(); |
| 52 } |
| 53 |
| 54 final RawReceivePort _rawPort; |
| 55 StreamController _controller; |
| 56 } |
| 57 |
| 58 class _RawReceivePortImpl implements RawReceivePort { |
| 59 factory _RawReceivePortImpl() native "RawReceivePortImpl_factory"; |
| 60 |
75 close() { | 61 close() { |
76 _portMap.remove(_id); | 62 _portMap.remove(_id); |
77 _closeInternal(_id); | 63 _closeInternal(_id); |
78 } | 64 } |
79 | 65 |
80 SendPort toSendPort() { | 66 SendPort get sendPort { |
81 return new _SendPortImpl(_id); | 67 return new _SendPortImpl(_id); |
82 } | 68 } |
83 | 69 |
84 /**** Internal implementation details ****/ | 70 /**** Internal implementation details ****/ |
85 // Called from the VM to create a new ReceivePort instance. | 71 // Called from the VM to create a new RawReceivePort instance. |
86 static _ReceivePortImpl _get_or_create(int id) { | 72 static _RawReceivePortImpl _get_or_create(int id) { |
87 if (_portMap != null) { | 73 _RawReceivePortImpl port = _portMap[id]; |
88 _ReceivePortImpl port = _portMap[id]; | 74 if (port != null) { |
89 if (port != null) { | 75 return port; |
90 return port; | |
91 } | |
92 } | 76 } |
93 return new _ReceivePortImpl._internal(id); | 77 return new _RawReceivePortImpl._internal(id); |
94 } | 78 } |
95 | 79 |
96 _ReceivePortImpl._internal(int id) : _id = id { | 80 _RawReceivePortImpl._internal(int id) : _id = id { |
97 if (_portMap == null) { | |
98 _portMap = new Map(); | |
99 } | |
100 _portMap[id] = this; | 81 _portMap[id] = this; |
101 } | 82 } |
102 | 83 |
103 // Called from the VM to retrieve the ReceivePort for a message. | 84 // Called from the VM to retrieve the RawReceivePort for a message. |
104 static _ReceivePortImpl _lookupReceivePort(int id) { | 85 static _RawReceivePortImpl _lookupReceivePort(int id) { |
105 assert(_portMap != null); | |
106 return _portMap[id]; | 86 return _portMap[id]; |
107 } | 87 } |
108 | 88 |
109 // Called from the VM to dispatch to the handler. | 89 // Called from the VM to dispatch to the handler. |
110 static void _handleMessage(_ReceivePortImpl port, int replyId, var message) { | 90 static void _handleMessage( |
| 91 _RawReceivePortImpl port, int replyId, var message) { |
111 assert(port != null); | 92 assert(port != null); |
112 SendPort replyTo = (replyId == 0) ? null : new _SendPortImpl(replyId); | 93 port._handler(message); |
113 (port._onMessage)(message, replyTo); | |
114 } | 94 } |
115 | 95 |
116 // Call into the VM to close the VM maintained mappings. | 96 // Call into the VM to close the VM maintained mappings. |
117 static _closeInternal(int id) native "ReceivePortImpl_closeInternal"; | 97 static _closeInternal(int id) native "RawReceivePortImpl_closeInternal"; |
| 98 |
| 99 void set handler(Function newHandler) { |
| 100 this._handler = newHandler; |
| 101 } |
118 | 102 |
119 final int _id; | 103 final int _id; |
120 var _onMessage; | 104 Function _handler; |
121 | 105 |
122 // id to ReceivePort mapping. | 106 // id to RawReceivePort mapping. |
123 static Map _portMap; | 107 static final Map _portMap = new HashMap(); |
124 } | 108 } |
125 | 109 |
126 | 110 |
127 class _SendPortImpl implements SendPort { | 111 class _SendPortImpl implements SendPort { |
128 /*--- public interface ---*/ | 112 /*--- public interface ---*/ |
129 void send(var message, [SendPort replyTo = null]) { | 113 void send(var message, [SendPort replyTo = null]) { |
130 this._sendNow(message, replyTo); | 114 this._sendNow(message, replyTo); |
131 } | 115 } |
132 | 116 |
133 void _sendNow(var message, SendPort replyTo) { | 117 void _sendNow(var message, SendPort replyTo) { |
134 int replyId = (replyTo == null) ? 0 : replyTo._id; | 118 int replyId = (replyTo == null) ? 0 : replyTo._id; |
135 _sendInternal(_id, replyId, message); | 119 _sendInternal(_id, replyId, message); |
136 } | 120 } |
137 | 121 |
138 Future call(var message) { | |
139 final completer = new Completer.sync(); | |
140 final port = new _ReceivePortImpl(); | |
141 send(message, port.toSendPort()); | |
142 port.receive((value, ignoreReplyTo) { | |
143 port.close(); | |
144 if (value is Exception) { | |
145 completer.completeError(value); | |
146 } else { | |
147 completer.complete(value); | |
148 } | |
149 }); | |
150 return completer.future; | |
151 } | |
152 | |
153 bool operator==(var other) { | 122 bool operator==(var other) { |
154 return (other is _SendPortImpl) && _id == other._id; | 123 return (other is _SendPortImpl) && _id == other._id; |
155 } | 124 } |
156 | 125 |
157 int get hashCode { | 126 int get hashCode { |
158 return _id; | 127 const int MASK = 0x3FFFFFFF; |
| 128 int hash = _id; |
| 129 hash = (hash + ((hash & (MASK >> 10)) << 10)) & MASK; |
| 130 hash ^= (hash >> 6); |
| 131 hash = (hash + ((hash & (MASK >> 3)) << 3)) & MASK; |
| 132 hash ^= (hash >> 11); |
| 133 hash = (hash + ((hash & (MASK >> 15)) << 15)) & MASK; |
| 134 return hash; |
159 } | 135 } |
160 | 136 |
161 /*--- private implementation ---*/ | 137 /*--- private implementation ---*/ |
162 const _SendPortImpl(int id) : _id = id; | 138 const _SendPortImpl(int id) : _id = id; |
163 | 139 |
164 // _SendPortImpl._create is called from the VM when a new SendPort instance is | 140 // _SendPortImpl._create is called from the VM when a new SendPort instance is |
165 // needed by the VM code. | 141 // needed by the VM code. |
166 static SendPort _create(int id) { | 142 static SendPort _create(int id) { |
167 return new _SendPortImpl(id); | 143 return new _SendPortImpl(id); |
168 } | 144 } |
169 | 145 |
170 // Forward the implementation of sending messages to the VM. Only port ids | 146 // Forward the implementation of sending messages to the VM. Only port ids |
171 // are being handed to the VM. | 147 // are being handed to the VM. |
172 static _sendInternal(int sendId, int replyId, var message) | 148 static _sendInternal(int sendId, int replyId, var message) |
173 native "SendPortImpl_sendInternal_"; | 149 native "SendPortImpl_sendInternal_"; |
174 | 150 |
175 final int _id; | 151 final int _id; |
176 } | 152 } |
177 | 153 |
178 _getPortInternal() native "isolate_getPortInternal"; | 154 _getPortInternal() native "isolate_getPortInternal"; |
179 | 155 |
180 ReceivePort _portInternal; | 156 typedef _MainFunction(); |
| 157 typedef _MainFunctionArgs(args); |
| 158 typedef _MainFunctionArgsMessage(args, message); |
181 | 159 |
182 patch class _Isolate { | 160 /** |
183 /* patch */ static ReceivePort get port { | 161 * Takes the real entry point as argument and invokes it with the initial |
184 if (_portInternal == null) { | 162 * message. |
185 _portInternal = _getPortInternal(); | 163 * |
| 164 * The initial message is (currently) received through the global port variable. |
| 165 */ |
| 166 void _startIsolate(Function entryPoint, bool isSpawnUri) { |
| 167 Isolate._port.first.then((message) { |
| 168 SendPort replyTo = message[0]; |
| 169 // TODO(floitsch): don't send ok-message if we can't find the entry point. |
| 170 replyTo.send("started"); |
| 171 if (isSpawnUri) { |
| 172 assert(message.length == 3); |
| 173 List<String> args = message[1]; |
| 174 var isolateMessage = message[2]; |
| 175 if (entryPoint is _MainFunctionArgsMessage) { |
| 176 entryPoint(args, isolateMessage); |
| 177 } else if (entryPoint is _MainFunctionArgs) { |
| 178 entryPoint(args); |
| 179 } else { |
| 180 entryPoint(); |
| 181 } |
| 182 } else { |
| 183 assert(message.length == 2); |
| 184 var entryMessage = message[1]; |
| 185 entryPoint(entryMessage); |
186 } | 186 } |
187 return _portInternal; | 187 }); |
| 188 } |
| 189 |
| 190 patch class Isolate { |
| 191 /* patch */ static Future<Isolate> spawn( |
| 192 void entryPoint(message), var message) { |
| 193 Completer completer = new Completer<Isolate>.sync(); |
| 194 try { |
| 195 // The VM will invoke [_startIsolate] with entryPoint as argument. |
| 196 SendPort controlPort = _spawnFunction(entryPoint); |
| 197 RawReceivePort readyPort = new RawReceivePort(); |
| 198 controlPort.send([readyPort.sendPort, message]); |
| 199 readyPort.handler = (readyMessage) { |
| 200 assert(readyMessage == 'started'); |
| 201 readyPort.close(); |
| 202 completer.complete(new Isolate._fromControlPort(controlPort)); |
| 203 }; |
| 204 } catch(e, st) { |
| 205 // TODO(floitsch): we want errors to go into the returned future. |
| 206 rethrow; |
| 207 }; |
| 208 return completer.future; |
188 } | 209 } |
189 | 210 |
190 /* patch */ static SendPort spawnFunction(void topLevelFunction(), | 211 /* patch */ static Future<Isolate> spawnUri( |
191 [bool unhandledExceptionCallback(IsolateUnhandledException e)]) | 212 Uri uri, List<String> args, var message) { |
| 213 Completer completer = new Completer<Isolate>.sync(); |
| 214 try { |
| 215 // The VM will invoke [_startIsolate] and not `main`. |
| 216 SendPort controlPort = _spawnUri(uri.path); |
| 217 RawReceivePort readyPort = new RawReceivePort(); |
| 218 controlPort.send([readyPort.sendPort, args, message]); |
| 219 readyPort.handler = (readyMessage) { |
| 220 assert(readyMessage == 'started'); |
| 221 readyPort.close(); |
| 222 completer.complete(new Isolate._fromControlPort(controlPort)); |
| 223 }; |
| 224 } catch(e, st) { |
| 225 // TODO(floitsch): we want errors to go into the returned future. |
| 226 rethrow; |
| 227 }; |
| 228 return completer.future; |
| 229 } |
| 230 |
| 231 static final ReceivePort _port = |
| 232 new ReceivePort.fromRawReceivePort(_getPortInternal()); |
| 233 |
| 234 static SendPort _spawnFunction(Function topLevelFunction) |
192 native "isolate_spawnFunction"; | 235 native "isolate_spawnFunction"; |
193 | 236 |
194 /* patch */ static SendPort spawnUri(String uri) native "isolate_spawnUri"; | 237 static SendPort _spawnUri(String uri) native "isolate_spawnUri"; |
195 } | 238 } |
OLD | NEW |