C++ pubsub subscribers not pulling messages

[iamsamskup]

Hi all,

I'm attempting to test my pubsub code but am running into issues with creating subscriber objects connecting to subscriptions. I have a pubsub emulator running in a docker container separate from the container running the pubsub code. I'm able to create subscriptions to the topics running in the emulator container from my pubsub container but am not receiving published messages. I have a unique subscription created for each topic (randomly generated string of numbers) so I know that my code is creating unique subscription Id's.

Below are the relevant parts of the code:

Class members:

  std::string projectID{}; // read in from an env var
  std::vector<google::cloud::pubsub::Subscription> subscriptions;
  std::vector<google::cloud::pubsub::Subscriber> subscribers;
  std::vector<std::pair<std::string, std::string>> topicSubscriptionList{}; // <topicId, subscriptionId>

Functions called:

  createSubscriptionIds(topicSubscriptionList);
  createGCPSubscriptions(topicSubscriptionList);
  createLocalSubscriptions(topicSubscriptionList);
  createLocalSubscribers(subscriptions);

Function definitions:

createSubscriptionIds(std::vector<std::pair<std::string, std::string>> &topSubList) {
  std::srand(std::time(NULL));

  // create invalidate.zone.* subscriptions
  topSubList.emplace_back("invalidate.zone.all", "tm_" + std::to_string(rand()) + "_invalidate.zone.all");

  topSubList.emplace_back("invalidate.zone.list", "tm_" + std::to_string(rand()) + "_invalidate.zone.list");

  topSubList.emplace_back("invalidate.zone.customer", "tm_" + std::to_string(rand()) + "_invalidate.zone.customer");
}

createGCPSubscriptions(std::vector<std::pair<std::string, std::string>> topSubList) {

  google::cloud::pubsub::SubscriptionAdminClient subscription_admin_client(
      google::cloud::pubsub::MakeSubscriptionAdminConnection());

  bool createdAllTopics{true};

  for (auto topicSubPair : topSubList) {
    try {
      auto sub = subscription_admin_client.CreateSubscription(
          google::cloud::pubsub::Topic(projectID, std::move(topicSubPair.first)), //topicSubPair.first == topicId
          google::cloud::pubsub::Subscription(projectID, std::move(topicSubPair.second))); //topicSubPair.second == subscriptionId

      if(sub.status().code() == google::cloud::StatusCode::kAlreadyExists
          || sub.status().code() == google::cloud::StatusCode::kOk) {
        WIRELOG_DEBUG(topicSubPair.second << " created") //topicSubPair.second == subscriptionId
      }
      if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
        WIRELOG_DEBUG(topicSubPair.second << " already exists") //topicSubPair.second == subscriptionId
        continue;
      }
      if (!sub) {
        WIRELOG_DEBUG("Error create GCP subscription to topic: " << topicSubPair.second << " and error: " << sub.status().message())
        createdAllTopics =  false;
      }
      WIRELOG_DEBUG("The subscription was successfully created: " << sub->DebugString())
    } catch (std::exception &e) {
      WIRELOG_DEBUG("Caught exception while trying to create GCP subscriptions. Exception was: " << e.what())
      createdAllTopics =  false;
    }
  }
  return createdAllTopics;
}

createLocalSubscriptions(std::vector<std::pair<std::string, std::string>> topSubList) {
  WIRELOG_DEBUG("Creating Subscriptions")

  if(topSubList.empty()) {
    WIRELOG_DEBUG("subscriptionID list is empty. Going to failure.")
    return false;
  }

  try {
    for (auto subscriptionId : topSubList) {
      WIRELOG_DEBUG("Creating subscription with subscription Id: " << subscriptionId.second) 
      google::cloud::pubsub::Subscription tmpSub(projectID, subscriptionId.second); // subscriptionId.second == subscriptionId
      subscriptions.emplace_back(tmpSub);
    }
  } catch (std::exception &e) {
    WIRELOG_DEBUG("Exception caught when creating subscriptions: " << e.what())
    WIRELOG_DEBUG("Not going to failure but outputting logs")
    return false;
  }
  return true;
}

createLocalSubscribers(std::vector<google::cloud::pubsub::Subscription> subscriptionsList) {

  WIRELOG_DEBUG("Creating local subscribers")

  try {
    for (auto subscription : subscriptionsList) {
      WIRELOG_DEBUG("Subscription project id: " << subscription.project_id() << " subscription id" << subscription.subscription_id())
      auto subscriber = google::cloud::pubsub::Subscriber(google::cloud::pubsub::MakeSubscriberConnection(subscription));
      //auto subscriber = google::cloud::pubsub::Subscriber(google::cloud::pubsub::MakeSubscriberConnection(subscription));
      auto session = subscriber.Subscribe(
          //[&,this](google::cloud::pubsub::Message const &m, google::cloud::pubsub::AckHandler h) {
            [](google::cloud::pubsub::Message const &m, google::cloud::pubsub::AckHandler h) {
            //std::string subscriptionId{subscription.subscription_id()};
            std::move(h).ack();
            //WIRELOG_DEBUG("Received message " << m.data() << " from subscriptionId: " << subscriptionId)
            WIRELOG_DEBUG("Got message")
            WIRELOG_DEBUG( "Received message " << m.data())
            //iamsystem::ScopeLock scopeLock(messageMutex);
            //messageQueue.emplace_back(subscriptionId, m.data());
          });
      WIRELOG_DEBUG("session.valid: " << session.valid())
      WIRELOG_DEBUG("session.is_ready: " << session.is_ready())
      //WIRELOG_DEBUG("session.get.code: " << session.get().code())
      //WIRELOG_DEBUG("session.get.error_info.reason: " << session.get().error_info().reason())
      //WIRELOG_DEBUG("session.get.message: " << session.get().message())
      //WIRELOG_DEBUG("session.get.ok: " << session.get().ok())
    }
  } catch (std::exception &e) {
    WIRELOG_DEBUG("Exception caught when creating subscribers: " << e.what())
    WIRELOG_DEBUG("Not going to failure but outputting logs")
    return false;
  }
  return true;
}

