import com.alibaba.fastjson.JSONObject;
import config.Config;
import consumer.LocalConsumer;
import consumer.MqttConsumer;
import data.CommonData;
import extensions.KafkaHelper;
import extensions.MessgeRecorder;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by Administrator on 2017/10/25.
 */
public class App {
    public static void main(String[] args){

        MqttConsumer mqttConsumer = new MqttConsumer();
        MqttCallback callback = new MqttCallback() {
            LinkedBlockingQueue<String> rocketMQMsgCache = CommonData.getInstance().rocketMQMsgCache;
            public void connectionLost(Throwable throwable) {

            }

            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                String data = new String (mqttMessage.getPayload());
                rocketMQMsgCache.put(data);

                // todo send msg[json] to kafka
                JSONObject json = JSONObject.parseObject(data);
                String uId = json.getString("uId");
                String dataType = json.getString("dataType");
                String info = json.getString("info");

                if (!dataType.equals("telemetry")) return ;
                MessgeRecorder mr = new MessgeRecorder(info) ;
                mr.addCurTime().addUid(uId) ;

                KafkaHelper kh = new KafkaHelper() ;
                kh.sengMsg(mr.toJson().toString());
            }

            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        };
        mqttConsumer.init(callback,Config.TOPIC);


        LocalConsumer localConsumer = new LocalConsumer(1);
        localConsumer.init();

        localConsumer.startConsume();
        mqttConsumer.startConsume();
    }
}

@@ -1099,16 +1099,42 @@
 o kafka%0A
+                try %7B%0A    
         
@@ -1189,16 +1189,20 @@
 (data);%0A
+    
         
@@ -1253,32 +1253,36 @@
                 
+    
 String dataType 
@@ -1323,24 +1323,28 @@
             
+  
+  
 String info 
@@ -1366,24 +1366,28 @@
 g(%22info%22);%0A%0A
+    
             
@@ -1431,19 +1431,22 @@
 ) return
- 
 ;%0A
+    
         
@@ -1497,19 +1497,22 @@
 er(info)
- 
 ;%0A
+    
         
@@ -1546,17 +1546,16 @@
 Uid(uId)
- 
 ;%0A%0A     
@@ -1565,16 +1565,20 @@
         
+    
 KafkaHel
@@ -1603,17 +1603,16 @@
 Helper()
- 
 ;%0A      
@@ -1621,16 +1621,20 @@
         
+ 
+   
 kh.sengM
@@ -1657,24 +1657,149 @@
 oString());%0A
+                %7D catch (Exception e) %7B%0A                    System.err.println(%22fail to send to kafka!%22) ;%0A                %7D%0A
             

add extension for sending data to kafka Add: config.Kafka_Config extension.KafkaHelper extension.MessageRecorder Update: App