Before consuming a message, make sure database connection is valid

The previous implementation would fail to process a message if the
database connection was lost.  Unfortunately it looked like the message
was consumed and the data was lost.