flatmap can now be used on publish messages, reimplemented hofs for pipe