content: /**
   * 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.util.HashMap;
  import java.util.Map;
  import java.util.Random;
  import java.util.Set;
  
  import javax.validation.constraints.Min;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.lib.counters.BasicCounters;
  import org.apache.apex.malhar.lib.util.HighLow;
  import org.apache.commons.lang.mutable.MutableLong;
  
  import com.google.common.base.Strings;
  import com.google.common.collect.Maps;
  import com.google.common.collect.Sets;
  
  import com.datatorrent.api.Context.OperatorContext;
  import com.datatorrent.api.DefaultInputPort;
  import com.datatorrent.api.DefaultOutputPort;
  import com.datatorrent.api.annotation.InputPortFieldAnnotation;
  import com.datatorrent.common.util.BaseOperator;
  
  /**
   * <p>
   * This operator generates the GPS locations for the phone numbers specified.
   * The range of phone numbers or a specific phone number can be set for which the GPS locations will be generated.
   * It supports querying the locations of a given phone number.
   * This is a partionable operator that can partition as the tuplesBlast increases.
   * </p>
   *
   * @since 0.3.2
   */
  public class PhoneMovementGenerator extends BaseOperator
  {
    public final transient DefaultInputPort<Integer> data = new DefaultInputPort<Integer>()
    {
      @Override
      public void process(Integer tuple)
      {
        HighLow<Integer> loc = gps.get(tuple);
        if (loc == null) {
          loc = new HighLow<Integer>(random.nextInt(range), random.nextInt(range));
          gps.put(tuple, loc);
        }
        int xloc = loc.getHigh();
        int yloc = loc.getLow();
        int state = rotate % 4;
  
        // Compute new location
        int delta = random.nextInt(100);
        if (delta >= threshold) {
          if (state < 2) {
            xloc++;
          } else {
            xloc--;
          }
          if (xloc < 0) {
            xloc += range;
          }
        }
        delta = random.nextInt(100);
        if (delta >= threshold) {
          if ((state == 1) || (state == 3)) {
            yloc++;
          } else {
            yloc--;
          }
          if (yloc < 0) {
            yloc += range;
          }
        }
        xloc %= range;
        yloc %= range;
  
        // Set new location
        HighLow<Integer> nloc = newgps.get(tuple);
        if (nloc == null) {
          newgps.put(tuple, new HighLow<Integer>(xloc, yloc));
        } else {
          nloc.setHigh(xloc);
          nloc.setLow(yloc);
        }
        rotate++;
      }
    };
  
    @InputPortFieldAnnotation(optional = true)
    public final transient DefaultInputPort<Map<String,String>> phoneQuery = new DefaultInputPort<Map<String,String>>()
    {
      @Override
      public void process(Map<String,String> tuple)
      {
        LOG.info("new query {}", tuple);
        String command = tuple.get(KEY_COMMAND);
        if (command != null) {
          if (command.equals(COMMAND_ADD)) {
            commandCounters.getCounter(CommandCounters.ADD).increment();
            String phoneStr = tuple.get(KEY_PHONE);
            registerPhone(phoneStr);
          } else if (command.equals(COMMAND_ADD_RANGE)) {
            commandCounters.getCounter(CommandCounters.ADD_RANGE).increment();
            registerPhoneRange(tuple.get(KEY_START_PHONE), tuple.get(KEY_END_PHONE));
          } else if (command.equals(COMMAND_DELETE)) {
            commandCounters.getCounter(CommandCounters.DELETE).increment();
            String phoneStr = tuple.get(KEY_PHONE);
            deregisterPhone(phoneStr);
          } else if (command.equals(COMMAND_CLEAR)) {
            commandCounters.getCounter(CommandCounters.CLEAR).increment();
            clearPhones();
          }
        }
      }
    };
  
    public static final String KEY_COMMAND = "command";
    public static final String KEY_PHONE = "phone";
    public static final String KEY_LOCATION = "location";
    public static final String KEY_REMOVED = "removed";
    public static final String KEY_START_PHONE = "startPhone";
    public static final String KEY_END_PHONE = "endPhone";
  
    public static final String COMMAND_ADD = "add";
    public static final String COMMAND_ADD_RANGE = "addRange";
    public static final String COMMAND_DELETE = "del";
    public static final String COMMAND_CLEAR = "clear";
  
    final Set<Integer> phoneRegister = Sets.newHashSet();
  
    private final transient HashMap<Integer, HighLow<Integer>> gps = new HashMap<Integer, HighLow<Integer>>();
    private final Random random = new Random();
    private int range = 50;
    private int threshold = 80;
    private int rotate = 0;
  
    protected BasicCounters<MutableLong> commandCounters;
  
    private transient OperatorContext context;
    private final transient HashMap<Integer, HighLow<Integer>> newgps = new HashMap<Integer, HighLow<Integer>>();
  
    public PhoneMovementGenerator()
    {
      this.commandCounters = new BasicCounters<MutableLong>(MutableLong.class);
    }
  
    /**
     * @return the range of the phone numbers
     */
    @Min(0)
    public int getRange()
    {
      return range;
    }
  
    /**
     * Sets the range of phone numbers for which the GPS locations need to be generated.
     *
     * @param i the range of phone numbers to set
     */
    public void setRange(int i)
    {
      range = i;
    }
  
    /**
     * @return the threshold
     */
    @Min(0)
    public int getThreshold()
    {
      return threshold;
    }
  
    /**
     * Sets the threshold that decides how frequently the GPS locations are updated.
     *
     * @param i the value that decides how frequently the GPS locations change.
     */
    public void setThreshold(int i)
    {
      threshold = i;
    }
  
    private void registerPhone(String phoneStr)
    {
      // register the phone channel
      if (Strings.isNullOrEmpty(phoneStr)) {
        return;
      }
      try {
        Integer phone = new Integer(phoneStr);
        registerSinglePhone(phone);
      } catch (NumberFormatException nfe) {
        LOG.warn("Invalid no {}", phoneStr);
      }
    }
  
    private void registerPhoneRange(String startPhoneStr, String endPhoneStr)
    {
      if (Strings.isNullOrEmpty(startPhoneStr) || Strings.isNullOrEmpty(endPhoneStr)) {
        LOG.warn("Invalid phone range {} {}", startPhoneStr, endPhoneStr);
        return;
      }
      try {
        Integer startPhone = new Integer(startPhoneStr);
        Integer endPhone = new Integer(endPhoneStr);
        if (endPhone < startPhone) {
          LOG.warn("Invalid phone range {} {}", startPhone, endPhone);
          return;
        }
        for (int i = startPhone; i <= endPhone; i++) {
          registerSinglePhone(i);
        }
      } catch (NumberFormatException nfe) {
        LOG.warn("Invalid phone range <{},{}>", startPhoneStr, endPhoneStr);
      }
    }
  
    private void registerSinglePhone(int phone)
    {
      phoneRegister.add(phone);
      LOG.debug("Registered query id with phone {}", phone);
      emitQueryResult(phone);
    }
  
    private void deregisterPhone(String phoneStr)
    {
      if (Strings.isNullOrEmpty(phoneStr)) {
        return;
      }
      try {
        Integer phone = new Integer(phoneStr);
        // remove the channel
        if (phoneRegister.contains(phone)) {
          phoneRegister.remove(phone);
          LOG.debug("Removing query id {}", phone);
          emitPhoneRemoved(phone);
        }
      } catch (NumberFormatException nfe) {
        LOG.warn("Invalid phone {}", phoneStr);
      }
    }
  
    private void clearPhones()
    {
      phoneRegister.clear();
      LOG.info("Clearing phones");
    }
  
    public final transient DefaultOutputPort<Map<String, String>> locationQueryResult = new DefaultOutputPort<Map<String, String>>();
  
    @Override
    public void setup(OperatorContext context)
    {
      this.context = context;
      commandCounters.setCounter(CommandCounters.ADD, new MutableLong());
      commandCounters.setCounter(CommandCounters.ADD_RANGE, new MutableLong());
      commandCounters.setCounter(CommandCounters.DELETE, new MutableLong());
      commandCounters.setCounter(CommandCounters.CLEAR, new MutableLong());
    }
  
    /**
     * Emit all the data and clear the hash
     */
    @Override
    public void endWindow()
    {
      for (Map.Entry<Integer, HighLow<Integer>> e: newgps.entrySet()) {
        HighLow<Integer> loc = gps.get(e.getKey());
        if (loc == null) {
          gps.put(e.getKey(), e.getValue());
        } else {
          loc.setHigh(e.getValue().getHigh());
          loc.setLow(e.getValue().getLow());
        }
      }
      boolean found = false;
      for (Integer phone: phoneRegister) {
        emitQueryResult( phone);
        found = true;
      }
      if (!found) {
        LOG.debug("No phone number");
      }
      newgps.clear();
      context.setCounters(commandCounters);
    }
  
    private void emitQueryResult(Integer phone)
    {
      HighLow<Integer> loc = gps.get(phone);
      if (loc != null) {
        Map<String, String> queryResult = new HashMap<String, String>();
        queryResult.put(KEY_PHONE, String.valueOf(phone));
        queryResult.put(KEY_LOCATION, loc.toString());
        locationQueryResult.emit(queryResult);
      }
    }
  
    private void emitPhoneRemoved(Integer phone)
    {
      Map<String,String> removedResult = Maps.newHashMap();
      removedResult.put(KEY_PHONE, String.valueOf(phone));
      removedResult.put(KEY_REMOVED,"true");
      locationQueryResult.emit(removedResult);
    }
  
    public static enum CommandCounters
    {
      ADD, ADD_RANGE, DELETE, CLEAR
    }
  
    private static final Logger LOG = LoggerFactory.getLogger(PhoneMovementGenerator.class);
  }
  