In this last function I commented out the code I implemented and reverted back to the example code to try and isolate the problem there but I still am not receiving any messages. I'm also able to confirm that I'm am actually publishing messages to the topics correctly as well.

I suspect that my issue is not using the correct subscription object when calling google:: cloud ::pubsub::MakeSubscriberConnection(subscription) but am not sure. It seems like I should be passing the object returned from subscription_admin_client.CreateSubscription() instead of passing a regular subscription object that only contains the project and subscription ids?

Lastly I attempted to output info about the session object from subscriber.Subscribe via:

  WIRELOG_DEBUG("session.valid: " << session.valid())                         this was "1"
  WIRELOG_DEBUG("session.is_ready: " << session.is_ready())            this was "0"
  //WIRELOG_DEBUG("session.get.code: " << session.get().code())
  //WIRELOG_DEBUG("session.get.error_info.reason: " << session.get().error_info().reason())
  //WIRELOG_DEBUG("session.get.message: " << session.get().message())
  //WIRELOG_DEBUG("session.get.ok: " << session.get().ok())

but had to comment out everything but the top 2 as the process was hung trying to get the other information. Not sure if that info was valuable or not. valid() returned 1 but is_ready() returned 0.

Thanks for the help so far.

[coryan]

This is a little too much code for me to diagnose. Can you create a smaller, reproducible program that shows the problem? I will try to answer your specific questions if I can:

I suspect that my issue is not using the correct subscription object when calling google:: cloud ::pubsub::MakeSubscriberConnection(subscription) but am not sure. It seems like I should be passing the object returned from subscription_admin_client.CreateSubscription() instead of passing a regular subscription object that only contains the project and subscription ids?

No. You should be able to use a pubsub::Subscription there. A pubusb::Subscription is a convenience class to just hold the name (just a string really) of a subscription. A subscription is a persistent, named, resource in the Pub/Sub service. This is not unlike to using std::ofstream to create a file, and then passing the name of the file to a function that is going to read from it.

Most applications create a subscription (via subscription_admin_client.CreateSubscription(), or using the gcloud CLI, or even using the UI) and then just provide the project id and subscription id to their C++ code to use. You seem to be creating and connecting to subscriptions in the same application. This is not unheard of, but less common than treating

In this last function I commented out the code I implemented and reverted back to the example code to try and isolate the problem there but I still am not receiving any messages. I'm also able to confirm that I'm am actually publishing messages to the topics correctly as well.

You are creating and then deleting the subscriber objects in a loop. The destructor for subscriber terminates the background threads and joins them. You need to make sure these objects are not deleted.

If you use google-cloud-cpp v2.5.0, I think you could write:

std::vector<google::cloud::future<Status>>
createLocalSubscribers(
    google::cloud::pubsub::Subscriber subscriber,
    std::vector<google::cloud::pubsub::Subscription> subscriptionsList) {

  WIRELOG_DEBUG("Creating local subscribers")

  std::vector<google::cloud::future<Status>> sessions;
  try {
    for (auto const& subscription : subscriptionsList) {
      auto session = subscriber.Subscribe(
            [](google::cloud::pubsub::Message const &m, google::cloud::pubsub::AckHandler h) {
                std::move(h).ack();
                WIRELOG_DEBUG("Got message")
                WIRELOG_DEBUG( "Received message " << m.data())
          },
          // *** NEW ** override the default subscription and reuse the Subscriber.
          google::cloud::Options{}.set<google::cloud::pubsub::SubscriptionOption>(subscription));
      sessions.push_back(std::move(session);
    }
  } catch (std::exception &e) {
    WIRELOG_DEBUG("Exception caught when creating subscribers: " << e.what())
    throw;
  }
  return sessions;
}

---

I have a unique subscription created for each topic (randomly generated string of numbers) so I know that my code is creating unique subscription Id's.

As an aside, as I don't think it is relevant to the problem at hand. You seem to be using std::rand() with std::srand(std::time(NULL)). That is not a very good source of unique numbers and strings. Programs starting close to the same second will have the same seed. And std::rand() is a terrible number generator:

https://en.cppreference.com/w/cpp/numeric/random/rand

"There are no guarantees as to the quality of the random sequence produced. In the past, some implementations of rand() have had serious shortcomings in the randomness, distribution and period of the sequence produced (in one well-known example, the low-order bit simply alternated between 1 and 0 between calls)"

This is a large topic, but this is a good start:

https://www.pcg-random.org/using-pcg.html

[iamsamskup]

Thanks again for the help, the issue was deleting the subscriber objects in the loop. I'll check out the rand() fix too.