fluencelabs/fluence-js
1
0
Fork 0
mirror of https://github.com/fluencelabs/fluence-js.git synced 2024-12-05 02:10:18 +00:00
Code Issues Projects Releases Wiki Activity

fix subscribtions

Browse Source
This commit is contained in:
Pavel Murygin 2021-12-10 16:07:46 +03:00
parent 157d6309c1
commit ba76e4919d
1 changed files with 75 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
src/internal/FluencePeer.ts
View File

@ -461,87 +461,92 @@ export class FluencePeer {
}),
);
interpreted.pipe(
const successful = interpreted.pipe(
// force new line
filter((x) => x.isInterpretationSuccessful),
);
const failed = interpreted.pipe(
// force new line
filter((x) => !x.isInterpretationSuccessful),
tap((item) =>
setTimeout(() => {
item.onStageChange({
stage: 'interpreterError',
errorMessage: item.interpreterResult.errorMessage,
});
return;
}, 0),
),
);
interpreted.pipe(
filter((x) => x.isInterpretationSuccessful),
tap((item) =>
setTimeout(() => {
item.onStageChange({ stage: 'interpreted' });
}, 0),
),
failed.subscribe((item) =>
setTimeout(() => {
item.onStageChange({
stage: 'interpreterError',
errorMessage: item.interpreterResult.errorMessage,
});
return;
}, 0),
);
interpreted.pipe(
filter((x) => x.isInterpretationSuccessful),
filter((x) => x.hasPeerPks),
tap((item) => {
successful.subscribe((item) =>
setTimeout(() => {
item.onStageChange({ stage: 'interpreted' });
}, 0),
);
successful
.pipe(
// force new line
filter((x) => x.hasPeerPks),
)
.subscribe((item) => {
const newParticle = item.particle.clone();
newParticle.data = item.newData;
this._outgoingParticles.next({ ...item, particle: newParticle });
}),
);
});
interpreted.pipe(
filter((x) => x.isInterpretationSuccessful),
filter((x) => !x.hasCallRequests),
tap((item) => {
item.onStageChange({ stage: 'localWorkDone' });
}),
);
interpreted
successful
.pipe(
filter((x) => x.isInterpretationSuccessful),
filter((x) => x.hasCallRequests),
concatMap((item) => item.interpreterResult.callRequests.map(([key, cr]) => [item, key, cr] as const)),
map(([item, key, cr]) => {
const req = {
fnName: cr.functionName,
args: cr.arguments,
serviceId: cr.serviceId,
tetraplets: cr.tetraplets,
particleContext: item.particle.getParticleContext(),
};
return [item, key, req] as const;
}),
concatMap(([item, key, req]) => {
return this._execSingleCallRequest(req)
.catch(
(err): CallServiceResult => ({
retCode: ResultCodes.exceptionInHandler,
result: `Handler failed. fnName="${req.fnName}" serviceId="${
req.serviceId
}" error: ${err.toString()}`,
}),
)
.then((res) => [item, key, res] as const);
}),
map(([item, key, res]) => {
const serviceResult = {
result: w(res.result),
retCode: res.retCode,
};
const newParticle = item.particle.clone();
newParticle.callResults = [[key, serviceResult]];
newParticle.data = Buffer.from([]);
return { particle: newParticle, onStageChange: item.onStageChange };
}),
// force new line
filter((x) => !x.hasCallRequests),
)
.subscribe((item) => particlesQueue.next(item));
.subscribe((item) => {
item.onStageChange({ stage: 'localWorkDone' });
});
const readyToSend = successful.pipe(
filter((x) => x.hasCallRequests),
concatMap((item) => item.interpreterResult.callRequests.map(([key, cr]) => [item, key, cr] as const)),
map(([item, key, cr]) => {
const req = {
fnName: cr.functionName,
args: cr.arguments,
serviceId: cr.serviceId,
tetraplets: cr.tetraplets,
particleContext: item.particle.getParticleContext(),
};
return [item, key, req] as const;
}),
concatMap(([item, key, req]) => {
return this._execSingleCallRequest(req)
.catch(
(err): CallServiceResult => ({
retCode: ResultCodes.exceptionInHandler,
result: `Handler failed. fnName="${req.fnName}" serviceId="${
req.serviceId
}" error: ${err.toString()}`,
}),
)
.then((res) => [item, key, res] as const);
}),
map(([item, key, res]) => {
const serviceResult = {
result: w(res.result),
retCode: res.retCode,
};
const newParticle = item.particle.clone();
newParticle.callResults = [[key, serviceResult]];
newParticle.data = Buffer.from([]);
return { particle: newParticle, onStageChange: item.onStageChange };
}),
);
readyToSend.subscribe((item) => particlesQueue.next(item));
return particlesQueue;
}