OLD | NEW |
1 // Copyright 2015 The Chromium Authors. All rights reserved. | 1 // Copyright 2015 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #include "modules/fetch/ReadableStreamDataConsumerHandle.h" | 5 #include "modules/fetch/ReadableStreamDataConsumerHandle.h" |
6 | 6 |
7 #include "bindings/core/v8/ExceptionState.h" | 7 #include "bindings/core/v8/ExceptionState.h" |
8 #include "bindings/core/v8/ReadableStreamOperations.h" | 8 #include "bindings/core/v8/ReadableStreamOperations.h" |
| 9 #include "bindings/core/v8/ScopedPersistent.h" |
9 #include "bindings/core/v8/ScriptFunction.h" | 10 #include "bindings/core/v8/ScriptFunction.h" |
10 #include "bindings/core/v8/ScriptState.h" | 11 #include "bindings/core/v8/ScriptState.h" |
11 #include "bindings/core/v8/ScriptValue.h" | 12 #include "bindings/core/v8/ScriptValue.h" |
12 #include "bindings/core/v8/V8BindingMacros.h" | 13 #include "bindings/core/v8/V8BindingMacros.h" |
13 #include "bindings/core/v8/V8IteratorResultValue.h" | 14 #include "bindings/core/v8/V8IteratorResultValue.h" |
14 #include "bindings/core/v8/V8RecursionScope.h" | 15 #include "bindings/core/v8/V8RecursionScope.h" |
15 #include "bindings/core/v8/V8Uint8Array.h" | 16 #include "bindings/core/v8/V8Uint8Array.h" |
16 #include "core/dom/DOMTypedArray.h" | 17 #include "core/dom/DOMTypedArray.h" |
17 #include "public/platform/Platform.h" | 18 #include "public/platform/Platform.h" |
18 #include "public/platform/WebTaskRunner.h" | 19 #include "public/platform/WebTaskRunner.h" |
19 #include "public/platform/WebThread.h" | 20 #include "public/platform/WebThread.h" |
20 #include "public/platform/WebTraceLocation.h" | 21 #include "public/platform/WebTraceLocation.h" |
21 #include "wtf/Assertions.h" | 22 #include "wtf/Assertions.h" |
22 #include "wtf/Functional.h" | 23 #include "wtf/Functional.h" |
23 #include "wtf/RefCounted.h" | 24 #include "wtf/RefCounted.h" |
24 #include "wtf/WeakPtr.h" | |
25 #include <algorithm> | 25 #include <algorithm> |
26 #include <string.h> | 26 #include <string.h> |
27 #include <v8.h> | 27 #include <v8.h> |
28 | 28 |
29 namespace blink { | 29 namespace blink { |
30 | 30 |
31 using Result = WebDataConsumerHandle::Result; | 31 using Result = WebDataConsumerHandle::Result; |
32 using Flags = WebDataConsumerHandle::Flags; | 32 using Flags = WebDataConsumerHandle::Flags; |
33 | 33 |
34 // This context is not yet thread-safe. | 34 // This context is not yet thread-safe. |
35 class ReadableStreamDataConsumerHandle::ReadingContext final : public RefCounted
<ReadingContext> { | 35 class ReadableStreamDataConsumerHandle::ReadingContext final : public RefCounted
<ReadingContext> { |
36 WTF_MAKE_NONCOPYABLE(ReadingContext); | 36 WTF_MAKE_NONCOPYABLE(ReadingContext); |
37 public: | 37 public: |
38 class OnFulfilled final : public ScriptFunction { | 38 class OnFulfilled final : public ScriptFunction { |
39 public: | 39 public: |
40 static v8::Local<v8::Function> createFunction(ScriptState* scriptState,
WeakPtr<ReadingContext> context) | 40 static v8::Local<v8::Function> createFunction(ScriptState* scriptState,
PassRefPtr<ReadingContext> context) |
41 { | 41 { |
42 return (new OnFulfilled(scriptState, context))->bindToV8Function(); | 42 return (new OnFulfilled(scriptState, context))->bindToV8Function(); |
43 } | 43 } |
44 | 44 |
45 ScriptValue call(ScriptValue v) override | 45 ScriptValue call(ScriptValue v) override |
46 { | 46 { |
47 RefPtr<ReadingContext> readingContext(m_readingContext.get()); | 47 RefPtr<ReadingContext> readingContext(m_readingContext); |
48 if (!readingContext) | 48 if (!readingContext) |
49 return v; | 49 return v; |
50 bool done; | 50 bool done; |
51 v8::Local<v8::Value> item = v.v8Value(); | 51 v8::Local<v8::Value> item = v.v8Value(); |
52 ASSERT(item->IsObject()); | 52 ASSERT(item->IsObject()); |
53 v8::Local<v8::Value> value = v8CallOrCrash(v8UnpackIteratorResult(v.
scriptState(), item.As<v8::Object>(), &done)); | 53 v8::Local<v8::Value> value = v8CallOrCrash(v8UnpackIteratorResult(v.
scriptState(), item.As<v8::Object>(), &done)); |
54 if (done) { | 54 if (done) { |
55 readingContext->onReadDone(); | 55 readingContext->onReadDone(); |
56 return v; | 56 return v; |
57 } | 57 } |
58 if (!V8Uint8Array::hasInstance(value, v.isolate())) { | 58 if (!V8Uint8Array::hasInstance(value, v.isolate())) { |
59 readingContext->onRejected(); | 59 readingContext->onRejected(); |
60 return ScriptValue(); | 60 return ScriptValue(); |
61 } | 61 } |
62 readingContext->onRead(V8Uint8Array::toImpl(value.As<v8::Object>()))
; | 62 readingContext->onRead(V8Uint8Array::toImpl(value.As<v8::Object>()))
; |
63 return v; | 63 return v; |
64 } | 64 } |
65 | 65 |
66 private: | 66 private: |
67 OnFulfilled(ScriptState* scriptState, WeakPtr<ReadingContext> context) | 67 OnFulfilled(ScriptState* scriptState, PassRefPtr<ReadingContext> context
) |
68 : ScriptFunction(scriptState), m_readingContext(context) {} | 68 : ScriptFunction(scriptState), m_readingContext(context) {} |
69 | 69 |
70 WeakPtr<ReadingContext> m_readingContext; | 70 RefPtr<ReadingContext> m_readingContext; |
71 }; | 71 }; |
72 | 72 |
73 class OnRejected final : public ScriptFunction { | 73 class OnRejected final : public ScriptFunction { |
74 public: | 74 public: |
75 static v8::Local<v8::Function> createFunction(ScriptState* scriptState,
WeakPtr<ReadingContext> context) | 75 static v8::Local<v8::Function> createFunction(ScriptState* scriptState,
PassRefPtr<ReadingContext> context) |
76 { | 76 { |
77 return (new OnRejected(scriptState, context))->bindToV8Function(); | 77 return (new OnRejected(scriptState, context))->bindToV8Function(); |
78 } | 78 } |
79 | 79 |
80 ScriptValue call(ScriptValue v) override | 80 ScriptValue call(ScriptValue v) override |
81 { | 81 { |
82 RefPtr<ReadingContext> readingContext(m_readingContext.get()); | 82 RefPtr<ReadingContext> readingContext(m_readingContext); |
83 if (!readingContext) | 83 if (!readingContext) |
84 return v; | 84 return v; |
85 readingContext->onRejected(); | 85 readingContext->onRejected(); |
86 return v; | 86 return v; |
87 } | 87 } |
88 | 88 |
89 private: | 89 private: |
90 OnRejected(ScriptState* scriptState, WeakPtr<ReadingContext> context) | 90 OnRejected(ScriptState* scriptState, PassRefPtr<ReadingContext> context) |
91 : ScriptFunction(scriptState), m_readingContext(context) {} | 91 : ScriptFunction(scriptState), m_readingContext(context) {} |
92 | 92 |
93 WeakPtr<ReadingContext> m_readingContext; | 93 RefPtr<ReadingContext> m_readingContext; |
94 }; | 94 }; |
95 | 95 |
96 class ReaderImpl final : public FetchDataConsumerHandle::Reader { | 96 class ReaderImpl final : public FetchDataConsumerHandle::Reader { |
97 public: | 97 public: |
98 ReaderImpl(PassRefPtr<ReadingContext> context, Client* client) | 98 ReaderImpl(PassRefPtr<ReadingContext> context, Client* client) |
99 : m_readingContext(context) | 99 : m_readingContext(context) |
100 { | 100 { |
101 m_readingContext->attachReader(client); | 101 m_readingContext->attachReader(client); |
102 } | 102 } |
103 ~ReaderImpl() override | 103 ~ReaderImpl() override |
104 { | 104 { |
105 m_readingContext->detachReader(); | 105 m_readingContext->detachReader(); |
106 } | 106 } |
107 | 107 |
108 Result beginRead(const void** buffer, Flags, size_t* available) override | 108 Result beginRead(const void** buffer, Flags, size_t* available) override |
109 { | 109 { |
110 return m_readingContext->beginRead(buffer, available); | 110 return m_readingContext->beginRead(buffer, available); |
111 } | 111 } |
112 | 112 |
113 Result endRead(size_t readSize) override | 113 Result endRead(size_t readSize) override |
114 { | 114 { |
115 return m_readingContext->endRead(readSize); | 115 return m_readingContext->endRead(readSize); |
116 } | 116 } |
117 | 117 |
118 private: | 118 private: |
119 RefPtr<ReadingContext> m_readingContext; | 119 RefPtr<ReadingContext> m_readingContext; |
120 }; | 120 }; |
121 | 121 |
122 static PassRefPtr<ReadingContext> create(ScriptState* scriptState, ScriptVal
ue stream) | 122 static PassRefPtr<ReadingContext> create(ScriptState* scriptState, ScriptVal
ue streamReader) |
123 { | 123 { |
124 return adoptRef(new ReadingContext(scriptState, stream)); | 124 return adoptRef(new ReadingContext(scriptState, streamReader)); |
125 } | 125 } |
126 | 126 |
127 void attachReader(WebDataConsumerHandle::Client* client) | 127 void attachReader(WebDataConsumerHandle::Client* client) |
128 { | 128 { |
129 m_client = client; | 129 m_client = client; |
130 notifyLater(); | 130 notifyLater(); |
131 } | 131 } |
132 | 132 |
133 void detachReader() | 133 void detachReader() |
134 { | 134 { |
135 m_client = nullptr; | 135 m_client = nullptr; |
136 } | 136 } |
137 | 137 |
138 Result beginRead(const void** buffer, size_t* available) | 138 Result beginRead(const void** buffer, size_t* available) |
139 { | 139 { |
140 *buffer = nullptr; | 140 *buffer = nullptr; |
141 *available = 0; | 141 *available = 0; |
142 if (m_hasError) | 142 if (m_hasError) |
143 return WebDataConsumerHandle::UnexpectedError; | 143 return WebDataConsumerHandle::UnexpectedError; |
144 if (m_isDone) | 144 if (m_isDone) |
145 return WebDataConsumerHandle::Done; | 145 return WebDataConsumerHandle::Done; |
146 | 146 |
147 if (m_pendingBuffer) { | 147 if (m_pendingBuffer) { |
148 ASSERT(m_pendingOffset < m_pendingBuffer->length()); | 148 ASSERT(m_pendingOffset < m_pendingBuffer->length()); |
149 *buffer = m_pendingBuffer->data() + m_pendingOffset; | 149 *buffer = m_pendingBuffer->data() + m_pendingOffset; |
150 *available = m_pendingBuffer->length() - m_pendingOffset; | 150 *available = m_pendingBuffer->length() - m_pendingOffset; |
151 return WebDataConsumerHandle::Ok; | 151 return WebDataConsumerHandle::Ok; |
152 } | 152 } |
153 ASSERT(!m_reader.isEmpty()); | |
154 m_isInRecursion = true; | |
155 if (!m_isReading) { | 153 if (!m_isReading) { |
156 m_isReading = true; | 154 m_isReading = true; |
157 ScriptState::Scope scope(m_reader.scriptState()); | 155 ScriptState::Scope scope(m_scriptState.get()); |
158 V8RecursionScope recursionScope(m_reader.isolate()); | 156 ScriptValue reader(m_scriptState.get(), m_reader.newLocal(m_scriptSt
ate->isolate())); |
159 ReadableStreamOperations::read(m_reader.scriptState(), m_reader).the
n( | 157 if (reader.isEmpty()) { |
160 OnFulfilled::createFunction(m_reader.scriptState(), m_weakPtrFac
tory.createWeakPtr()), | 158 // The reader was collected. |
161 OnRejected::createFunction(m_reader.scriptState(), m_weakPtrFact
ory.createWeakPtr())); | 159 m_hasError = true; |
162 // Note: Microtasks may run here. | 160 m_isReading = false; |
| 161 return WebDataConsumerHandle::UnexpectedError; |
| 162 } |
| 163 ReadableStreamOperations::read(m_scriptState.get(), reader).then( |
| 164 OnFulfilled::createFunction(m_scriptState.get(), this), |
| 165 OnRejected::createFunction(m_scriptState.get(), this)); |
163 } | 166 } |
164 m_isInRecursion = false; | |
165 return WebDataConsumerHandle::ShouldWait; | 167 return WebDataConsumerHandle::ShouldWait; |
166 } | 168 } |
167 | 169 |
168 Result endRead(size_t readSize) | 170 Result endRead(size_t readSize) |
169 { | 171 { |
170 ASSERT(m_pendingBuffer); | 172 ASSERT(m_pendingBuffer); |
171 ASSERT(m_pendingOffset + readSize <= m_pendingBuffer->length()); | 173 ASSERT(m_pendingOffset + readSize <= m_pendingBuffer->length()); |
172 m_pendingOffset += readSize; | 174 m_pendingOffset += readSize; |
173 if (m_pendingOffset == m_pendingBuffer->length()) { | 175 if (m_pendingOffset == m_pendingBuffer->length()) { |
174 m_pendingBuffer = nullptr; | 176 m_pendingBuffer = nullptr; |
(...skipping 30 matching lines...) Expand all Loading... |
205 m_hasError = true; | 207 m_hasError = true; |
206 m_isReading = false; | 208 m_isReading = false; |
207 m_reader.clear(); | 209 m_reader.clear(); |
208 notify(); | 210 notify(); |
209 } | 211 } |
210 | 212 |
211 void notify() | 213 void notify() |
212 { | 214 { |
213 if (!m_client) | 215 if (!m_client) |
214 return; | 216 return; |
215 if (m_isInRecursion) { | |
216 notifyLater(); | |
217 return; | |
218 } | |
219 m_client->didGetReadable(); | 217 m_client->didGetReadable(); |
220 } | 218 } |
221 | 219 |
222 void notifyLater() | 220 void notifyLater() |
223 { | 221 { |
224 ASSERT(m_client); | 222 ASSERT(m_client); |
225 Platform::current()->currentThread()->taskRunner()->postTask(BLINK_FROM_
HERE, bind(&ReadingContext::notify, PassRefPtr<ReadingContext>(this))); | 223 Platform::current()->currentThread()->taskRunner()->postTask(BLINK_FROM_
HERE, bind(&ReadingContext::notify, PassRefPtr<ReadingContext>(this))); |
226 } | 224 } |
227 | 225 |
228 private: | 226 private: |
229 ReadingContext(ScriptState* scriptState, ScriptValue stream) | 227 ReadingContext(ScriptState* scriptState, ScriptValue streamReader) |
230 : m_client(nullptr) | 228 : m_reader(scriptState->isolate(), streamReader.v8Value()) |
231 , m_weakPtrFactory(this) | 229 , m_scriptState(scriptState) |
| 230 , m_client(nullptr) |
232 , m_pendingOffset(0) | 231 , m_pendingOffset(0) |
233 , m_isReading(false) | 232 , m_isReading(false) |
234 , m_isDone(false) | 233 , m_isDone(false) |
235 , m_hasError(false) | 234 , m_hasError(false) |
236 , m_isInRecursion(false) | |
237 { | 235 { |
238 if (!ReadableStreamOperations::isLocked(scriptState, stream)) { | 236 m_reader.setWeak(this, &ReadingContext::onCollected); |
239 // Here the stream implementation must not throw an exception. | |
240 NonThrowableExceptionState es; | |
241 m_reader = ReadableStreamOperations::getReader(scriptState, stream,
es); | |
242 } | |
243 m_hasError = m_reader.isEmpty(); | |
244 } | 237 } |
245 | 238 |
246 // This ScriptValue is leaky because it stores a strong reference to a | 239 void onCollected() |
247 // JavaScript object. | 240 { |
248 // TODO(yhirano): Fix it. | 241 m_reader.clear(); |
249 // | 242 if (m_isDone || m_hasError) |
250 // Holding a ScriptValue here is safe in terms of cross-world wrapper | 243 return; |
| 244 m_hasError = true; |
| 245 if (m_client) |
| 246 notifyLater(); |
| 247 } |
| 248 |
| 249 static void onCollected(const v8::WeakCallbackInfo<ReadableStreamDataConsume
rHandle::ReadingContext>& data) |
| 250 { |
| 251 data.GetParameter()->onCollected(); |
| 252 } |
| 253 |
| 254 // |m_reader| is a weak persistent. It should be kept alive by someone |
| 255 // outside of ReadableStreamDataConsumerHandle. |
| 256 // Holding a ScopedPersistent here is safe in terms of cross-world wrapper |
251 // leakage because we read only Uint8Array chunks from the reader. | 257 // leakage because we read only Uint8Array chunks from the reader. |
252 ScriptValue m_reader; | 258 ScopedPersistent<v8::Value> m_reader; |
| 259 RefPtr<ScriptState> m_scriptState; |
253 WebDataConsumerHandle::Client* m_client; | 260 WebDataConsumerHandle::Client* m_client; |
254 RefPtr<DOMUint8Array> m_pendingBuffer; | 261 RefPtr<DOMUint8Array> m_pendingBuffer; |
255 WeakPtrFactory<ReadingContext> m_weakPtrFactory; | |
256 size_t m_pendingOffset; | 262 size_t m_pendingOffset; |
257 bool m_isReading; | 263 bool m_isReading; |
258 bool m_isDone; | 264 bool m_isDone; |
259 bool m_hasError; | 265 bool m_hasError; |
260 bool m_isInRecursion; | |
261 }; | 266 }; |
262 | 267 |
263 ReadableStreamDataConsumerHandle::ReadableStreamDataConsumerHandle(ScriptState*
scriptState, ScriptValue stream) | 268 ReadableStreamDataConsumerHandle::ReadableStreamDataConsumerHandle(ScriptState*
scriptState, ScriptValue streamReader) |
264 : m_readingContext(ReadingContext::create(scriptState, stream)) | 269 : m_readingContext(ReadingContext::create(scriptState, streamReader)) |
265 { | 270 { |
266 } | 271 } |
267 ReadableStreamDataConsumerHandle::~ReadableStreamDataConsumerHandle() = default; | 272 ReadableStreamDataConsumerHandle::~ReadableStreamDataConsumerHandle() = default; |
268 | 273 |
269 FetchDataConsumerHandle::Reader* ReadableStreamDataConsumerHandle::obtainReaderI
nternal(Client* client) | 274 FetchDataConsumerHandle::Reader* ReadableStreamDataConsumerHandle::obtainReaderI
nternal(Client* client) |
270 { | 275 { |
271 return new ReadingContext::ReaderImpl(m_readingContext, client); | 276 return new ReadingContext::ReaderImpl(m_readingContext, client); |
272 } | 277 } |
273 | 278 |
274 } // namespace blink | 279 } // namespace blink |
275 | 280 |
OLD | NEW |