code: function readNewMessages(state) {
    // channel already closed
    if (state.closed) return Promise.resolve(); // if no one is listening, we do not need to scan for new messages
  
    if (!state.messagesCallback) return Promise.resolve();
    return getMessagesHigherThen(state.db, state.lastCursorId).then(function (newerMessages) {
      var useMessages = newerMessages.map(function (msgObj) {
        if (msgObj.id > state.lastCursorId) {
          state.lastCursorId = msgObj.id;
        }
  
        return msgObj;
      }).filter(function (msgObj) {
        return _filterMessage(msgObj, state);
      }).sort(function (msgObjA, msgObjB) {
        return msgObjA.time - msgObjB.time;
      }); // sort by time
  
      useMessages.forEach(function (msgObj) {
        if (state.messagesCallback) {
          state.eMIs.add(msgObj.id);
          state.messagesCallback(msgObj.data);
        }
      });
      return Promise.resolve();
    });
  }
docs: reads all new messages from the database and emits them