can consume redis messages now