/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.well.mysql.sink;

import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.well.well.base.AbstractResponse;
import org.well.well.manager.DeviceManager;
import org.well.well.util.ResponseResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.util.List;

public class WellSink extends AbstractSink implements Configurable {

    private Logger LOG = LoggerFactory.getLogger(WellSink.class);
    private int batchSize;
    private ClassPathXmlApplicationContext ac = null;
    public WellSink() {
        LOG.info("wellMysqlSink start...");
    }
    public void configure(Context context) {
        String s[] = System.getProperty("java.class.path").split(";");
        for (String string : s) {
            System.out.println("**********************"+string+"************************");
        }
        ac = new ClassPathXmlApplicationContext(
                new String[]{"classpath:wellSensor/*.xml"});
        batchSize = context.getInteger("batchSize", 100);
    }

    @Override
    public void start() {
        super.start();
        LOG.info("--------wellMysqlSink start-------");
        System.out.println("--------wellMysqlSink start-------");


    }

    @Override
    public void stop() {
        super.stop();
        LOG.info("--------wellMysqlSink stop-------");

    }

    public Status process() throws EventDeliveryException {
        Status result = Status.READY;
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        Event event;
        String content;
        List<String> actions = Lists.newArrayList();
        transaction.begin();
        try {
            for (int i = 0; i < batchSize; i++) {
                event = channel.take();
                if (event != null) {
                    content = new String(event.getBody());
                    actions.add(content);
                } else {
                    result = Status.BACKOFF;
                    break;
                }
            }
            if (actions.size() > 0) {
                for (String temp : actions) {
                    LOG.info("--------wellMysqlSink接收数据：" + temp + "-------");
                    AbstractResponse resp = ResponseResolver.makeResponse(temp);
                    if (resp == null) continue;
                    resp.setAc(ac);
                    resp.process(temp);//对应的消息处理
                    LOG.info("--------wellMysqlSink存库处理完！------");
                }

            }
            transaction.commit();
        } catch (Throwable e) {
            try {
//                transaction.rollback();
                transaction.commit();
            } catch (Exception e2) {
                LOG.error("Exception in rollback. Rollback might not have been" +
                        "successful.", e2);
            }
            LOG.error("Failed to commit transaction." +
                    "Transaction rolled back.", e);
            Throwables.propagate(e);
        } finally {
            transaction.close();
        }
        return result;
    }


