using pipes for activity stream processing