Automatically create topic / subscription if not found.

[theacodes]

Presently I have some code to publish to a particular topic, creating it if necessary:

function publish(message, cb){
  topic.publish(message, function(err){
    if(err && err.code == 404){
      logging.info("Creating pub/sub topic.");
      pubsub.createTopic(topicName, function(err, _){
        if(err) return cb(err);
        publish(message, cb);
      });
    } else {
      cb(err);
    }
  });
}

And similar code to listen for messages to a topic. It goes the other way around - trying to create one at first then re-using if it's already there.

function subscribe(cb){
  topic.subscribe(subscriptionName, function(err, subscription){
    if(!err) return wireSubscription(subscription);
    if(err && err.code == 409) return wireSubscription(topic.subscription(subscriptionName));
    throw err;
  });
  ...
}

According to this comment, it seems that the intended usage is to create the topic and subscription outside of the application content and never touch them afterwards. It's not clear in the documentation that this is expected.

I think it would be great to either:

1. Consolidate subscribe/subscription createTopic/topic into combined create/reuse functions (e.g. .topic() .subscription()). There can be an optional argument for erring instead of re-using. This hides all of this somewhat ugly logic from users that do want this use-case.
2. Explicitly document that topics and subscriptions should be created beforehand.

I would personally prefer the former as it removes the need to write a "bootstrapping" script and simplifies the user-facing API.

[stephenplusplus]

Do you mean a Topic and Subscription object should create itself if necessary after you attempt to use it? I'm ok with this idea, but what's the trouble with manually creating the topic/sub before using it?

[theacodes]

Yes. The trouble with manually creating it before use leads to having to write unnecessary code to handle creating it if necessary- unless you meant creating it with a bootstrap script beforehand outside of the application context. In that case, it's just inconvenient for no real benefit. I could see myself forgetting to create new subscriptions/topics while iterating new features.

[ryanseys]

The approach we have taken for every object in this library is that you can create one or reference an existing one. If we changed this, it would need to be changed everywhere and would require two API requests rather than one, causing increased delays for responses.

Now actually, the latest pubsub v1beta2 (in this library master branch) call to create a topic/subscription is a PUT call meaning its idempotent (basically no matter how many times you call it, it won't cause more effects than if you call it just once). This means that you can effectively "create or get" by always creating the topic regardless of whether it exists or not. If we know we don't need to create it, it would be silly to try to every time we want to do something with it, this would multiply every request time by a factor of two and provide less stability overall.

Hope this makes sense. Let me know if there's a smart compromise that could be made here.

[theacodes]

Hmm. I definitely don't want to make a recommendation that would go against the grain of the library as a whole.

Instead of trying to create every time, couldn't the "use existing" method create if the response is a 404? That should only add 2 additional RPCs (create, then re-submit whatever request originally gave the 404) and only once. Similar to the publish example in my first comment. This wouldn't change the surface of the API but could really help the use case I'm designing for. We could even choose to have it explicit:

var topic = pubsub.topic('my-topic', {autoCreate: true});
topic.publish(message);  // will catch 404, create topic, and then publish the message.

var topic2 = pubsub.topic('my-topic-2');
topic2.publish(message); // will behave as today.

[ryanseys]

I like this idea.

[stephenplusplus]

There might be some useful information to abstract from this pr when I had basically this exact behavior! #107

[theacodes]

Interesting conversation there, and @rakyll seems to be pretty hardline on avoiding "magic". Two thoughts:

1. I agree with not having methods that will potentially delete and recreate resources.
2. I think the explicit form {autoCreate: true} avoids "magic".

[theacodes]

Here's another example to further illustrate the need for this. I followed the "encouraged" flow by writing a one-off script to create pub/sub topics and subscriptions ahead of time:

if (!module.parent) {
  console.log(
    "Creating pub/sub topics and subscriptions on project %s.",
    config.gcloud.projectId);

  pubsub.createTopic(topicName, function(err, topic){
    if(err && err.code == 409 && err.errors && err.errors[0].reason == "alreadyExists") {
      console.log("Topic %s already exists.", topicName);
      topic = pubsub.topic(topicName);
    } else if(err || !topic) {
      throw err;
    } else {
      console.log('Created topic %s', topicName);
    }

    topic.subscribe(subscriptionName, function(err, subscription){
      if(err && err.code == 409 && err.errors && err.errors[0].reason == "alreadyExists") {
        console.log("Subscription %s already exists.", subscriptionName);
      } else if(err) {
        throw err;
      } else{
        console.log('Created subscription %s', subscriptionName);
      }
      console.log('Done');
      process.exit();
    });
  });
}

Definitely seems a bit unwieldy for just one topic & subscription. Of course, if there were a gcloud pubsub topics create ... command this could be workable, but as it stands I think this is poor developer experience.

[tmatsuo]

I have a little concern.

With the current version of Cloud Pub/Sub, a topic without any subscriptions will throw away your messages until the first subscription will be created and attached to it.

So if we auto create on publish, then the messages subsequently published to this topic will go /dev/null, the situation will continue until you create the first subscription.

[tmatsuo]

Also, hopefully we will start providing gcloud pubsub subcommand, as well as Developer Console UI, so the first time burden will be mitigated.

[theacodes]

I personally think it's a better to document that a topic with no subscriptions doesn't deliver messages and add the autoCreate option than to keep the developer experience we have now. The experience we have now is quite painful and frustrating.

Do we have any idea when gcloud pubsub will land? I think we should weigh the effort of adding this vs the time until gcloud pubsub is available.

[ryanseys]

We welcome contributions too so feel free to mock something up for this. I'm a fan of the autoCreate flag to reduce magic. We have no currently planned release date for the next version unfortunately. We're awaiting library reviews from various teams at Google. I hope to have an update for you soon!

[tmatsuo]

Do we have any idea when gcloud pubsub will land?

Not so far away. For a time being, you may find it useful to use my command line tool:
https://github.com/GoogleCloudPlatform/cloud-pubsub-samples-python
(cmdline-pull)

Although it still doesn't support push subscription, it can be used for other operations (create, list, publish, pull, etc).

[theacodes]

@tmatsuo this is helpful, but for the context of incorporating pub/sub into the end-to-end node on cloud platform sample app, it would be strange to ask the developers to use a python tool.

I'll try to mock something up this week. This actually shouldn't be too complex to incorporate.

[tmatsuo]

Got it, however, I still don't recommend that you spend a lot of time implementing it. It might become more complex than you think, if you want it to be perfect. For example, what happens if creation fails? Also, in the worst case, what happens if the creation fails, but the subsequent API calls return 404 error (it's extremely unlikely, but it can still happen e.g. bad config push etc).

