examples/mobile/src/main/java/org/apache/apex/examples/mobile/PhoneMovementGenerator.java - apex-malhar - Git at Google

 /**
  * 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);
 }
	/**
	* 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);
	}