Can't you implement that using CoGroup?