code: public void forEachCachedMessageWhere(Predicate<Message> filter, Consumer<Message> action) {
          synchronized (messages) {
              messages.values().stream()
                      .map(Reference::get)
                      .filter(Objects::nonNull)
                      .filter(filter)
                      .forEach(action);
          }
      }
docs: Execute a task for every message in cache that satisfied a given condition.
  
  @param filter The condition on which to execute the code.
  @param action The action to be applied to the messages.