Possible error in broker.emit(). I cannot recognize that an event has been sent to the transport successfully. #1105
Comments
Vslava
changed the title
Vslava
changed the title
|
Hi @AndreMaz It seems no. They tell that the local emit request is waiting when the code is completed. I tell that I don't get any information about the emit has a problem while it is running. For example, if the transport get down and after that I emit an event, I will get that the operation has been resolved successfully, but I see in the log that actually the event hasn't been sent to the transport and I cannot react to the problem, for example, to resend the event again. For example, I have two files. I use RabbitMQ as a transport. --- reciver.js --- const { Context, ServiceBroker } = require('moleculer');
const broker = new ServiceBroker({
nodeID: 'node-1',
transporter: 'amqp://rabbitmq:5672',
logLevel: 'debug',
serializer: 'CBOR',
});
broker.createService({
name: 'test',
events: {
'created': {
handler(ctx) {
console.log('---- event is got');
}
},
}
});
broker.start()
.catch((err) => console.error('!!! Error 1 !!!', err));--- sender.js --- const util = require('util');
const { ServiceBroker } = require('moleculer');
const setTimeoutProm = util.promisify(setTimeout);
const broker = new ServiceBroker({
nodeID: 'node-2',
transporter: 'amqp://rabbitmq:5672',
logLevel: 'debug',
serializer: 'CBOR',
});
broker.start()
.then(async () => {
while (true) {
broker.emit('created', { aaa: 111 })
.then(() => { console.log('----- resolved'); })
.catch((err) => { console.log('----- error', err); })
await setTimeoutProm(2000);
}
})
.catch((err) => console.error('!!! Error 2 !!!', err));I started them and after that I shut down RabbitMQ. As we can see, the emit got failed but it returned that it had executed successfully because the promise was resolved, not rejected. I think that the emit result promise must be rejected in such situation. |
|
I think that this need further discussion. The overall idea is that events are fire-and-forget, i.e., you don't get any guarantees about the delivery of the event.
Even if the event is sent again (and it reaches the transporter) there's still no guarantee that it will be delivered to the Nevertheless, there are some internal events (e.g., Overall, for guaranteed event delivery https://github.com/moleculerjs/moleculer-channels might be a good solution. It was designed to provide a "secure" and persistent event system. In any case, @Vslava @Embraser01, since this issue and the #1065 are related I would like to hear your ideas about this. What should be the behavior for the local and remote |
|
@Vslava the fact is that you need to separate two different systems. |
You're right @intech it's a different issue but I think that In the case of local |
I agree that there’s no guarantee that an event will be delivered because you don’t know that somebody is listening to the event or not. The event sender cannot know who is subscribed for the event. However, a moleculer user has to be sure that an event will reach the transporter and be accepted for delivery and the event will be accepted by the consumer successfully but you don’t guarantee these all. Now, if the transporter has a problem, the sender cannot recognize that this problem exists. broker.emit() doesn’t return any problem which it has bumped into. Such implementation doesn’t give the user any possibility to create a simple subsystem to resend events. Yes, Moleculer doesn’t have a built-in storage but it can give a possibility to the user to implement this storage on their side. You can reject the promise returned by broker.emit() if there is some problem when the event is being transmitted to the transporter. If the sender knows about the problem, the sender can react on the problem and resend the event. As I understood, Moleculer doesn’t guarantee that the event will be accepted by the consumer successfully. If the event starts sending the event, the transporter forgets about the event. There is’t any ACK implementation which the cosumer can send to the transporter to confirm that the event has been accepted successfully. I think that not all transaporters have this feature but it can be implemented where possible. About $transit.error, $transporter.error. How can I understand that an error was caused by the event I sent? Maybe my event was sent successfully and after that the error was happened. Especially It is important when the error is happend in the same time when I am sending the event. There is nothing inside the error except a message and a code. Maybe it is simpler to throw an exception from broker.emit() ? About https://github.com/moleculerjs/moleculer-channels. Yes, it can be good choice but now It has a bug with RabbitMQ. It doesn’t recover connection with RabbitMQ when RabbitMQ got down and started again. I will make an issue about this. So It is not choice just now. And… with all respect but it is strange that Moleculer is a microservices framework and it has own feature for events but I need to use another module to use events instead of using native events out of box. |
|
@Vslava as I see, your main problem is that in the |
icebob
added
Status: Proposal
Module: Transporter
Module: Core
and removed
Module: Transporter
labels
|
As @icebob suggested, adding a However, introducing this could be a breaking change. I'm saying could because I saw several times situation like await ctx.emit("some.event", {data}) // <--- No catchThrowing an error without the The workaround for you would be to overwrite the
I think that this statement is a little unfair. We simply can't compete against project like NATS, Kafka, RabitMQ, etc. They have big budgets and lots of people dedicated to design a message delivery solution. I want to highlight, the sole purpose of those projects is reliable delivery and nothing more. Moleculer is a little bit different as it provides more features. If we were to implement a custom (and reliable) event system, then the discussion would be different. We would be discussing the performance of our solution against NATS, Kafka, etc. Again, we can't compete with those projects without having the same resources as they have. To summarize, I think that our decision of creating moleculer/channels was correct. Instead of competing with the previously mentioned solutions we provide a common and easy-to-use interface to interact with them, alongside with all the benefits that they provide. |
|
@icebob Yes. You are right. |
If I invoke
broker.emit(), I will never recognize that the event was sent to the transport successfully. The problem is that you returnthis.transit.sendEvent(ctx)orthis.Promise.all(promises), where promises are the results ofthis.transit.sendEvent(newCtx), inServiceBroker.emit(). InsideTransit.sendEvent(ctx), you returnIn such situation, the returned promise will never be rejected if the transporter exception has raised. I think it needs to save the result promise of publish method in a variable and return it, and apply catch to this promise too. Like
It gives a possibility to recognize a problem with transport.
The text was updated successfully, but these errors were encountered: