package com.mremes.weatherlytics;

import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.options.Description;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.transforms.DoFn;
import com.google.cloud.dataflow.sdk.transforms.ParDo;
import com.google.cloud.dataflow.sdk.io.BigQueryIO;
import com.google.cloud.dataflow.sdk.io.PubsubIO;
import com.google.cloud.dataflow.sdk.values.PCollection;
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;

public class StarterPipeline {
	static class StringToRawTableRow extends DoFn<String,TableRow> {
		@Override
		public void processElement(ProcessContext c) throws Exception {
			String timestamp = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").print(DateTime.now());
			c.output(new TableRow()
					.set("timestamp",timestamp)
					.set("data",c.element()));
		}
	}

	static class SplitInputString extends DoFn<String,String> {
		@Override
		public void processElement(ProcessContext c) throws Exception {
			for(String s : c.element().split("\n")) {
				c.output(s);
			}
		}
	}

	private static TableSchema getTableSchema() {
	    List<TableFieldSchema> fields = new ArrayList();
	    fields.add(new TableFieldSchema().setName("timestamp").setType("TIMESTAMP"));
	    fields.add(new TableFieldSchema().setName("data").setType("STRING"));

	    return new TableSchema().setFields(fields);
  }

	public static void main(String[] args) {
		PipelineOptionsFactory.register(BasicOptions.class);
		BasicOptions options = PipelineOptionsFactory.fromArgs(args)
			.withValidation()
			.create()
			.as(BasicOptions.class);

		Pipeline p = Pipeline.create(options);
		TableSchema schema = getTableSchema();

		String pubSubSubscription = String.format(
				"projects/%s/subscriptions/%s", options.getProjectId(), options.getSubscription());

		p.apply(PubsubIO
				.Read
				.named("ReadFromPubSub")
				.subscription(pubSubSubscription))
		.apply(ParDo.of(new SplitInputString()))
		.apply(ParDo.of(new StringToRawTableRow()))
		.apply(BigQueryIO.Write
				.named("WriteToBigQuery")
				.to(String.format("%s:%s.raw", options.getProjectName(), options.getDataSet()))
				.withSchema(schema)
				.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
				.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));
		p.run();
  }
}

@@ -318,16 +318,76 @@
 s.DoFn;%0A
+import com.google.cloud.dataflow.sdk.transforms.PTransform;%0A
 import c
@@ -433,16 +433,16 @@
 .ParDo;%0A
-
 import c
@@ -989,16 +989,17 @@
 eline %7B%0A
+%0A
 %09static 
@@ -1023,16 +1023,18 @@
 TableRow
+Fn
  extends
@@ -1360,16 +1360,18 @@
 utString
+Fn
  extends
@@ -1546,16 +1546,418 @@
 %09%09%7D%0A%09%7D%0A%0A
+%09static class InputToTableRows extends PTransform%3CPCollection%3CString%3E,PCollection%3CTableRow%3E%3E %7B%0A%09%09@Override%0A        public PCollection%3CTableRow%3E apply(PCollection%3CString%3E input) %7B%0A            PCollection%3CString%3E lines = input.apply(ParDo.of(new SplitInputStringFn()));%0A%0A            PCollection%3CTableRow%3E rows = lines.apply(ParDo.of(new StringToRawTableRowFn()));%0A%0A            return rows;%0A        %7D%0A%09%7D%0A%0A
 %09private
@@ -2812,83 +2812,27 @@
 ply(
-ParDo.of(new SplitInputString()))%0A%09%09.apply(ParDo.of(new StringToRaw
+new InputTo
 TableRow
 ()))
@@ -2827,18 +2827,18 @@
 TableRow
+s
 (
-)
 ))%0A%09%09.ap

make a composite transform from fns