### Question:
StreamingDataGenerator { public static PipelineResult run(StreamingDataGeneratorOptions options) { Pipeline pipeline = Pipeline.create(options); pipeline .apply( "Trigger", GenerateSequence.from(0L).withRate(options.getQps(), Duration.standardSeconds(1L))) .apply("GenerateMessages", ParDo.of(new MessageGeneratorFn(options.getSchemaLocation()))) .apply("WriteToPubsub", PubsubIO.writeMessages().to(options.getTopic())); return pipeline.run(); }  static void main(String[] args); static PipelineResult run(StreamingDataGeneratorOptions options);  }### Answer:
@Test public void testMessageGenerator() throws IOException { String schema = "{" + "\"id\": \"{{uuid()}}\", " + "\"eventTime\": \"{{timestamp()}}\", " + "\"username\": \"{{username()}}\", " + "\"score\": {{integer(0,100)}}" + "}"; File file = tempFolder.newFile(); writeToFile(file.getAbsolutePath(), schema); PCollection<PubsubMessage> results = pipeline .apply("CreateInput", Create.of(0L)) .apply("GenerateMessage", ParDo.of(new MessageGeneratorFn(file.getAbsolutePath()))); PAssert.that(results) .satisfies( input -> { PubsubMessage message = input.iterator().next(); assertThat(message, is(notNullValue())); assertThat(message.getPayload(), is(notNullValue())); assertThat(message.getAttributeMap(), is(notNullValue())); return null; }); pipeline.run(); }
@Test public void testMessageGeneratorInvalidSchema() throws IOException { String schema = "{\"name: \"Invalid\""; File file = tempFolder.newFile(); writeToFile(file.getAbsolutePath(), schema); PCollection<PubsubMessage> results = pipeline .apply("CreateInput", Create.of(0L)) .apply("GenerateMessage", ParDo.of(new MessageGeneratorFn(file.getAbsolutePath()))); PAssert.that(results).satisfies(input -> { PubsubMessage message = input.iterator().next(); assertThat(message, is(notNullValue())); assertThat(new String(message.getPayload()), is(equalTo(schema))); return null; }); pipeline.run(); }