Cleanup Pub/Sub functions to create test *Connection classes

[coryan]

To test the *Connection classes we have special functions in pubsub_internal:: that decorate a *Mock stub and then create the *Connection, for example:

google-cloud-cpp/google/cloud/pubsub/publisher_connection.h

Lines 167 to 169 in 1e29dce

	std::shared_ptr<pubsub::PublisherConnection> MakePublisherConnection(
	pubsub::Topic topic, Options opts,
	std::vector<std::shared_ptr<PublisherStub>> stubs);

Their implementation can be quite involved:

google-cloud-cpp/google/cloud/pubsub/publisher_connection.cc

Lines 112 to 147 in 1e29dce

	std::shared_ptr<pubsub::PublisherConnection> MakePublisherConnection(
	pubsub::Topic topic, Options opts,
	std::vector<std::shared_ptr<PublisherStub>> stubs) {
	if (stubs.empty()) return nullptr;
	std::shared_ptr<PublisherStub> stub =
	std::make_shared<PublisherRoundRobin>(std::move(stubs));
	stub = std::make_shared<PublisherMetadata>(std::move(stub));
	if (internal::Contains(opts.get<TracingComponentsOption>(), "rpc")) {
	GCP_LOG(INFO) << "Enabled logging for gRPC calls";
	stub = std::make_shared<PublisherLogging>(
	std::move(stub), opts.get<GrpcTracingOptionsOption>());
	}
	auto background = internal::MakeBackgroundThreadsFactory(opts)();
	auto make_connection = [&]() -> std::shared_ptr<pubsub::PublisherConnection> {
	auto cq = background->cq();
	std::shared_ptr<BatchSink> sink = DefaultBatchSink::Create(stub, cq, opts);
	if (opts.get<pubsub::MessageOrderingOption>()) {
	auto factory = [topic, opts, sink, cq](std::string const& key) {
	return BatchingPublisherConnection::Create(
	topic, opts, key, SequentialBatchSink::Create(sink), cq);
	};
	return OrderingKeyPublisherConnection::Create(std::move(factory));
	}
	return RejectsWithOrderingKey::Create(BatchingPublisherConnection::Create(
	std::move(topic), opts, {}, std::move(sink), std::move(cq)));
	};
	auto connection = make_connection();
	if (opts.get<pubsub::FullPublisherActionOption>() !=
	pubsub::FullPublisherAction::kIgnored) {
	connection = FlowControlledPublisherConnection::Create(
	std::move(opts), std::move(connection));
	}
	return std::make_shared<pubsub::ContainingPublisherConnection>(
	std::move(background), std::move(connection));
	}

Unfortunately, this breaks down with the introduction of the *Auth decorators, where the channel is created by the GrpcAuthenticationStrategy, then we create the default *Stub and then we wrap said stub with the *Auth decorator and we need to preserve the BackgroundThreads object to be owned by the connection.

We should reorganize, and maybe rename, these functions to reflect the new structure.

[dbolduc]

Update: the below comment is no longer valid

I will just point out that this is slightly different in classes with >1 stub (Publisher, Subscriber) than those with only 1 (TopicAdmin, SchemaAdmin, SubscriptionAdmin). I deleted the test-only methods in classes with >1 stub (for example: here). Maybe I shouldn't have, but I did.

MakePublisherConnection implementation shown above is an implementation used by our user-facing methods and our test method.

Whereas we have two separate MakeSchemaAdminConnection methods that do the auth wrapping. One user-facing. One for testing.