So instead, how about to use the API explorer for a time being?

[tmatsuo]

I meant to say, if the creation suceeds, but..., in the last example.

[stephenplusplus]

I think we've covered this in #465, which was merged to master.

[gustawdaniel-acaisoft]

I would like to share code to solve this problem.

import 'dotenv/config';
import {PubSub} from '@google-cloud/pubsub';

process.env.PUB_SUB_PREFIX = process.env.PUB_SUB_PREFIX || 'local';

process.env.SEARCH_REQUESTS_TOPIC = process.env.SEARCH_REQUESTS_TOPIC || 'search.requests';
process.env.SEARCH_REQUESTS_SUBSCRIPTION = process.env.SEARCH_REQUESTS_SUBSCRIPTION || process.env.SEARCH_REQUESTS_TOPIC + '-sub';

process.env.SEARCH_RESPONSES_TOPIC = process.env.SEARCH_RESPONSES_TOPIC || 'search.responses';
process.env.SEARCH_RESPONSES_SUBSCRIPTION = process.env.SEARCH_RESPONSES_SUBSCRIPTION || process.env.SEARCH_RESPONSES_TOPIC + '-sub';

process.env.CREATE_INDEX_REQUESTS_TOPIC = process.env.CREATE_INDEX_REQUESTS_TOPIC || 'index.create.requests';
process.env.CREATE_INDEX_REQUESTS_SUBSCRIPTION = process.env.CREATE_INDEX_REQUESTS_SUBSCRIPTION || process.env.CREATE_INDEX_REQUESTS_TOPIC + '-sub';

process.env.TRUNCATE_INDEX_REQUESTS_TOPIC = process.env.TRUNCATE_INDEX_REQUESTS_TOPIC || 'index.truncate.requests';
process.env.TRUNCATE_INDEX_REQUESTS_SUBSCRIPTION = process.env.TRUNCATE_INDEX_REQUESTS_SUBSCRIPTION || process.env.TRUNCATE_INDEX_REQUESTS_TOPIC + '-sub';

process.env.UPSERT_DOCUMENT_REQUESTS_TOPIC = process.env.UPSERT_DOCUMENT_REQUESTS_TOPIC || 'document.upsert.requests';
process.env.UPSERT_DOCUMENT_REQUESTS_SUBSCRIPTION = process.env.UPSERT_DOCUMENT_REQUESTS_SUBSCRIPTION || process.env.UPSERT_DOCUMENT_REQUESTS_TOPIC + '-sub';

