Unhandled "error" event

[oliversalzburg]

I sometimes run into the following error:

events.js:160
      throw er; // Unhandled 'error' event
      ^

Error: write ECONNRESET
    at exports._errnoException (util.js:1012:11)
    at Socket._writeGeneric (net.js:705:26)
    at Socket._write (net.js:724:8)
    at doWrite (_stream_writable.js:307:12)
    at writeOrBuffer (_stream_writable.js:293:5)
    at Socket.Writable.write (_stream_writable.js:220:11)
    at Socket.write (net.js:650:40)
    at roundrobin (C:\foo\node_modules\amqplib\lib\mux.js:65:27)
    at Mux._readIncoming (C:\foo\node_modules\amqplib\lib\mux.js:82:3)
    at Immediate.<anonymous> (C:\foo\node_modules\amqplib\lib\mux.js:112:12)
    at runCallback (timers.js:570:20)
    at tryOnImmediate (timers.js:550:5)
    at processImmediate [as _immediateCallback] (timers.js:529:5)

How can I handle this event? Where should I attach the event listener?

[holoduke]

same problem.
if you try to reconnect after manually stopping/starting rabbitmq you ll get this error.

[cressie176]

Have you tried listening for 'error' on the connection object?

[oliversalzburg]

@cressie176 I don't think I have. I added the appropriate code now, but I can't verify it right now, as I'm not in an environment where I could properly verify it.

A more definitive answer would be appreciated :)

[cressie176]

Hi @oliversalzburg. Sorry I'm not being as helpful as you would like. My guess is that it's the amqplib connection which is emitting that error. You can listen for error as follows...

connection.once('error', handleConnectionError)

However what you do when you've encountered a connection error is more complicated. The simplest (but not always acceptable) approach is to gracefully stop and restart your application. If you can't do this then you need to...

1. Reconnect
2. Rebind the error handler to the new connection
3. Create new channels that were associated with the broken connection
4. Consume from the new channels instead of the old ones
5. Publish to the new channels instead of the old ones

As you can see this is quite a complicated process - especially if the channels are referenced from other modules. If you don't fancy doing all this yourself then you might find rascal useful. It does all this for you. At the very least you can grok the code and see the approach I took.

[alextes]

Just ran into this.
Calling channel.checkQueue does return a rejected promise indicating the action failed if the queue doesn't exist, in the process blowing up the channel as the docs warn.

However it also emits an event of type 'error'. Which is a proper thing to do as an event emitter in most cases. In this case I'd say it's probably not as the Node Docs have the following to say about it:

If an EventEmitter does not have at least one listener registered for the 'error' event, and an 'error' event is emitted, the error is thrown, a stack trace is printed, and the Node.js process exits.

In other words before you can do anything useful with your rejection the whole process exits.
Seeing as the event is emitted from the channel I found you don't have to capture on the connection as @cressie176 suggested you can also do:

  channel.once('error', (error) => {
    if (error.code !== 404) {
      throw err;
    }
  });

This way you change as little as possible whilst at the same time catching the only error I think is interesting when using checkQueue because what else could go wrong if the queue name is the only argument you can pass.

If it's helpful to anyone, here's how I now semi-safely execute checkExchange or checkQueue:
Create a new channel. Pass the channel together with a the checkQueue/checkExchange function. Bind the latter to the channel plus the name of your queue or exchange you're looking for (channel.checkQueue.bind(channel, queueName)). You'll get a promise back that resolves to the reply on success, resolves to null on not finding the queue/exchange and rejects in case of any other unexpected error.

function translateErrorToErrorCode(error) {
  if (error.message) {
    const splitString = error.message.split(' ');
    if (splitString.length >= 3) {
      return splitString[3];
    }
  }
  return null;
}

/**
 * Executes one of AMQPs checkX functions and deals with errors returning the
 * reply on success
 * @param {Object} channel - channel that may be closed on error
 * @param {Function} checkFunction - checkQueue or checkExchange function to execute
 * @returns {Promise.<*>}
 */
export async function safeAMQPCheck(channel, checkFunction) {
  // listen to emitted error event for library because it doesn't
  channel.once('error', (error) => {
    // we only expect the check's 404 error
    if (error.code !== 404) {
      throw error;
    }
  });

  let reply;
  try {
    reply = await checkFunction.call(channel);
  } catch (error) {
    // since we only get a useful message, try to split it to get the error code
    const code = translateErrorToErrorCode(error);
    if (code === '404') {
      return Promise.resolve(null);
    }
    // if its not the 404 message something unexpected happened
    return Promise.reject(error);
  }
  // if the channel didn't blow up we should close it
  channel.close();
  return Promise.resolve(reply);
}

[oliversalzburg]

Sorry for the delay, listening for the event on the connection itself, as @cressie176 suggested, is the right thing to do.

Thanks to everyone for their input!

[alextes]

@oliversalzburg Can you please leave this issue open? Judging by the documentation amqp.node definitely does not mean to exit your process when you checkQueue / checkExchange. This is the oldest issue mentioning this problem I could find. It should stay open until the issue is resolved. You can unsubscribe in the lower right if you do not wish to be updated on it anymore.

Second, why is listening on the connection the right thing to do. The check is supposed to 'bork' the channel. If any other channel on the same connection encounters an unexpected error the connection should not simply catch and ignore it wouldn't you say? Just the channel used should catch and ignore any error that comes up when calling checkQueue / checkExchange, and if you're re-using the channel, definitely remove the catch/ignore listener again.

[oliversalzburg]

@alextes I opened this ticket because I wanted to know how to handle this event. This question was answered, thus the closure is the right thing to do.

You can handle this event and re-connect, but you also need to re-establish anything you previously established on the old connection.

I assume that this is something that is out of the scope of this library. IMHO, implementing it in the library itself needlessly complicates the code and it could be more optimally implemented in a library that extends this one. I believe that is what @cressie176 has done with rascal.

If you believe that this behavior should be changed in this library, I would encourage you to open a new ticket and reference this one

As a side note, in our environment it was always beneficial to exit the process when the connection was terminated. Our orchestration would simply restart the container until the connection would be successfully re-established. This was much simpler than trying to handle the reconnection logic within the same process. We were simply looking for a way to clearly log the reason for the process termination before exiting.

[alextes]

@oliversalzburg thanks for the link to rascal! I might switch to that.

If you're convinced the library means to blow up your process by default when checking if a queue is there and finding it is not (without documenting this drastic behaviour), by all means keep this closed, and I'll sub to #282.