@@ -175,17 +175,37 @@ int StreamBase::Shutdown(const FunctionCallbackInfo<Value>& args) {
175175void StreamBase::SetWriteResult (const StreamWriteResult& res) {
176176 env_->stream_base_state ()[kBytesWritten ] = res.bytes ;
177177 env_->stream_base_state ()[kLastWriteWasAsync ] = res.async ;
178+ env_->stream_base_state ()[kLastWriteErr ] = res.err ;
179+ }
180+
181+ // Finish a JS-initiated write. When the caller did not pass a request
182+ // object (`lazy_req`), the write wrap object exists only if the write did
183+ // not complete synchronously; hand it to JS so that it can attach its
184+ // completion callback. Otherwise the numeric error code (via JSMethod) is
185+ // all JS needs.
186+ int StreamBase::FinishWrite (const v8::FunctionCallbackInfo<v8::Value>& args,
187+ const StreamWriteResult& res,
188+ bool lazy_req) {
189+ SetWriteResult (res);
190+ if (lazy_req && res.wrap_obj ) {
191+ args.GetReturnValue ().Set (res.wrap_obj ->object ());
192+ return kReturnValueSet ;
193+ }
194+ return res.err ;
178195}
179196
180197int StreamBase::Writev (const FunctionCallbackInfo<Value>& args) {
181198 Environment* env = Environment::GetCurrent (args);
182199 Isolate* isolate = env->isolate ();
183200 Local<Context> context = env->context ();
184201
185- CHECK (args[0 ]->IsObject ());
186202 CHECK (args[1 ]->IsArray ());
187203
188- Local<Object> req_wrap_obj = args[0 ].As <Object>();
204+ // When no request object is passed in, one is created by Write() only if
205+ // the write does not complete synchronously; see FinishWrite().
206+ const bool lazy_req = !args[0 ]->IsObject ();
207+ Local<Object> req_wrap_obj;
208+ if (!lazy_req) req_wrap_obj = args[0 ].As <Object>();
189209 Local<Array> chunks = args[1 ].As <Array>();
190210 bool all_buffers = args[2 ]->IsTrue ();
191211
@@ -287,20 +307,19 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
287307 }
288308
289309 StreamWriteResult res = Write (*bufs, count, nullptr , req_wrap_obj);
290- SetWriteResult (res);
291310 if (res.wrap != nullptr && storage_size > 0 )
292311 res.wrap ->SetBackingStore (std::move (bs));
293- return res. err ;
312+ return FinishWrite (args, res, lazy_req) ;
294313}
295314
296-
297315int StreamBase::WriteBuffer (const FunctionCallbackInfo<Value>& args) {
298- CHECK (args[0 ]->IsObject ());
299316 CHECK (args[1 ]->IsUint8Array ());
300317
301318 Environment* env = Environment::GetCurrent (args);
302319
303- Local<Object> req_wrap_obj = args[0 ].As <Object>();
320+ bool lazy_req = !args[0 ]->IsObject ();
321+ Local<Object> req_wrap_obj;
322+ if (!lazy_req) req_wrap_obj = args[0 ].As <Object>();
304323 uv_buf_t buf;
305324 buf.base = Buffer::Data (args[1 ]);
306325 buf.len = Buffer::Length (args[1 ]);
@@ -310,6 +329,16 @@ int StreamBase::WriteBuffer(const FunctionCallbackInfo<Value>& args) {
310329 if (args[2 ]->IsObject () && IsIPCPipe ()) {
311330 Local<Object> send_handle_obj = args[2 ].As <Object>();
312331
332+ if (lazy_req) {
333+ // Sending a handle requires a request object up front to reference it.
334+ if (!env->write_wrap_template ()
335+ ->NewInstance (env->context ())
336+ .ToLocal (&req_wrap_obj)) {
337+ return UV_EBUSY ;
338+ }
339+ StreamReq::ResetObject (req_wrap_obj);
340+ }
341+
313342 HandleWrap* wrap;
314343 ASSIGN_OR_RETURN_UNWRAP (&wrap, send_handle_obj, UV_EINVAL );
315344 send_handle = reinterpret_cast <uv_stream_t *>(wrap->GetHandle ());
@@ -323,20 +352,18 @@ int StreamBase::WriteBuffer(const FunctionCallbackInfo<Value>& args) {
323352 }
324353
325354 StreamWriteResult res = Write (&buf, 1 , send_handle, req_wrap_obj);
326- SetWriteResult (res);
327-
328- return res.err ;
355+ return FinishWrite (args, res, lazy_req);
329356}
330357
331-
332358template <enum encoding enc>
333359int StreamBase::WriteString (const FunctionCallbackInfo<Value>& args) {
334360 Environment* env = Environment::GetCurrent (args);
335361 Isolate* isolate = env->isolate ();
336- CHECK (args[0 ]->IsObject ());
337362 CHECK (args[1 ]->IsString ());
338363
339- Local<Object> req_wrap_obj = args[0 ].As <Object>();
364+ const bool lazy_req = !args[0 ]->IsObject ();
365+ Local<Object> req_wrap_obj;
366+ if (!lazy_req) req_wrap_obj = args[0 ].As <Object>();
340367 Local<String> string = args[1 ].As <String>();
341368 Local<Object> send_handle_obj;
342369 if (args[2 ]->IsObject ())
@@ -417,6 +444,16 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
417444 uv_stream_t * send_handle = nullptr ;
418445
419446 if (IsIPCPipe () && !send_handle_obj.IsEmpty ()) {
447+ if (lazy_req && req_wrap_obj.IsEmpty ()) {
448+ // Sending a handle requires a request object up front to reference it.
449+ if (!env->write_wrap_template ()
450+ ->NewInstance (env->context ())
451+ .ToLocal (&req_wrap_obj)) {
452+ return UV_EBUSY ;
453+ }
454+ StreamReq::ResetObject (req_wrap_obj);
455+ }
456+
420457 HandleWrap* wrap;
421458 ASSIGN_OR_RETURN_UNWRAP (&wrap, send_handle_obj, UV_EINVAL );
422459 send_handle = reinterpret_cast <uv_stream_t *>(wrap->GetHandle ());
@@ -432,11 +469,10 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
432469 StreamWriteResult res = Write (&buf, 1 , send_handle, req_wrap_obj, try_write);
433470 res.bytes += synchronously_written;
434471
435- SetWriteResult (res);
436472 if (res.wrap != nullptr )
437473 res.wrap ->SetBackingStore (std::move (bs));
438474
439- return res. err ;
475+ return FinishWrite (args, res, lazy_req) ;
440476}
441477
442478
@@ -662,7 +698,8 @@ void StreamBase::JSMethod(const FunctionCallbackInfo<Value>& args) {
662698 if (!wrap->IsAlive ()) return args.GetReturnValue ().Set (UV_EINVAL );
663699
664700 AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope (wrap->GetAsyncWrap ());
665- args.GetReturnValue ().Set ((wrap->*Method)(args));
701+ int ret = (wrap->*Method)(args);
702+ if (ret != kReturnValueSet ) args.GetReturnValue ().Set (ret);
666703}
667704
668705int StreamResource::DoTryWrite (uv_buf_t ** bufs, size_t * count) {
@@ -771,20 +808,50 @@ void ReportWritesToJSStreamListener::OnStreamAfterReqFinished(
771808 CHECK (!async_wrap->persistent ().IsEmpty ());
772809 Local<Object> req_wrap_obj = async_wrap->object ();
773810
811+ Local<Value> oncomplete;
812+ if (!req_wrap_obj->Get (env->context (), env->oncomplete_string ())
813+ .ToLocal (&oncomplete)) {
814+ return ;
815+ }
816+
817+ const char * msg = stream->Error ();
818+
819+ if (!oncomplete->IsFunction ()) {
820+ // The completion callback has not been attached yet: the write finished
821+ // synchronously inside the JS write call, before the request object was
822+ // returned to JS. Record the status so that JS can replay the callback.
823+ if (req_wrap_obj
824+ ->Set (env->context (),
825+ env->write_status_string (),
826+ Integer::New (env->isolate (), status))
827+ .IsNothing ()) {
828+ return ;
829+ }
830+ if (msg != nullptr ) {
831+ if (req_wrap_obj
832+ ->Set (env->context (),
833+ env->error_string (),
834+ OneByteString (env->isolate (), msg))
835+ .IsNothing ()) {
836+ return ;
837+ }
838+ stream->ClearError ();
839+ }
840+ return ;
841+ }
842+
774843 Local<Value> argv[] = {
775844 Integer::New (env->isolate (), status),
776845 stream->GetObject (),
777846 Undefined (env->isolate ())
778847 };
779848
780- const char * msg = stream->Error ();
781849 if (msg != nullptr ) {
782850 argv[2 ] = OneByteString (env->isolate (), msg);
783851 stream->ClearError ();
784852 }
785853
786- if (req_wrap_obj->Has (env->context (), env->oncomplete_string ()).FromJust ())
787- async_wrap->MakeCallback (env->oncomplete_string (), arraysize (argv), argv);
854+ async_wrap->MakeCallback (oncomplete.As <Function>(), arraysize (argv), argv);
788855}
789856
790857void ReportWritesToJSStreamListener::OnStreamAfterWrite (
0 commit comments