/**
 * 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
 *
 *   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 org.apache.apex.examples.mobile;

import java.net.URI;
import java.util.Arrays;
import java.util.Map;
import java.util.Random;

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

import org.apache.apex.malhar.lib.counters.BasicCounters;
import org.apache.apex.malhar.lib.io.PubSubWebSocketInputOperator;
import org.apache.apex.malhar.lib.io.PubSubWebSocketOutputOperator;
import org.apache.apex.malhar.lib.partitioner.StatelessThroughputBasedPartitioner;
import org.apache.apex.malhar.lib.testbench.RandomEventGenerator;
import org.apache.apex.malhar.lib.utils.PubSubHelper;
import org.apache.commons.lang.mutable.MutableLong;
import org.apache.commons.lang3.Range;
import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DAG;
import com.datatorrent.api.StatsListener;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.annotation.ApplicationAnnotation;

/**
 * Mobile Example Application:
 * <p>
 * This example simulates large number of cell phones in the range of 40K to 200K
 * and tracks a given cell number across cell towers. It also displays the changing locations of the cell number on a google map.
 *
 * This example demonstrates the scalability feature of the Apex platform.
 * It showcases the ability of the platform to scale up and down as the phone numbers generated increase and decrease respectively.
 * If the tuples processed per second by the pmove operator increase beyond 30,000, more partitions of the pmove operator gets deployed until
 * each of the partition processes around 10000 to 30000 tuples per second.
 * If the tuples processed per second drops below 10,000, the platform merges the operators until the partition count drops down to the original.
 * The load can be varied using the tuplesBlast property.
 * If the tuplesBlast is set to 200, 40K cell phones are generated.
 * If the tuplesBlast is set to 1000, 200K cell phones are generated.
 * The tuplesBlast property can be set using dtcli command: 'set-operator-property pmove tuplesBlast 1000'.
 *
 *
 * The specs are as such<br>
 * Depending on the tuplesBlast property, large number of cell phone numbers are generated.
 * They jump a cell tower frequently. Sometimes
 * within a second sometimes in 10 seconds. The aim is to demonstrate the
 * following abilities<br>
 * <ul>
 * <li>Entering query dynamically: The phone numbers are added to locate its gps
 * in run time.</li>
 * <li>Changing functionality dynamically: The load is changed by making
 * functional changes on the load generator operator (phonegen)(</li>
 * <li>Auto Scale up/Down with load: Operator pmove increases and decreases
 * partitions as per load</li>
 * <li></li>
 * </ul>
 *
 * Refer to examples/docs/MobileDemo.md for more information.
 *
 * <p>
 *
 * Running Java Test or Main app in IDE:
 *
 * <pre>
 * LocalMode.runApp(new Application(), 600000); // 10 min run
 * </pre>
 *
 * Run Success : <br>
 * For successful deployment and run, user should see following output on
 * console: <br>
 *
 * <pre>
 * phoneLocationQueryResult: {phone=5556101, location=(5,9), queryId=q3}
 * phoneLocationQueryResult: {phone=5554995, location=(10,4), queryId=q1}
 * phoneLocationQueryResult: {phone=5556101, location=(5,9), queryId=q3}
 * phoneLocationQueryResult: {phone=5554995, location=(10,4), queryId=q1}
 * phoneLocationQueryResult: {phone=5554995, location=(10,5), queryId=q1}
 * phoneLocationQueryResult: {phone=5556101, location=(5,9), queryId=q3}
 * phoneLocationQueryResult: {phone=5554995, location=(9,5), queryId=q1}
 * phoneLocationQueryResult: {phone=5556101, location=(5,9), queryId=q3}
 * phoneLocationQueryResult: {phone=5556101, location=(5,9), queryId=q3}
 * phoneLocationQueryResult: {phone=5554995, location=(9,5), queryId=q1}
 * phoneLocationQueryResult: {phone=5554995, location=(9,5), queryId=q1}
 * phoneLocationQueryResult: {phone=5556101, location=(5,9), queryId=q3}
 * </pre>
 *
 * * <b>Application DAG : </b><br>
 * <img src="doc-files/mobile.png" width=600px > <br>
 *
 * @since 0.3.2
 */
