function onread(err, messages) {
    // If there was an error we still want to emit it.
    // Essentially, if the user does not register an error
    // handler, it will still cause the stream to blow up.
    //
    // But... if one is provided, consumption will move on
    // as normal
    if (err) {
      self.emit('error', err);
    }

    // If there are no messages it means we reached EOF or a timeout.
    // Do what we used to do

    if (err || messages.length < 1) {
      // If we got an error or if there were no messages, initiate a retry
      retry();
      return;
    } else {
      if (self.streamAsBatch) {
        self.push(messages);
      } else {
        for (var i = 0; i < messages.length; i++) {
          self.messages.push(messages[i].value);
        }
        // Now that we have added them all the inner messages buffer,
        // we can just push the most recent one
        self.push(self.messages.shift());
      }
    }

  }
Internal stream read method. This method reads message buffers.
@param {number} size - This parameter is ignored for our cases.
@private