def generate_events(self):
    publish_client = pubsub.Client(project=self.project)
    topic = publish_client.topic(self.topic_name)
    sub = topic.subscription(self.subscription_name)
    logging.info('Generating auction events to topic %s', topic.name)
    if self.args.input.startswith('gs://'):
        from apache_beam.io.gcp.gcsfilesystem import GCSFileSystem
        fs = GCSFileSystem(self.pipeline_options)
        with fs.open(self.args.input) as infile:
            for line in infile:
                topic.publish(line)
    else:
        with open(self.args.input) as infile:
            for line in infile:
                topic.publish(line)
    logging.info('Finished event generation.')
    if self.args.subscription_name:
        raw_events = (self.pipeline | ('ReadPubSub' >> beam.io.ReadFromPubSub(subscription=sub.full_name)))
    else:
        raw_events = (self.pipeline | ('ReadPubSub' >> beam.io.ReadFromPubSub(topic=topic.full_name)))
    return raw_events
def generate_events(self):
    publish_client = pubsub.Client(project=self.project)
    topic = publish_client.topic(self.topic_name)
    sub = topic.subscription(self.subscription_name)
    logging.info('Generating auction events to topic %s', topic.name)
    if self.args.input.startswith('gs://'):
        from apache_beam.io.gcp.gcsfilesystem import GCSFileSystem
        fs = GCSFileSystem(self.pipeline_options)
        with fs.open(self.args.input) as infile:
            for line in infile:
                topic.publish(line)
    else:
        with open(self.args.input) as infile:
            for line in infile:
                topic.publish(line)
    logging.info('Finished event generation.')
    if self.args.subscription_name:
        raw_events = (self.pipeline | ('ReadPubSub' >> beam.io.ReadFromPubSub(subscription=sub.full_name)))
    else:
        raw_events = (self.pipeline | ('ReadPubSub' >> beam.io.ReadFromPubSub(topic=topic.full_name)))
    return raw_events