implement stream subscription