process.env.DELETE_DOCUMENT_REQUESTS_TOPIC = process.env.DELETE_DOCUMENT_REQUESTS_TOPIC || 'document.delete.requests';
process.env.DELETE_DOCUMENT_REQUESTS_SUBSCRIPTION = process.env.DELETE_DOCUMENT_REQUESTS_SUBSCRIPTION || process.env.DELETE_DOCUMENT_REQUESTS_TOPIC + '-sub';


process.env.PUBSUB_EMULATOR_HOST = process.env.PUBSUB_EMULATOR_HOST || '0.0.0.0:8085';
process.env.PUBSUB_PROJECT_ID = process.env.PUBSUB_PROJECT_ID || 'pure-plaform';

const createTopicAndSubscription = async (pubsub: PubSub, {topic: topicName, subscription: subscriptionName}: {topic: string, subscription: string}) => {
    let topic = pubsub.topic(topicName);

    const [topicExists] = await topic.exists();

    if(!topicExists) {
        [topic] =  await pubsub.createTopic(topicName);
        console.log(`Topic created ${topicName}`);
    } else {
        console.log(`Topic exists ${topicName}`);
    }

    let subscription = topic.subscription(subscriptionName);

    const [subscriptionExists] = await subscription.exists()

    if(!subscriptionExists) {
        [subscription] = await topic.createSubscription(subscriptionName)
        console.log(`Subscription created ${topicName}`);
    } else {
        console.log(`Subscription exists ${topicName}`);
    }

    return {topic, subscription};
}

const main = async (): Promise<string> => {
    console.log("Creating pub/sub topics and subscriptions on project %s.", process.env.PUBSUB_PROJECT_ID);

    const client = new PubSub({projectId: process.env.PUBSUB_PROJECT_ID});

    const pubsubs = [
        { topic: process.env.SEARCH_REQUESTS_TOPIC, subscription: process.env.SEARCH_REQUESTS_SUBSCRIPTION },
        { topic: process.env.SEARCH_RESPONSES_TOPIC, subscription: process.env.SEARCH_RESPONSES_SUBSCRIPTION },
        { topic: process.env.CREATE_INDEX_REQUESTS_TOPIC, subscription: process.env.CREATE_INDEX_REQUESTS_SUBSCRIPTION },
        { topic: process.env.TRUNCATE_INDEX_REQUESTS_TOPIC, subscription: process.env.TRUNCATE_INDEX_REQUESTS_SUBSCRIPTION },
        { topic: process.env.UPSERT_DOCUMENT_REQUESTS_TOPIC, subscription: process.env.UPSERT_DOCUMENT_REQUESTS_SUBSCRIPTION },
        { topic: process.env.DELETE_DOCUMENT_REQUESTS_TOPIC, subscription: process.env.DELETE_DOCUMENT_REQUESTS_SUBSCRIPTION },
    ]

    for(const pubsub of pubsubs) {
        await createTopicAndSubscription(client, pubsub)
    }

    return 'Done';
}

main().then(console.log).catch(console.error);

After run it by

 ts-node .scripts/init-pubsub.ts

you can go to pages

http://localhost:8085/v1/projects/pure-plaform/topics
http://localhost:8085/v1/projects/pure-plaform/subscriptions

and see

[sylvainmouquet]

Another solution with curl :

curl -X PUT http://localhost:8085/v1/projects/myproject-local/topics/topicone-local
curl -X PUT http://localhost:8085/v1/projects/myproject-local/topics/topictwo-local
curl -X GET http://localhost:8085/v1/projects/myproject-local/topics

and the full version :

File:Makefile 

.PHONY: pubsub
pubsub: ## pubsub
	echo "Starting pubsub in local : port 8085"
	docker stop pubsub || true
	docker rm pubsub || true
	sleep 1
	docker run --name pubsub -d -it -p 8085:8085 -e PUBSUB_EMULATOR_HOST=0.0.0.0:8085 -e PUBSUB_PROJECT_ID="myproject-local" knarz/pubsub-emulator:latest
	sleep 2

	curl -X PUT http://localhost:8085/v1/projects/myproject-local/topics/topicone-local
	curl -X PUT http://localhost:8085/v1/projects/myproject-local/topics/topictwo-local
	curl -X GET http://localhost:8085/v1/projects/myproject-local/topics