<commit_before>package com.ftchinese.jobs

import java.util.Properties

import akka.actor.Actor
import com.alibaba.fastjson.JSON
import com.ftchinese.jobs.common._
import com.ftchinese.jobs.database.{AnalyticDB, AnalyticDataSource, BeanConfig}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.slf4j.MDC
import org.springframework.context.annotation.AnnotationConfigApplicationContext


/**
 * The thread of token receiving worker.
 * Created by wanbo on 16/3/14.
 */
class ReceiveWorker(topics: Array[String], kafkaConf: Properties, conf: JobsConfig) extends Actor with Logging {

    private val _consumer = new MessageConsumer(kafkaConf)

    private var _latestConsume = Map[TopicPartition, Long]()

    private var _status = "start"

    private var _dbConf = Map[String, String]()

    private var _bufferedDataList = List[TokenMessage]()

    MDC.put("destination", "system")

    def consumeMessage(): Unit = {


        _consumer.consume((recordsList: List[ConsumerRecord[String, String]]) => {

            try {

                recordsList.foreach(record => {

                    val topic = record.topic()
                    val partition = record.partition()
                    val offset = record.offset()
                    val message = record.value()

                    log.info(topic + "###" + partition + "###" + offset + "###" + message)

                    _latestConsume = _latestConsume.updated(new TopicPartition(topic, partition), offset)

                    val obj = JSON.parseObject(message)

                    var deviceToken = ""
                    var timeZone = ""
                    var status = ""
                    var preference = ""
                    var deviceType = ""
                    var appNumber = "0"
                    var timestamp = ""

                    if(obj.containsKey("deviceToken"))
                        deviceToken = obj.getString("deviceToken").trim
                    if(obj.containsKey("timeZone"))
                        timeZone = obj.getString("timeZone").trim
                    if(obj.containsKey("status"))
                        status = obj.getString("status").trim
                    if(obj.containsKey("preference"))
                        preference = obj.getString("preference").trim
                    if(obj.containsKey("deviceType"))
                        deviceType = obj.getString("deviceType").trim
                    if(obj.containsKey("appNumber"))
                        appNumber = obj.getString("appNumber").trim
                    if(obj.containsKey("timestamp"))
                        timestamp = obj.getString("timestamp").trim

                    _bufferedDataList = _bufferedDataList :+ TokenMessage(deviceToken, timeZone, status, preference, deviceType, appNumber, timestamp)
                })

                if(_bufferedDataList.size > 0) {
                    _bufferedDataList.synchronized {
                        save(_bufferedDataList)
                        _bufferedDataList = List[TokenMessage]()
                    }
                }

                // Slow down the consume speed.
                Thread.sleep(conf.kafka_consumer_consumeInterval)

            } catch {
                case e: Exception => // Ignore
            }

        })

        if(_bufferedDataList.size > 0) {
            _bufferedDataList.synchronized {
                save(_bufferedDataList)
                _bufferedDataList = List[TokenMessage]()
            }
        }

        if (_status == "start") {
            self ! "consume"
        } else {
            _status = "stopped"
        }
    }

    override def receive: Receive = {
        case "Start" =>

            ZookeeperManager.init(conf)

            _dbConf = conf.fetchDriverConf("analytic", "mysql")

            if(_dbConf.size > 0) {

                // 85792L
                if (conf.kafka_consumer_defaultOffset > 0)
                    _consumer.subscribe(topics, conf.kafka_consumer_defaultOffset)
                else
                    _consumer.subscribe(topics)

                self ! "consume"<commit_msg>Update properties.<commit_after>package com.ftchinese.jobs

import java.util.Properties

import akka.actor.Actor
import com.alibaba.fastjson.JSON
import com.ftchinese.jobs.common._
import com.ftchinese.jobs.database.{AnalyticDB, AnalyticDataSource, BeanConfig}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.slf4j.MDC
import org.springframework.context.annotation.AnnotationConfigApplicationContext


/**
 * The thread of token receiving worker.
 * Created by wanbo on 16/3/14.
 */
class ReceiveWorker(topics: Array[String], kafkaConf: Properties, conf: JobsConfig) extends Actor with Logging {

    private val _consumer = new MessageConsumer(kafkaConf)

    private var _latestConsume = Map[TopicPartition, Long]()

    private var _status = "start"

    private var _dbConf = Map[String, String]()

    private var _bufferedDataList = List[TokenMessage]()

    MDC.put("destination", "system")

    def consumeMessage(): Unit = {


        _consumer.consume((recordsList: List[ConsumerRecord[String, String]]) => {

            try {

                recordsList.foreach(record => {

                    val topic = record.topic()
                    val partition = record.partition()
                    val offset = record.offset()
                    val message = record.value()

                    log.info(topic + "###" + partition + "###" + offset + "###" + message)

                    _latestConsume = _latestConsume.updated(new TopicPartition(topic, partition), offset)

                    val obj = JSON.parseObject(message)

                    var deviceToken = ""
                    var timeZone = ""
                    var status = ""
                    var preference = ""
                    var deviceType = ""
                    var appNumber = "0"
                    var timestamp = ""

                    if(obj.containsKey("deviceToken"))
                        deviceToken = obj.getString("deviceToken").trim
                    if(obj.containsKey("timeZone"))
                        timeZone = obj.getString("timeZone").trim
                    if(obj.containsKey("status"))
                        status = obj.getString("status").trim
                    if(obj.containsKey("preference"))
                        preference = obj.getString("preference").trim
                    if(obj.containsKey("deviceType"))
                        deviceType = obj.getString("deviceType").trim
                    if(obj.containsKey("appNumber"))
                        appNumber = obj.getString("appNumber").trim
                    if(obj.containsKey("timestamp"))
                        timestamp = obj.getString("timestamp").trim

                    _bufferedDataList = _bufferedDataList :+ TokenMessage(deviceToken, timeZone, status, preference, deviceType, appNumber, timestamp)
                })

                if(_bufferedDataList.nonEmpty) {
                    _bufferedDataList.synchronized {
                        save(_bufferedDataList)
                        _bufferedDataList = List[TokenMessage]()
                    }
                }

                // Slow down the consume speed.
                Thread.sleep(conf.kafka_consumer_consumeInterval)

            } catch {
                case e: Exception => // Ignore
            }

        })

        if(_bufferedDataList.nonEmpty) {
            _bufferedDataList.synchronized {
                save(_bufferedDataList)
                _bufferedDataList = List[TokenMessage]()
            }
        }

        if (_status == "start") {
            self ! "consume"
        } else {
            _status = "stopped"
        }
    }

    override def receive: Receive = {
        case "Start" =>

            ZookeeperManager.init(conf)

            _dbConf = conf.fetchDriverConf("analytic", "mysql")

            if(_dbConf.nonEmpty) {

                // 85792L
                if (conf.kafka_consumer_defaultOffset > 0)
                    _consumer.subscribe(topics, conf.kafka_consumer_defaultOffset)
                else
                    _consumer.subscribe(topics)

                self ! "consume"
{'fraction_non_alphanumeric': 0.0968783638320775, 'fraction_numerical': 0.003468484631024997, 'mean_word_length': 1.6503961965134706, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1212582', 'n_tokens_mistral': 2198, 'n_tokens_neox': 1964, 'n_words': 395}