| OLD | NEW |
|---|
| (Empty) |
| 1 // Copyright (c) 2011, the Dart project authors. Please see the AUTHORS file | |
| 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. | |
| 4 | |
| 5 // Dart core library. | |
| 6 | |
| 7 class PromiseImpl<T> implements Promise<T> { | |
| 8 | |
| 9 // Enumeration of possible states: | |
| 10 static final int CREATED = 0; | |
| 11 static final int RUNNING = 1; | |
| 12 static final int COMPLETE_NORMAL = 2; | |
| 13 static final int COMPLETE_ERROR = 3; | |
| 14 static final int CANCELLED = 4; | |
| 15 | |
| 16 // TODO(sigmund): consider whether this is what people want, or if we should | |
| 17 // discard values/errors after cancellation. | |
| 18 static final int COMPLETE_NORMAL_AFTER_CANCELLED = 5; | |
| 19 static final int COMPLETE_ERROR_AFTER_CANCELLED = 6; | |
| 20 | |
| 21 | |
| 22 /** Internal state, one of the above constants. */ | |
| 23 int _state; | |
| 24 | |
| 25 /** Value that was provided, if any. */ | |
| 26 T _value; | |
| 27 | |
| 28 /** Error that was provided, if any. */ | |
| 29 var _error; | |
| 30 | |
| 31 /** Listeners waiting for a normal completion of this promise. */ | |
| 32 Queue<Function> _normalListeners; | |
| 33 | |
| 34 /** Error listeners. */ | |
| 35 Queue<Function> _errorListeners; | |
| 36 | |
| 37 /** Cancellation listeners. */ | |
| 38 Queue<Function> _cancelListeners; | |
| 39 | |
| 40 PromiseImpl() | |
| 41 : _state = CREATED, | |
| 42 _value = null, | |
| 43 _error = null, | |
| 44 _normalListeners = null, | |
| 45 _errorListeners = null, | |
| 46 _cancelListeners = null {} | |
| 47 | |
| 48 PromiseImpl.fromValue(T val) | |
| 49 : _state = COMPLETE_NORMAL, | |
| 50 _value = val, | |
| 51 _error = null, | |
| 52 _normalListeners = null, | |
| 53 _errorListeners = null, | |
| 54 _cancelListeners = null {} | |
| 55 | |
| 56 // Properties and methods from Promise: | |
| 57 | |
| 58 T get value() { | |
| 59 if (!isDone()) { | |
| 60 // TODO(kasperl): Turn this into a proper exception object. | |
| 61 throw new Exception("Attempted to get the value of an uncompleted promise.
"); | |
| 62 } | |
| 63 if (hasError()) { | |
| 64 throw _error; | |
| 65 } else { | |
| 66 return _value; | |
| 67 } | |
| 68 } | |
| 69 | |
| 70 get error() { | |
| 71 if (!isDone()) { | |
| 72 // TODO(kasperl): Turn this into a proper exception object. | |
| 73 throw "Attempted to examine the state of an uncompleted promise."; | |
| 74 } | |
| 75 return _error; | |
| 76 } | |
| 77 | |
| 78 bool isDone() { | |
| 79 return _state != CREATED && _state != RUNNING; | |
| 80 } | |
| 81 | |
| 82 bool isCancelled() { | |
| 83 return _state == CANCELLED | |
| 84 || _state == COMPLETE_NORMAL_AFTER_CANCELLED | |
| 85 || _state == COMPLETE_ERROR_AFTER_CANCELLED; | |
| 86 } | |
| 87 | |
| 88 bool hasValue() { | |
| 89 return _state == COMPLETE_NORMAL | |
| 90 || _state == COMPLETE_NORMAL_AFTER_CANCELLED; | |
| 91 } | |
| 92 | |
| 93 bool hasError() { | |
| 94 return _state == COMPLETE_ERROR | |
| 95 || _state == COMPLETE_ERROR_AFTER_CANCELLED; | |
| 96 } | |
| 97 | |
| 98 void complete(T newVal) { | |
| 99 if (_state == CANCELLED) { | |
| 100 _value = newVal; | |
| 101 _state = COMPLETE_NORMAL_AFTER_CANCELLED; | |
| 102 return; | |
| 103 } | |
| 104 | |
| 105 if (isDone()) { | |
| 106 throw "Attempted to complete an already completed promise."; | |
| 107 } | |
| 108 | |
| 109 _value = newVal; | |
| 110 _state = COMPLETE_NORMAL; | |
| 111 if (_normalListeners !== null) { | |
| 112 _normalListeners.forEach((listener) { | |
| 113 listener(newVal); | |
| 114 }); | |
| 115 } | |
| 116 _clearListeners(); | |
| 117 } | |
| 118 | |
| 119 void _clearListeners() { | |
| 120 _normalListeners = null; | |
| 121 _errorListeners = null; | |
| 122 _cancelListeners = null; | |
| 123 } | |
| 124 | |
| 125 void fail(var err) { | |
| 126 if (_state == CANCELLED) { | |
| 127 _error = err; | |
| 128 _state = COMPLETE_ERROR_AFTER_CANCELLED; | |
| 129 return; | |
| 130 } | |
| 131 | |
| 132 if (isDone()) { | |
| 133 throw "Can't fail an already completed promise."; | |
| 134 } | |
| 135 | |
| 136 _error = err; | |
| 137 _state = COMPLETE_ERROR; | |
| 138 if (_errorListeners !== null) { | |
| 139 _errorListeners.forEach((listener) { | |
| 140 listener(err); | |
| 141 }); | |
| 142 } | |
| 143 _clearListeners(); | |
| 144 } | |
| 145 | |
| 146 bool cancel() { | |
| 147 if (!isDone()) { | |
| 148 _state = CANCELLED; | |
| 149 if (_cancelListeners !== null) { | |
| 150 _cancelListeners.forEach((listener) { | |
| 151 listener(); | |
| 152 }); | |
| 153 } | |
| 154 _clearListeners(); | |
| 155 return true; | |
| 156 } | |
| 157 return false; | |
| 158 } | |
| 159 | |
| 160 void addCompleteHandler(void completeHandler(T result)) { | |
| 161 if (_state == COMPLETE_NORMAL) { | |
| 162 completeHandler(_value); | |
| 163 } else if (!isDone()) { | |
| 164 if (_normalListeners === null) { | |
| 165 _normalListeners = new Queue<Function>(); | |
| 166 } | |
| 167 _normalListeners.addLast(completeHandler); | |
| 168 } | |
| 169 } | |
| 170 | |
| 171 void addErrorHandler(void errorHandler(err)) { | |
| 172 if (_state == COMPLETE_ERROR) { | |
| 173 errorHandler(_error); | |
| 174 } else if (!isDone()) { | |
| 175 if (_errorListeners === null) { | |
| 176 _errorListeners = new Queue<Function>(); | |
| 177 } | |
| 178 _errorListeners.addLast(errorHandler); | |
| 179 } | |
| 180 } | |
| 181 | |
| 182 void addCancelHandler(void cancelHandler()) { | |
| 183 if (isCancelled()) { | |
| 184 cancelHandler(); | |
| 185 } else if (!isDone()) { | |
| 186 if (_cancelListeners === null) { | |
| 187 _cancelListeners = new Queue<Function>(); | |
| 188 } | |
| 189 _cancelListeners.addLast(cancelHandler); | |
| 190 } | |
| 191 } | |
| 192 | |
| 193 // TODO(sigmund): consider adding to the API a method that does the following: | |
| 194 // Promise chain(callback(T r)) { | |
| 195 // return then(callback).flatten(); | |
| 196 // } | |
| 197 | |
| 198 Promise then(callback(T result)) { | |
| 199 Promise promise = new Promise(); | |
| 200 addCompleteHandler((T val) { | |
| 201 promise.complete(callback(val)); | |
| 202 }); | |
| 203 addErrorHandler((err) { promise.fail(err); }); | |
| 204 addCancelHandler(() { | |
| 205 promise.fail("Source promise was cancelled"); | |
| 206 }); | |
| 207 return promise; | |
| 208 } | |
| 209 | |
| 210 // TODO(sigmund): consider adding to the API a method to represent | |
| 211 // containment. For instance, promiseA spawns internally promiseB, where | |
| 212 // promiseB is not visible to the user. Like [then] this creates a relation | |
| 213 // between A, and B: | |
| 214 // - b.complete -> a.complete (as above) | |
| 215 // - b.error -> a.error (as above) | |
| 216 // - b.cancel -> a.error (as above) | |
| 217 // - a.cancel -> b.cancel (unlike [then]) | |
| 218 | |
| 219 Promise flatten() { | |
| 220 Promise res = new Promise(); | |
| 221 then((T thisVal) { | |
| 222 if (thisVal is Promise) { | |
| 223 Promise thisPromise = thisVal.dynamic; | |
| 224 thisPromise.flatten().then((lastVal) { | |
| 225 res.complete(lastVal); | |
| 226 }); | |
| 227 } else { | |
| 228 res.complete(thisVal); | |
| 229 } | |
| 230 }); | |
| 231 return res; | |
| 232 } | |
| 233 | |
| 234 void join(Collection<Promise> promises, bool joinDone(Promise completed)) { | |
| 235 promises.forEach((promise) { | |
| 236 promise.addCompleteHandler((value) { | |
| 237 if (joinDone(promise)) { | |
| 238 complete(value); | |
| 239 } | |
| 240 }); | |
| 241 promise.addErrorHandler((err) { | |
| 242 fail(err); | |
| 243 }); | |
| 244 }); | |
| 245 addCancelHandler(() { | |
| 246 promises.forEach((promise) { | |
| 247 promise.cancel(); | |
| 248 }); | |
| 249 }); | |
| 250 } | |
| 251 | |
| 252 void waitFor(Collection<Promise> promises, int n) { | |
| 253 int counter = 0; | |
| 254 join(promises, (p) => (++counter == n)); | |
| 255 addCompleteHandler((val) { | |
| 256 promises.forEach((promise) { | |
| 257 if (!promise.isDone()) { | |
| 258 promise.cancel(); | |
| 259 } | |
| 260 }); | |
| 261 }); | |
| 262 } | |
| 263 } | |
| 264 | |
| 265 // For now, extend Promise<bool> rather than either | |
| 266 // a) create a new base, Completable, for Promise and Proxy, or | |
| 267 // b) extend Promise<SendPort> which would expose the port. | |
| 268 class ProxyBase extends PromiseImpl<bool> { | |
| 269 | |
| 270 ProxyBase.forPort(SendPort port) { | |
| 271 _promise = new Promise<SendPort>(); | |
| 272 _promise.complete(port); | |
| 273 complete(true); | |
| 274 } | |
| 275 | |
| 276 // Construct a proxy for a message reply; see the [Proxy.forReply] | |
| 277 // documentation for more details. | |
| 278 ProxyBase.forReply(Promise<SendPort> port) { | |
| 279 _promise = port; | |
| 280 port.addCompleteHandler((_) => complete(true)); | |
| 281 } | |
| 282 | |
| 283 // Note that comparing proxies or using them in maps or sets is | |
| 284 // illegal until they complete. | |
| 285 bool operator ==(var other) { | |
| 286 return (other is ProxyBase) && _promise.value == other._promise.value; | |
| 287 } | |
| 288 | |
| 289 int hashCode() => _promise.value.hashCode(); | |
| 290 | |
| 291 static ReceivePort register(Dispatcher dispatcher) { | |
| 292 if (_dispatchers === null) { | |
| 293 _dispatchers = new Map<SendPort, Dispatcher>(); | |
| 294 } | |
| 295 ReceivePort result = new ReceivePort(); | |
| 296 _dispatchers[result.toSendPort()] = dispatcher; | |
| 297 return result; | |
| 298 } | |
| 299 | |
| 300 get local() { | |
| 301 if (_dispatchers !== null) { | |
| 302 Dispatcher dispatcher = _dispatchers[_promise.value]; | |
| 303 if (dispatcher !== null) return dispatcher.target; | |
| 304 } | |
| 305 throw new Exception("Cannot access object of non-local proxy."); | |
| 306 } | |
| 307 | |
| 308 void send(List message) { | |
| 309 _marshal(message, (List marshalled) { | |
| 310 SendPort port = _promise.value; | |
| 311 port.send(marshalled, null); | |
| 312 }); | |
| 313 } | |
| 314 | |
| 315 Promise call(List message) { | |
| 316 return _marshal(message, (List marshalled) { | |
| 317 // TODO(kasperl): For now, the [Promise.then] implementation allows | |
| 318 // me to return a promise and it will do the promise chaining. | |
| 319 final result = new Promise(); | |
| 320 // The promise queue implementation guarantees that promise is | |
| 321 // resolved at this point. | |
| 322 SendPort outgoing = _promise.value; | |
| 323 ReceivePort incoming = outgoing.call(marshalled); | |
| 324 incoming.receive((List receiveMessage, replyTo) { | |
| 325 result.complete(receiveMessage[0]); | |
| 326 }); | |
| 327 return result; | |
| 328 }); | |
| 329 } | |
| 330 | |
| 331 // Marshal the [message] and pass it to the [process] callback | |
| 332 // function. Any promises are converted to a port which expects to | |
| 333 // receive a port from the other side down which the remote promise | |
| 334 // can be completed by sending the promise's completion value. | |
| 335 Promise _marshal(List message, process(List marshalled)) { | |
| 336 return _promise.then((SendPort port) { | |
| 337 List marshalled = new List(message.length); | |
| 338 | |
| 339 for (int i = 0; i < marshalled.length; i++) { | |
| 340 var entry = message[i]; | |
| 341 if (entry is Proxy) { | |
| 342 entry = entry._promise; | |
| 343 } | |
| 344 // Obviously this will be true if [entry] was a Proxy. | |
| 345 if (entry is Promise) { | |
| 346 // Note that we could optimise this by just sending the value | |
| 347 // if the promise is already complete. Let's get this working | |
| 348 // first! | |
| 349 | |
| 350 // This port will receive a SendPort that can be used to | |
| 351 // signal completion of this promise to the corresponding | |
| 352 // promise that the other end has created. | |
| 353 ReceivePort receiveCompleter = new ReceivePort.singleShot(); | |
| 354 marshalled[i] = receiveCompleter.toSendPort(); | |
| 355 Promise<SendPort> completer = new Promise<SendPort>(); | |
| 356 receiveCompleter.receive((var msg, SendPort replyPort) { | |
| 357 completer.complete(msg[0]); | |
| 358 }); | |
| 359 entry.addCompleteHandler((value) { | |
| 360 completer.addCompleteHandler((SendPort completePort) { | |
| 361 _marshal([value], (List completeMessage) => completePort.send(comp
leteMessage, null)); | |
| 362 }); | |
| 363 }); | |
| 364 } else { | |
| 365 // FIXME(kasperl, benl): this should probably be a copy? | |
| 366 marshalled[i] = entry; | |
| 367 } | |
| 368 if (marshalled[i] is ReceivePort) { | |
| 369 throw new Exception("Despite the documentation, you cannot send a Rece
ivePort"); | |
| 370 } | |
| 371 } | |
| 372 return process(marshalled); | |
| 373 }).flatten(); | |
| 374 } | |
| 375 | |
| 376 Promise<SendPort> _promise; | |
| 377 static Map<SendPort, Dispatcher> _dispatchers; | |
| 378 | |
| 379 } | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 9401030:
remove promise from corelib (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart