tap implementation in Rx