<commit_before>/*******************************************************************************
 * Copyright 2014 Stratio
 * 
 * Licensed 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
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * 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 com.stratio.streaming.examples;

import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple2;

import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.stratio.streaming.commons.constants.BUS;
import com.stratio.streaming.commons.constants.STREAM_OPERATIONS;
import com.stratio.streaming.commons.messages.ColumnNameTypeValue;
import com.stratio.streaming.commons.messages.StratioStreamingMessage;

public class SensorGridSimulation {

    private static Logger logger = LoggerFactory.getLogger(SensorGridSimulation.class);

    public static final String[] sensors = { "Temperature", "Humidity", "Carbon dioxide", "Pressure", "Oxygen",
            "Anemometer" };
    public static final String[] dataRanges = { "20,30", "60,75", "0,1", "50,65", "88,100", "3,16" };
    public static final String[] topics = { "commonSensor", "commonSensor", "commonSensor", "commonSensor",
            "commonSensor", "commonSensor" };
    // public static final String[] ids = {"3", "4", "5", "6", "7", "8"};
    // //thingspeak
    // public static final String[] ids = {"52516", "52519", "52520", "52521",
    // "52522", "52523"}; //open.sen.se
    public static final String[] ids = { "1", "1", "1", "1", "1", "1" }; // open.sen.se
    public static final String sensorDataStream = "sensor_grid";
    public static final long streamSessionId = System.currentTimeMillis();

    public SensorGridSimulation() {
        // TODO Auto-generated constructor stub
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        SensorGridSimulation mySelf = new SensorGridSimulation();
        try {
            mySelf.launchSensorStreaming(args);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

    protected void launchSensorStreaming(String[] args) throws Exception {
        // DECODING ARGUMENTS FROM COMMAND LINE
        String usage = "usage: --broker-list ip:port --events n";
        // ParseCmd cmd = new ParseCmd.Builder()
        // .help(usage)
        // .parm("--broker-list", "node.stratio.com:9092"
        // ).rex("^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\\-]*[a-zA-Z0-9])\\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\\-]*[A-Za-z0-9]):[0-9]{1,4}+(,(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\\-]*[a-zA-Z0-9])\\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\\-]*[A-Za-z0-9]):[0-9]{1,4}+)*$").req()
        // .parm("--events", "1000" ).req()
        // .build();

        HashMap<String, String> R = new HashMap<String, String>();
        // String parseError = cmd.validate(args);
        // if( cmd.isValid(args) ) {
        // R = (HashMap<String, String>) cmd.parse(args);
        // logger.info("Launching with these params:");
        // logger.info(cmd.displayMap(R));
        // }
        // else {
        // logger.error(parseError);
        // System.exit(1);
        // }

        launchStreamingSensorsData(R.get("--broker-list").toString(), Integer.parseInt(R.get("--events")));

    }

    private void launchStreamingSensorsData(String brokerList, int dataToGenerate) throws Exception {

        CountDownLatch shutdownLatch = new CountDownLatch(sensors.length);
        AtomicInteger globalMessagesSent = new AtomicInteger(0);

        // LAUNCH THROUGHPUT SERVICE
        ExecutorService throughputExecutorService = Executors.newFixedThreadPool(1);
        throughputExecutorService.execute(new ThroughputThread(globalMessagesSent));<commit_msg>deleted old parameters processor lib<commit_after>/*******************************************************************************
 * Copyright 2014 Stratio
 * 
 * Licensed 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
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * 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 com.stratio.streaming.examples;

import java.util.List;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple2;

import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.stratio.streaming.commons.constants.BUS;
import com.stratio.streaming.commons.constants.STREAM_OPERATIONS;
import com.stratio.streaming.commons.messages.ColumnNameTypeValue;
import com.stratio.streaming.commons.messages.StratioStreamingMessage;

public class SensorGridSimulation {

    private static Logger logger = LoggerFactory.getLogger(SensorGridSimulation.class);

    public static final String[] sensors = { "Temperature", "Humidity", "Carbon dioxide", "Pressure", "Oxygen",
            "Anemometer" };
    public static final String[] dataRanges = { "20,30", "60,75", "0,1", "50,65", "88,100", "3,16" };
    public static final String[] topics = { "commonSensor", "commonSensor", "commonSensor", "commonSensor",
            "commonSensor", "commonSensor" };
    // public static final String[] ids = {"3", "4", "5", "6", "7", "8"};
    // //thingspeak
    // public static final String[] ids = {"52516", "52519", "52520", "52521",
    // "52522", "52523"}; //open.sen.se
    public static final String[] ids = { "1", "1", "1", "1", "1", "1" }; // open.sen.se
    public static final String sensorDataStream = "sensor_grid";
    public static final long streamSessionId = System.currentTimeMillis();

    public SensorGridSimulation() {
        // TODO Auto-generated constructor stub
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        SensorGridSimulation mySelf = new SensorGridSimulation();
        try {
            mySelf.launchSensorStreaming(args);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

    protected void launchSensorStreaming(String[] args) throws Exception {

        launchStreamingSensorsData(args[0], Integer.parseInt(args[1]));

    }

    private void launchStreamingSensorsData(String brokerList, int dataToGenerate) throws Exception {

        CountDownLatch shutdownLatch = new CountDownLatch(sensors.length);
        AtomicInteger globalMessagesSent = new AtomicInteger(0);

        // LAUNCH THROUGHPUT SERVICE
        ExecutorService throughputExecutorService = Executors.newFixedThreadPool(1);
        throughputExecutorService.execute(new ThroughputThread(globalMessagesSent));
{'fraction_non_alphanumeric': 0.16748532406852762, 'fraction_numerical': 0.02455972205582844, 'mean_word_length': 3.6274944567627494, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3317223', 'n_tokens_mistral': 2558, 'n_tokens_neox': 2415, 'n_words': 598}