/*
 * Copyright (c) 2013 DataTorrent, Inc. ALL Rights Reserved.
 *
 * 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.datatorrent.demos.uniquecountdemo;

import com.datatorrent.api.Context;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.InputOperator;
import com.datatorrent.lib.util.KeyValPair;

import java.util.HashMap;
import java.util.Map;

/**
 *
 */
public class RandomDataGenerator implements InputOperator
{
  public final transient DefaultOutputPort<KeyValPair<String, Object>> outPort = new DefaultOutputPort<KeyValPair<String, Object>>();
  private HashMap<String, Integer> dataInfo;

  public RandomDataGenerator()
  {
    this.dataInfo = new HashMap<String, Integer>();
    this.dataInfo.put("a", 1000);
    this.dataInfo.put("b", 1000);
    this.dataInfo.put("c", 1000);
    this.dataInfo.put("d", 1000);
  }

  @Override
  public void emitTuples()
  {
    for (Map.Entry<String, Integer> e : dataInfo.entrySet()) {
      String key = e.getKey();
      int count = e.getValue();
      for (int i = 0; i < count; ++i) {
        outPort.emit(new KeyValPair<String, Object>(key, i));
      }
    }
  }

  @Override
  public void beginWindow(long l)
  {

  }

  @Override
  public void endWindow()
  {

  }

  @Override
  public void setup(Context.OperatorContext operatorContext)
  {

  }

  @Override
  public void teardown()
  {

  }
}

@@ -877,19 +877,22 @@
 va.util.
-Map
+Random
 ;%0A%0A/**%0A 
@@ -892,16 +892,128 @@
 %0A%0A/**%0A *
+ Generate random Key value pairs.%0A * key is string and value is int, it emits the pair as KeyValPair on outPort,
 %0A */%0Apub
@@ -1252,521 +1252,914 @@
 fo;%0A
-%0A
   p
-ublic RandomDataGenerator()%0A  %7B%0A    this.dataInfo = new HashMap%3CString, Integer%3E();%0A    this.dataInfo.put(%22a%22,
+rivate int count;%0A  private int sleepMs = 10;%0A  private int keyRange = 100;%0A  private int valueRange =
  100
-0
-)
 ;%0A  
-  this.dataInfo.put(%22b%22,
+private long tupleBlast =
  1000
-)
+0
 ;%0A  
-  this.dataInfo.put(%22c%22, 1000);%0A    this.dataInfo.put(%22d%22, 1000);%0A  %7D%0A%0A  @Override%0A  public void emitTuples()%0A  %7B%0A    for (Map.Entry%3CString, Integer%3E e : dataInfo.entrySet()) %7B%0A      String key = e.getKey(
+private Random random;%0A%0A  public RandomDataGenerator()%0A  %7B%0A    random = new Random();%0A  %7D%0A%0A  @Override%0A  public void emitTuples()%0A  %7B%0A    for(int i = 0 ; i %3C tupleBlast; i++) %7B%0A      String key = String.valueOf(random.nextInt(keyRange));%0A      int val = random.nextInt(valueRange);%0A      outPort.emit(new KeyValPair%3CString, Object%3E(key, val));%0A    %7D%0A    try %7B%0A      Thread.sleep(sleepMs);%0A    %7D catch(Exception ex) %7B%0A      System.out.println(ex.getMessage()
 );%0A    
-  int
+%7D%0A   
  count
- = e.getValue();%0A      for (int i = 0; i %3C count; ++i) %7B%0A        outPort.emit(new KeyValPair%3CString, Object%3E(key, i));
+++;%0A  %7D%0A%0A  public int getSleepMs()%0A  %7B%0A    return sleepMs;%0A  %7D%0A%0A  public void setSleepMs(int sleepMs)%0A  %7B%0A    this.sleepMs = sleepMs;%0A  %7D%0A%0A  public long getTupleBlast()%0A  %7B%0A    return tupleBlast;%0A  %7D%0A%0A  public void setTupleBlast(long tupleBlast)
 %0A  
+%7B%0A
     
-%7D%0A    %7D
+this.tupleBlast = tupleBlast;
 %0A  %7D
@@ -2254,24 +2254,119 @@
 indow()%0A  %7B%0A
+    System.out.println(%22emitTuples called  %22 + count + %22 times in this window%22);%0A    count = 0;
 %0A  %7D%0A%0A  @Ove

Generate key value pairs randomly