github/meteor
1
1
Fork 0
mirror of https://github.com/meteor/meteor.git synced 2026-05-02 03:01:46 -04:00
Code Issues Packages Projects Releases Wiki Activity

using Meteor._runAsync to wrap _runHandler

Browse Source
This commit is contained in:
denihs
2022-11-29 18:22:20 -04:00
parent 2238df20ac
commit db03c7f6a2
4 changed files with 63 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
packages/ddp-server-async/livedata_server.js
View File

@@ -1106,52 +1106,60 @@ var Subscription = function (
Object.assign(Subscription.prototype, {
_runHandler: function() {
// XXX should we unblock() here? Either before running the publish
// function, or before running _publishCursor.
//
// Right now, each publish function blocks all future publishes and
// methods waiting on data from Mongo (or whatever else the function
// blocks on). This probably slows page load in common cases.
Meteor._runAsync(
() => {
// XXX should we unblock() here? Either before running the publish
// function, or before running _publishCursor.
//
// Right now, each publish function blocks all future publishes and
// methods waiting on data from Mongo (or whatever else the function
// blocks on). This probably slows page load in common cases.
if (!this.unblock) {
this.unblock = () => {};
}
if (!this.unblock) {
this.unblock = () => {};
}
const self = this;
let resultOrThenable = null;
try {
resultOrThenable = DDP._CurrentPublicationInvocation.withValue(self, () =>
maybeAuditArgumentChecks(
self._handler,
self,
EJSON.clone(self._params),
// It's OK that this would look weird for universal subscriptions,
// because they have no arguments so there can never be an
// audit-argument-checks failure.
"publisher '" + self._name + "'"
)
);
} catch (e) {
self.error(e);
return;
}
const self = this;
let resultOrThenable = null;
try {
resultOrThenable = DDP._CurrentPublicationInvocation.withValue(
self,
() =>
maybeAuditArgumentChecks(
self._handler,
self,
EJSON.clone(self._params),
// It's OK that this would look weird for universal subscriptions,
// because they have no arguments so there can never be an
// audit-argument-checks failure.
"publisher '" + self._name + "'"
)
);
} catch (e) {
self.error(e);
return;
}
// Did the handler call this.error or this.stop?
if (self._isDeactivated()) return;
// Did the handler call this.error or this.stop?
if (self._isDeactivated()) return;
// Both conventional and async publish handler functions are supported.
// If an object is returned with a then() function, it is either a promise
// or thenable and will be resolved asynchronously.
const isThenable =
resultOrThenable && typeof resultOrThenable.then === 'function';
if (isThenable) {
Promise.resolve(resultOrThenable).then(
(...args) => self._publishHandlerResult.bind(self)(...args),
e => self.error(e)
);
} else {
self._publishHandlerResult(resultOrThenable);
}
// Both conventional and async publish handler functions are supported.
// If an object is returned with a then() function, it is either a promise
// or thenable and will be resolved asynchronously.
const isThenable =
resultOrThenable && typeof resultOrThenable.then === 'function';
if (isThenable) {
Promise.resolve(resultOrThenable).then(
(...args) => self._publishHandlerResult.bind(self)(...args),
e => self.error(e)
);
} else {
self._publishHandlerResult(resultOrThenable);
}
},
this,
{ callId: '_runHandler' }
);
},
_publishHandlerResult: function (res) {

5
packages/ddp-server-async/livedata_server_async_tests.js
View File

@@ -35,11 +35,6 @@ Meteor.publish('livedata_server_test_sub_context_async', async function(
this.onStop(function() {
var onStopMethodInvocation = DDP._CurrentMethodInvocation.get();
var onStopPublicationInvocation = DDP._CurrentPublicationInvocation.get();
// console.log('onStopMethodInvocation', onStopMethodInvocation);
console.log('onStopPublicationInvocation', !!onStopPublicationInvocation, this.userId);
callback.call(
this,

4
packages/ddp-server-async/livedata_server_tests.js
View File

@@ -179,7 +179,7 @@ Meteor.publish("livedata_server_test_sub_method", async function (connectionId)
this.stop();
});
Meteor.publish("livedata_server_test_sub_context", function (connectionId, userId) {
Meteor.publish("livedata_server_test_sub_context", async function (connectionId, userId) {
var callback = onSubscription[connectionId];
var methodInvocation = DDP._CurrentMethodInvocation.get();
var publicationInvocation = DDP._CurrentPublicationInvocation.get();
@@ -201,7 +201,7 @@ Meteor.publish("livedata_server_test_sub_context", function (connectionId, userI
this.stop();
} else {
this.ready();
Meteor.call('livedata_server_test_setuserid', userId);
await Meteor.callAsync('livedata_server_test_setuserid', userId);
}
});

19
packages/meteor/asl-helpers.js
View File

@@ -7,18 +7,21 @@ Meteor._getAslStore = getAslStore;
Meteor._getValueFromAslStore = getValueFromAslStore;
Meteor._updateAslStore = updateAslStore;
Meteor._runAsync = (fn, ctx) => {
Meteor._runAsync = (fn, ctx, store = {}) => {
if (Meteor._isFibersEnabled) {
const Fiber = Npm.require('fibers');
const Fiber = Npm.require('fibers');
return Fiber(() => {
fn.call(ctx);
}).run();
return Fiber(() => {
fn.call(ctx);
}).run();
}
return global.asyncLocalStorage.run(Meteor._getAslStore(), () => {
fn.call(ctx);
});
return global.asyncLocalStorage.run(
{ ...Meteor._getAslStore(), ...store },
() => {
return fn.call(ctx);
}
);
};
Meteor._isPromise = (r) => {