client: move subscription to a separate function
This allows overloading the subscribe() function, e.g for manual assignment of partitions.