@@ -0,0 +1,3304 @@
+package com.rootcss.flink;%0A%0A/**%0A * Created by rootcss on 13/11/16.%0A */%0A%0Aimport com.rabbitmq.client.AMQP;%0Aimport org.apache.flink.api.common.functions.FlatMapFunction;%0Aimport org.apache.flink.api.java.tuple.Tuple2;%0Aimport org.apache.flink.streaming.api.datastream.DataStream;%0Aimport org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;%0Aimport org.apache.flink.streaming.api.windowing.time.Time;%0Aimport org.apache.flink.streaming.connectors.rabbitmq.RMQSource;%0Aimport org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;%0Aimport org.apache.flink.streaming.util.serialization.DeserializationSchema;%0Aimport org.apache.flink.streaming.util.serialization.SimpleStringSchema;%0Aimport org.apache.flink.util.Collector;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0Aimport java.io.IOException;%0A%0Apublic class RabbitmqWindowStreamProcessor extends RMQSource %7B%0A%0A    static String exchangeName          = %22simpl_exchange%22;%0A    static String queueName             = %22simpl_spark_dev%22;%0A    static String rabbitmqHostname      = %22localhost%22;%0A    static String rabbitmqVirtualHost   = %22/%22;%0A    static String rabbitmqUsername      = %22rootcss%22;%0A    static String rabbitmqPassword      = %22indian%22;%0A    static boolean durableQueue         = false;%0A    static Integer rabbitmqPort         = 5672;%0A    static int windowSeconds            = 10;%0A    static String outputFile            = %22/Users/rootcss/infrastructure/codes/flink/flink-rabbitmq/output.out%22;%0A%0A    private static Logger logger = LoggerFactory.getLogger(RabbitmqStreamProcessor.class);%0A%0A    public RabbitmqWindowStreamProcessor(RMQConnectionConfig rmqConnectionConfig, String queueName, DeserializationSchema deserializationSchema) %7B%0A        super(rmqConnectionConfig, queueName, deserializationSchema);%0A    %7D%0A%0A    @Override%0A    protected void setupQueue() throws IOException %7B%0A        AMQP.Queue.DeclareOk result = channel.queueDeclare(queueName, true, durableQueue, false, null);%0A        channel.queueBind(result.getQueue(), exchangeName, %22*%22);%0A    %7D%0A%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        logger.info(%22Starting Rabbitmq Stream Processor..%22);%0A%0A        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()%0A                .setHost(rabbitmqHostname).setPort(rabbitmqPort).setUserName(rabbitmqUsername)%0A                .setPassword(rabbitmqPassword).setVirtualHost(rabbitmqVirtualHost).build();%0A%0A        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();%0A%0A        DataStream%3CString%3E dataStream = env.addSource(new RMQSource%3CString%3E(connectionConfig,%0A                queueName,%0A                new SimpleStringSchema()));%0A%0A        DataStream%3CTuple2%3CString, Integer%3E%3E pairs = dataStream%0A                .flatMap(new TextLengthCalculator())%0A                .keyBy(0)%0A                .timeWindow(Time.seconds(windowSeconds))%0A                .sum(1);%0A%0A        pairs.writeAsText(outputFile);%0A        env.execute();%0A    %7D%0A%0A    public static final class TextLengthCalculator implements FlatMapFunction%3CString, Tuple2%3CString, Integer%3E%3E %7B%0A%0A        @Override%0A        public void flatMap(String value, Collector%3CTuple2%3CString, Integer%3E%3E out) %7B%0A            out.collect(new Tuple2%3CString, Integer%3E(value, value.length()));%0A        %7D%0A%0A    %7D%0A%0A%7D%0A

write to a file based on window