Broker/stream refactoring.