public void forEachCachedMessageWhere ( Predicate < Message > filter , Consumer < Message > action ) { synchronized ( messages ) { messages . values ( ) . stream ( ) . map ( Reference :: get ) . filter ( Objects :: nonNull ) . filter ( filter ) . forEach ( action ) ; } }
Execute a task for every message in cache that satisfied a given condition .