Function main

pubsub-avro/quickstart.cc:31–123

#include <string>

int main(int argc, char* argv[]) try {
  if (argc != 4) {
    std::cerr << "Usage: " << argv[0]
              << " <project-id> <subscription-id> <avro-file>\n";
    return 1;
  }

  std::string const project_id = argv[1];
  std::string const subscription_id = argv[2];
  std::string const avro_file = argv[3];

  auto constexpr kWaitTimeout = std::chrono::seconds(30);

  // Create a namespace alias to make the code easier to read.
  namespace pubsub = ::google::cloud::pubsub;

  //! [START pubsub_subscribe_avro_records_with_revisions]
  auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(
      pubsub::Subscription(project_id, subscription_id)));

  // Create a schema client.
  auto schema_client =
      pubsub::SchemaServiceClient(pubsub::MakeSchemaServiceConnection());

  // Read the reader schema. This is the schema you want the messages to be
  // evaluated using.
  std::ifstream ifs(avro_file);
  avro::ValidSchema reader_schema;
  avro::compileJsonSchema(ifs, reader_schema);

  std::unordered_map<std::string, avro::ValidSchema> revisions_to_schemas;
  auto session = subscriber.Subscribe(
      [&](pubsub::Message const& message, pubsub::AckHandler h) {
        // Get the reader schema revision for the message.
        auto schema_name = message.attributes()["googclient_schemaname"];
        auto schema_revision_id =
            message.attributes()["googclient_schemarevisionid"];
        // If we haven't received a message with this schema, look it up.
        if (revisions_to_schemas.find(schema_revision_id) ==
            revisions_to_schemas.end()) {
          auto schema_path = schema_name + "@" + schema_revision_id;
          // Use the schema client to get the path.
          auto schema = schema_client.GetSchema(schema_path);
          if (!schema) {
            std::cout << "Schema not found:" << schema_path << "\n";
            return;
          }
          avro::ValidSchema writer_schema;
          std::stringstream in;
          in << schema.value().definition();
          avro::compileJsonSchema(in, writer_schema);
          revisions_to_schemas[schema_revision_id] = writer_schema;
        }
        auto writer_schema = revisions_to_schemas[schema_revision_id];

        auto encoding = message.attributes()["googclient_schemaencoding"];
        if (encoding == "JSON") {
          std::stringstream in;