generic observable producer