@ApplicationAnnotation(name = "MobileExample")
public class Application implements StreamingApplication
{
  public static final String PHONE_RANGE_PROP = "dt.application.MobileExample.phoneRange";
  public static final String TOTAL_SEED_NOS = "dt.application.MobileExample.totalSeedNumbers";
  public static final String COOL_DOWN_MILLIS = "dt.application.MobileExample.coolDownMillis";
  public static final String MAX_THROUGHPUT = "dt.application.MobileExample.maxThroughput";
  public static final String MIN_THROUGHPUT = "dt.application.MobileExample.minThroughput";
  private static final Logger LOG = LoggerFactory.getLogger(Application.class);
  private Range<Integer> phoneRange = Range.between(5550000, 5559999);

  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {
    String lPhoneRange = conf.get(PHONE_RANGE_PROP, null);
    if (lPhoneRange != null) {
      String[] tokens = lPhoneRange.split("-");
      if (tokens.length != 2) {
        throw new IllegalArgumentException("Invalid range: " + lPhoneRange);
      }
      this.phoneRange = Range.between(Integer.parseInt(tokens[0]), Integer.parseInt(tokens[1]));
    }
    LOG.debug("Phone range {}", this.phoneRange);

    RandomEventGenerator phones = dag.addOperator("Receiver", RandomEventGenerator.class);
    phones.setMinvalue(this.phoneRange.getMinimum());
    phones.setMaxvalue(this.phoneRange.getMaximum());

    PhoneMovementGenerator movementGen = dag.addOperator("LocationFinder", PhoneMovementGenerator.class);
    dag.setAttribute(movementGen, OperatorContext.COUNTERS_AGGREGATOR, new BasicCounters.LongAggregator<MutableLong>());

    StatelessThroughputBasedPartitioner<PhoneMovementGenerator> partitioner = new StatelessThroughputBasedPartitioner<PhoneMovementGenerator>();
    partitioner.setCooldownMillis(conf.getLong(COOL_DOWN_MILLIS, 45000));
    partitioner.setMaximumEvents(conf.getLong(MAX_THROUGHPUT, 30000));
    partitioner.setMinimumEvents(conf.getLong(MIN_THROUGHPUT, 10000));
    dag.setAttribute(movementGen, OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{partitioner}));
    dag.setAttribute(movementGen, OperatorContext.PARTITIONER, partitioner);

    // generate seed numbers
    Random random = new Random();
    int maxPhone = phoneRange.getMaximum() - phoneRange.getMinimum();
    int phonesToDisplay = conf.getInt(TOTAL_SEED_NOS, 10);
    for (int i = phonesToDisplay; i-- > 0; ) {
      int phoneNo = phoneRange.getMinimum() + random.nextInt(maxPhone + 1);
      LOG.info("seed no: " + phoneNo);
      movementGen.phoneRegister.add(phoneNo);
    }
    // done generating data
    LOG.info("Finished generating seed data.");

    URI uri = PubSubHelper.getURI(dag);
    PubSubWebSocketOutputOperator<Object> wsOut = dag.addOperator("LocationResults", new PubSubWebSocketOutputOperator<Object>());
    wsOut.setUri(uri);
    PubSubWebSocketInputOperator<Map<String, String>> wsIn = dag.addOperator("QueryLocation", new PubSubWebSocketInputOperator<Map<String, String>>());
    wsIn.setUri(uri);
    // default partitioning: first connected stream to movementGen will be partitioned
    dag.addStream("Phone-Data", phones.integer_data, movementGen.data);
    dag.addStream("Results", movementGen.locationQueryResult, wsOut.input);
    dag.addStream("Query", wsIn.outputPort, movementGen.phoneQuery);
  }

}