Allow passing topic into consume method

The consume method should be allowed to consume whatever topic it wants without reinitializing.