public PCollection expand()

in src/main/java/com/google/solutions/df/log/aggregations/common/ReadFlowLogTransform.java [61:77]

• 14 lines of code
• 1 McCabe index (conditional complexity)

  public PCollection<Row> expand(PBegin input) {

    PCollection<String> fileRow =
        input
            .apply(
                "ReadFromGCS",
                TextIO.read().from(filePattern()).watchForNewFiles(pollInterval(), Growth.never()))
            .apply("AssignEventTimestamp", WithTimestamps.of((String rec) -> Instant.now()));

    PCollection<String> pubsubMessage =
        input.apply("ReadFromPubSub", PubsubIO.readStrings().fromSubscription(subscriber()));

    return PCollectionList.of(fileRow)
        .and(pubsubMessage)
        .apply(Flatten.<String>pCollections())
        .apply("FlowLogs Convert To Rows", new JsonToRowValidationTransform());
  }