    public static void main(String[] args) {

        ClassPathXmlApplicationContext ac = new ClassPathXmlApplicationContext(
                new String[]{"classpath:wellSensor/*.xml"});
        ac.start();
        String temp =
                "{\"mType\":\"Data\",\"devType\":\"Liquid\",\"devCode\":\"112018030001\",\"mBody\":{\"bType\":\"LiquidData\",\"cell\": 90,\"datas\": [{\"uptime\":\"20180911222000\",\"level\":0.55}, {\"uptime\":\"20181111212001\",\"level\": 0.46}],\"logTime\":\"20190315010000\"},\"ts\":\"12345678901\"}";

//        temp="{\"mType\":\"Event\",\"devType\":\"Concentrator\",\"devCode\":\"00003\",\"mBody\":{\"logTime\":\"20190605002024\",\"bType\":\"ConcentratorOnline\"},\"ts\":1559665224343}";

        temp="{\"mType\":\"Data\",\"devType\":\"Well\",\"devCode\":\"412019010212\",\"mBody\":{\"datas\":[{\"value\":\"00\",\"uptime\":\"20190605000000\"}],\"logTime\":\"201906010003002\",\"bType\":\"WellData\"},\"ts\":1559665802828}";
       temp="{\"mType\":\"Data\",\"devType\":\"Liquid\",\"devCode\":\"11201900001\",\"mBody\":{\"cell\":9.9,\"datas\":[{\"level\":1.1892,\"uptime\":\"20191213000000\"}],\"logTime\":\"20190502000125\",\"bType\":\"LiquidData\"},\"ts\":1556726485336}";
//      temp="{\"mType\":\"Event\",\"devType\":\"Liquid\",\"devCode\":\"11201900001\",\"mBody\":{\"eventType\":[\"LiquidPressureError\"],\"logTime\":\"20190510134635\",\"bType\":\"LiquidEvent\"},\"ts\":1557467195358}";
//        temp="{\"mType\":\"Event\",\"devType\":\"Well\",\"devCode\":\"412019010212\",\"mBody\":{\"eventType\":[\"WellLowBatteryAlarm\"],\"logTime\":\"20190624114710\",\"bType\":\"WellEvent\"},\"ts\":1560484030810}";
//                temp="{\"mType\":\"Data\",\"devType\":\"Locator\",\"devCode\":\"71201900001\",\"mBody\":{\"datas\":[{\"longitude\":126.243324343,\"latitude\":39.2546546546,\"uptime\":\"20190809140900\"}],\"logTime\":\"20190809141012\",\"bType\":\"LocatorData\"},\"ts\":1565331012453}";
//        temp="{\"mType\":\"Data\",\"devType\":\"NoiseDig\",\"devCode\":\"612019030052\",\"mBody\":{\"cell\":99,\"datas\":[{\"noiseVal\":1000.7926,\"noiseFreq\":1421,\"uptime\":\"20191017141000\"},{\"noiseVal\":48.1797,\"noiseFreq\":212,\"uptime\":\"20200407091300\"},{\"noiseVal\":33.1398,\"noiseFreq\":553,\"uptime\":\"20200407091300\"},{\"noiseVal\":0.85,\"noiseFreq\":1049,\"uptime\":\"20200407091300\"},{\"noiseVal\":1000,\"noiseFreq\":1545,\"uptime\":\"20200407091300\"}],\"logTime\":\"20200407091300\",\"bType\":\"NoiseDigData\"},\"ts\":1566454417130}";
//        temp="{\"mType\":\"Data\",\"devType\":\"WasteGas\",\"devCode\":\"21201900001\",\"mBody\":{\"datas\":[{\"CO\":60,\"O2\":25,\"H2S\":15,\"CH4\":7,\"liquidSwitch\":true,\"uptime\":\"20191018163319\"}],\"logTime\":\"20191018151844\",\"bType\":\"WasteGasData\"},\"ts\":1571296724288}";
//        temp="{\"mType\":\"Data\",\"devType\":\"TempHumi\",\"devCode\":\"11111111114\",\"mBody\":{\"cell\":99,\"datas\":[{\"temperature\":-12,\"humidity\":82,\"uptime\":\"20200119150001\"}],\"logTime\":\"20200119150001\",\"bType\":\"TempHumiData\"},\"ts\":1571295745881}";
//              temp="{\"mType\":\"Event\",\"devType\":\"Liquid\",\"devCode\":\"11201900001\",\"mBody\":{\"eventType\":[\"LiquidUltrasonicError\"],\"logTime\":\"20190710134635\",\"bType\":\"LiquidEvent\"},\"ts\":1557467195358}";


//        temp="{\"mType\":\"Event\",\"devType\":\"TempHumi\",\"devCode\":\"51201900001\",\"mBody\":{\"eventType\":[\"TemperatureFail\",\"TemperatureError\",\"HumidityFail\",\"HumidityError\"],\"logTime\":\"20191021140118\",\"bType\":\"TempHumiEvent\"},\"ts\":1571292078959}";
//        temp="{\"mType\":\"SetResponse\",\"devType\":\"NoiseDig\",\"devCode\":\"88888888881\",\"mBody\":{\"bType\":\"NoiseDigConfigSuccess\"},\"ts\":1556182310514}";
//        temp="{\"mType\":\"Data\",\"devType\":\"WasteGas\",\"devCode\":\"13131313132\",\"mBody\":{\"bType\":\"WasteGasData\",\"datas\":[{\"CO\":80,\"O2\":18,\"H2S\":10,\"CH4\":5,\"liquidSwitch\":true,\"uptime\":\"20191220165124\"}],\"logTime\":\"20191220165124\"},\"ts\":1556182310514}";
//        temp="{\"mType\":\"Data\",\"devType\":\"Liquid\",\"devCode\":\"13131313135\",\"mBody\":{\"bType\":\"LiquidData\",\"cell\":98,\"datas\":[{\"level\":6.73,\"uptime\":\"20191217161000\"},{\"level\":5.72,\"uptime\":\"20191217162000\"},{\"level\":7.71,\"uptime\":\"20200211103000\"}],\"logTime\":\"20200211103000\"},\"ts\":1556186030842}";
//       temp="{\"mType\":\"SetResponse\",\"devType\":\"Liquid\",\"devCode\":\"12121212125\",\"mBody\":{\"bType\":\"LiquidConfigSuccess\"},\"ts\":1556182310514}";
//        temp="{\"mType\":\"Data\",\"devType\":\"Liquid\",\"devCode\":\"12121212125\",\"mBody\":{\"bType\":\"LiquidData\",\"cell\":97,\"datas\":[{\"level\":4,\"uptime\":\"20191219150000\"},{\"level\":4,\"uptime\":\"20191219151000\"},{\"level\":6.5,\"uptime\":\"20191219152000\"}],\"logTime\":\"20191219152000\"},\"ts\":1556186030842}";
//        temp="{\"mType\":\"Data\",\"devType\":\"NoiseDig\",\"devCode\":\"14141414146\",\"mBody\":{\"bType\":\"NoiseDigData\",\"cell\":88,\"datas\":[{\"noiseVal\":60,\"noiseFreq\":50,\"uptime\":\"20200109123131\"}],\"logTime\":\"20200119123131\"},\"ts\":1556184691451}";
//        temp="{\"mType\":\"Event\",\"devType\":\"Well\",\"devCode\":\"13131313131\",\"mBody\":{\"eventType\":[\"WellOpenAlarm\"],\"logTime\":\"20191017154056\",\"bType\":\"WellEvent\"},\"ts\":1571298056241}";
//              temp="{\"mType\":\"Data\",\"devType\":\"Methane\",\"devCode\":\"312019020253\",\"mBody\":{\"cell\":95,\"datas\":[{\"gas\":0.064453125,\"uptime\":\"20200509085400\"},{\"gas\":0.080566406,\"uptime\":\"20200509085400\"},{\"gas\":25,\"uptime\":\"20200407085400\"},{\"gas\":0.09990235,\"uptime\":\"20200509085400\"},{\"gas\":0.070898436,\"uptime\":\"20200407085400\"},{\"gas\":60,\"uptime\":\"20200509085400\"}],\"logTime\":\"20200407085400\",\"bType\":\"MethaneData\"},\"ts\":1571292084960}";

//        temp="{\"mType\":\"Data\",\"devType\":\"Locator\",\"devCode\":\"642019010387\",\"mBody\":{\"datas\":[{\"longitude\":0.0,\"latitude\":0.0,\"uptime\":\"20200118111000\"}],\"logTime\":\"20200118111006\",\"bType\":\"LocatorData\"},\"ts\":1579317006078}";

//        temp="{\"mType\":\"Data\",\"devType\":\"LG\",\"devCode\":\"322000010001\",\"mBody\":{\"cell\":100,\"datas\":[{\"gas\":0.0,\"uptime\":\"20190825232100\"},{\"liquid\":0.0,\"uptime\":\"20190825232600\"},{\"liquid\":0.0,\"uptime\":\"20190825233100\"},{\"liquid\":0.0,\"uptime\":\"20190825233600\"},{\"liquid\":0.0,\"uptime\":\"20190825234100\"},{\"liquid\":0.0,\"uptime\":\"20190825234600\"},{\"liquid\":0.0,\"uptime\":\"20190825235100\"},{\"liquid\":0.0,\"uptime\":\"20190825235600\"},{\"liquid\":0.0,\"uptime\":\"20190826000100\"},{\"liquid\":0.0,\"uptime\":\"20190826000600\"},{\"liquid\":0.0,\"uptime\":\"20190826001100\"},{\"liquid\":0.0,\"uptime\":\"20190826001600\"},{\"liquid\":0.0,\"uptime\":\"20190826002100\"},{\"liquid\":0.0,\"uptime\":\"20190826002600\"},{\"liquid\":0.0,\"uptime\":\"20190826003100\"}],\"logTime\":\"20190826003335\",\"bType\":\"LGData\"},\"ts\":1566750815981}";

        AbstractResponse resp = ResponseResolver.makeResponse(temp);
        resp.setAc(ac);
        resp.process(temp);
//        DeviceManager deviceManager = ac.getBean(DeviceManager.class);
//        deviceManager.updateDeviceStatus();

    }
}  
  
