// Writes the given value to the sink. This function is called for every record.
default void invoke(IN value) throws Exception 
default void invoke(IN value, Context context) throws Exception 
// Context接口中返回关于时间的信息
interface Context<T> {

		/** Returns the current processing time. */
		long currentProcessingTime();

		/** Returns the current event-time watermark. */
		long currentWatermark();

		/**
		 * Returns the timestamp of the current input record or {@code null} if the element does not
		 * have an assigned timestamp.
		 */
		Long timestamp();
	}

public abstract class AbstractRichFunction implements RichFunction, Serializable

public DataStreamSink<T> print() {
	PrintSinkFunction<T> printFunction = new PrintSinkFunction<>();
	return addSink(printFunction).name("Print to Std. Out");
}

public class PrintSinkFunction<IN> extends RichSinkFunction<IN> {

	private static final long serialVersionUID = 1L;

	private final PrintSinkOutputWriter<IN> writer;

	/**
	 * Instantiates a print sink function that prints to standard out.
	 */
	public PrintSinkFunction() {
		writer = new PrintSinkOutputWriter<>(false);
	}

	/**
	 * Instantiates a print sink function that prints to standard out.
	 *
	 * @param stdErr True, if the format should print to standard error instead of standard out.
	 */
	public PrintSinkFunction(final boolean stdErr) {
		writer = new PrintSinkOutputWriter<>(stdErr);
	}

	/**
	 * Instantiates a print sink function that prints to standard out and gives a sink identifier.
	 *
	 * @param stdErr True, if the format should print to standard error instead of standard out.
	 * @param sinkIdentifier Message that identify sink and is prefixed to the output of the value
	 */
	public PrintSinkFunction(final String sinkIdentifier, final boolean stdErr) {
		writer = new PrintSinkOutputWriter<>(sinkIdentifier, stdErr);
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
		writer.open(context.getIndexOfThisSubtask(), context.getNumberOfParallelSubtasks());
	}

	@Override
	public void invoke(IN record) {
		writer.write(record);
	}

	@Override
	public String toString() {
		return writer.toString();
	}
}

object WriteToMySQL {

   def main(args: Array[String]): Unit = {

      val env = StreamExecutionEnvironment.getExecutionEnvironment

      val files = env.readTextFile("tunan-flink-stream/data/access.txt")

      val result = files.map(row => {
         val words = row.split(",").map(_.trim)
         Access(words(0).toLong, words(1), words(2).toLong)
      }).keyBy(x => (x.time, x.domain)).sum(2)

      result.addSink(new CustomMySQL)

      env.execute(this.getClass.getSimpleName)
   }
}

class CustomMySQL extends RichSinkFunction[Access]{
         var conn:Connection = _
         var pstate:PreparedStatement =_

         override def open(parameters: Configuration): Unit = {
            conn = MySQLUtils.getConnection
            pstate = conn.prepareStatement("REPLACE INTO access(time,domain,traffic) values(?,?,?)")
         }

         // 每条数据做一次插入操作，性能低下，需要根据window优化
         override def invoke(value: Access): Unit = {
            pstate.setLong(1,value.time)
            pstate.setString(2,value.domain)
            pstate.setLong(3,value.traffics)

            pstate.execute()
         }

         override def close(): Unit = {
            MySQLUtils.close(conn,pstate,null)
         }
      }