OLD | NEW |
| (Empty) |
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 | |
3 // BSD-style license that can be found in the LICENSE file. | |
4 | |
5 /** | |
6 * This library adapts ES6 generators to implement Dart's async/await. | |
7 * | |
8 * It's designed to interact with Dart's Future/Stream and follow Dart | |
9 * async/await semantics. | |
10 * | |
11 * See https://github.com/dart-lang/dev_compiler/issues/245 for ideas on | |
12 * reconciling Dart's Future and ES6 Promise. | |
13 * | |
14 * Inspired by `co`: https://github.com/tj/co/blob/master/index.js, which is a | |
15 * stepping stone for proposed ES7 async/await, and uses ES6 Promises. | |
16 */ | |
17 dart_library.library('dart/_generators', null, /* Imports */[ | |
18 ], /* Lazy Imports */[ | |
19 'dart/_operations', | |
20 'dart/_js_helper', | |
21 'dart/core', | |
22 'dart/collection', | |
23 'dart/async' | |
24 ], function(exports, _operations, _js_helper, core, collection, async) { | |
25 'use strict'; | |
26 | |
27 const _jsIterator = Symbol('_jsIterator'); | |
28 const _current = Symbol('_current'); | |
29 | |
30 function syncStar(gen, E, ...args) { | |
31 const SyncIterable_E = _js_helper.SyncIterable$(E); | |
32 return new SyncIterable_E(gen, args); | |
33 } | |
34 exports.syncStar = syncStar; | |
35 | |
36 function async_(gen, T, ...args) { | |
37 let iter; | |
38 function onValue(res) { | |
39 if (res === void 0) res = null; | |
40 return next(iter.next(res)); | |
41 } | |
42 function onError(err) { | |
43 // If the awaited Future throws, we want to convert this to an exception | |
44 // thrown from the `yield` point, as if it was thrown there. | |
45 // | |
46 // If the exception is not caught inside `gen`, it will emerge here, which | |
47 // will send it to anyone listening on this async function's Future<T>. | |
48 // | |
49 // In essence, we are giving the code inside the generator a chance to | |
50 // use try-catch-finally. | |
51 return next(iter.throw(err)); | |
52 } | |
53 function next(ret) { | |
54 if (ret.done) return ret.value; | |
55 // Checks if the awaited value is a Future. | |
56 let future = ret.value; | |
57 if (!_operations.instanceOf(future, async.Future$)) { | |
58 future = async.Future.value(future); | |
59 } | |
60 // Chain the Future so `await` receives the Future's value. | |
61 return future.then(onValue, {onError: onError}); | |
62 } | |
63 return async.Future$(T).new(function() { | |
64 iter = gen(...args)[Symbol.iterator](); | |
65 return onValue(); | |
66 }); | |
67 } | |
68 exports.async = async_; | |
69 | |
70 // Implementation inspired by _AsyncStarStreamController in | |
71 // dart-lang/sdk's runtime/lib/core_patch.dart | |
72 // | |
73 // Given input like: | |
74 // | |
75 // foo() async* { | |
76 // yield 1; | |
77 // yield* bar(); | |
78 // print(await baz()); | |
79 // } | |
80 // | |
81 // This generates as: | |
82 // | |
83 // function foo() { | |
84 // return dart.asyncStar(function*(stream) { | |
85 // if (stream.add(1)) return; | |
86 // yield; | |
87 // if (stream.addStream(bar()) return; | |
88 // yield; | |
89 // print(yield baz()); | |
90 // }); | |
91 // } | |
92 class _AsyncStarStreamController { | |
93 constructor(generator, T, args) { | |
94 this.isAdding = false; | |
95 this.isWaiting = false; | |
96 this.isScheduled = false; | |
97 this.isSuspendedAtYield = false; | |
98 this.canceler = null; | |
99 this.iterator = generator(this, ...args)[Symbol.iterator](); | |
100 this.controller = async.StreamController$(T).new({ | |
101 onListen: () => this.scheduleGenerator(), | |
102 onResume: () => this.onResume(), | |
103 onCancel: () => this.onCancel() | |
104 }); | |
105 } | |
106 | |
107 onResume() { | |
108 if (this.isSuspendedAtYield) { | |
109 this.scheduleGenerator(); | |
110 } | |
111 } | |
112 | |
113 onCancel() { | |
114 if (this.controller.isClosed) { | |
115 return null; | |
116 } | |
117 if (this.canceler == null) { | |
118 this.canceler = async.Completer.new(); | |
119 this.scheduleGenerator(); | |
120 } | |
121 return this.canceler.future; | |
122 } | |
123 | |
124 close() { | |
125 if (this.canceler != null && !this.canceler.isCompleted) { | |
126 // If the stream has been cancelled, complete the cancellation future | |
127 // with the error. | |
128 this.canceler.complete(); | |
129 } | |
130 this.controller.close(); | |
131 } | |
132 | |
133 scheduleGenerator() { | |
134 // TODO(jmesserly): is this paused check in the right place? Assuming the | |
135 // async* Stream yields, then is paused (by other code), the body will | |
136 // already be scheduled. This will cause at least one more iteration to | |
137 // run (adding another data item to the Stream) before actually pausing. | |
138 // It could be fixed by moving the `isPaused` check inside `runBody`. | |
139 if (this.isScheduled || this.controller.isPaused || | |
140 this.isAdding || this.isWaiting) { | |
141 return; | |
142 } | |
143 this.isScheduled = true; | |
144 async.scheduleMicrotask(() => this.runBody()); | |
145 } | |
146 | |
147 runBody(opt_awaitValue) { | |
148 this.isScheduled = false; | |
149 this.isSuspendedAtYield = false; | |
150 this.isWaiting = false; | |
151 let iter; | |
152 try { | |
153 iter = this.iterator.next(opt_awaitValue); | |
154 } catch (e) { | |
155 this.addError(e, _operations.stackTrace(e)); | |
156 this.close(); | |
157 return; | |
158 } | |
159 if (iter.done) { | |
160 this.close(); | |
161 return; | |
162 } | |
163 | |
164 // If we're suspended at a yield/yield*, we're done for now. | |
165 if (this.isSuspendedAtYield || this.isAdding) return; | |
166 | |
167 // Handle `await`: if we get a value passed to `yield` it means we are | |
168 // waiting on this Future. Make sure to prevent scheduling, and pass the | |
169 // value back as the result of the `yield`. | |
170 // | |
171 // TODO(jmesserly): is the timing here correct? The assumption here is | |
172 // that we should schedule `await` in `async*` the same as in `async`. | |
173 this.isWaiting = true; | |
174 let future = iter.value; | |
175 if (!_operations.instanceOf(future, async.Future$)) { | |
176 future = async.Future.value(future); | |
177 } | |
178 return future.then((x) => this.runBody(x), | |
179 { onError: (e, s) => this.throwError(e, s) }); | |
180 } | |
181 | |
182 // Adds element to stream, returns true if the caller should terminate | |
183 // execution of the generator. | |
184 add(event) { | |
185 // If stream is cancelled, tell caller to exit the async generator. | |
186 if (!this.controller.hasListener) return true; | |
187 this.controller.add(event); | |
188 this.scheduleGenerator(); | |
189 this.isSuspendedAtYield = true; | |
190 return false; | |
191 } | |
192 | |
193 // Adds the elements of stream into this controller's stream. | |
194 // The generator will be scheduled again when all of the | |
195 // elements of the added stream have been consumed. | |
196 // Returns true if the caller should terminate | |
197 // execution of the generator. | |
198 addStream(stream) { | |
199 // If stream is cancelled, tell caller to exit the async generator. | |
200 if (!this.controller.hasListener) return true; | |
201 | |
202 this.isAdding = true; | |
203 this.controller.addStream(stream, {cancelOnError: false}).then(() => { | |
204 this.isAdding = false; | |
205 this.scheduleGenerator(); | |
206 }, { onError: (e, s) => this.throwError(e, s) }); | |
207 } | |
208 | |
209 throwError(error, stackTrace) { | |
210 try { | |
211 this.iterator.throw(error); | |
212 } catch (e) { | |
213 this.addError(e, stackTrace); | |
214 } | |
215 } | |
216 | |
217 addError(error, stackTrace) { | |
218 if ((this.canceler != null) && !this.canceler.isCompleted) { | |
219 // If the stream has been cancelled, complete the cancellation future | |
220 // with the error. | |
221 this.canceler.completeError(error, stackTrace); | |
222 return; | |
223 } | |
224 if (!this.controller.hasListener) return; | |
225 this.controller.addError(error, stackTrace); | |
226 } | |
227 } | |
228 | |
229 /** Returns a Stream of T implemented by an async* function. */ | |
230 function asyncStar(gen, T, ...args) { | |
231 return new _AsyncStarStreamController(gen, T, args).controller.stream; | |
232 } | |
233 exports.asyncStar = asyncStar; | |
234 }); | |
OLD | NEW |