text:
package org.apache.tez.history;

import com.google.common.collect.Sets;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.server.namenode.EditLogFileOutputStream;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.tez.client.CallerContext;
import org.apache.tez.client.TezClient;
import org.apache.tez.common.counters.DAGCounter;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.dag.api.DAG;
import org.apache.tez.dag.api.DataSinkDescriptor;
import org.apache.tez.dag.api.DataSourceDescriptor;
import org.apache.tez.dag.api.Edge;
import org.apache.tez.dag.api.EdgeProperty;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
import org.apache.tez.dag.api.Vertex;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.StatusGetOpts;
import org.apache.tez.dag.api.event.VertexState;
import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
import org.apache.tez.dag.api.oldrecords.TaskState;
import org.apache.tez.dag.app.dag.DAGState;
import org.apache.tez.dag.history.logging.ats.ATSHistoryLoggingService;
import org.apache.tez.dag.history.logging.impl.SimpleHistoryLoggingService;
import org.apache.tez.dag.records.TezDAGID;
import org.apache.tez.examples.WordCount;
import org.apache.tez.history.parser.ATSFileParser;
import org.apache.tez.history.parser.SimpleHistoryParser;
import org.apache.tez.history.parser.datamodel.BaseInfo;
import org.apache.tez.history.parser.datamodel.DagInfo;
import org.apache.tez.history.parser.datamodel.EdgeInfo;
import org.apache.tez.history.parser.datamodel.TaskAttemptInfo;
import org.apache.tez.history.parser.datamodel.TaskAttemptInfo.DataDependencyEvent;
import org.apache.tez.history.parser.datamodel.TaskInfo;
import org.apache.tez.history.parser.datamodel.VersionInfo;
import org.apache.tez.history.parser.datamodel.VertexInfo;
import org.apache.tez.mapreduce.input.MRInput;
import org.apache.tez.mapreduce.output.MROutput;
import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
import org.apache.tez.runtime.api.ProcessorContext;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
import org.apache.tez.runtime.library.input.OrderedGroupedKVInput;
import org.apache.tez.runtime.library.output.OrderedPartitionedKVOutput;
import org.apache.tez.runtime.library.partitioner.HashPartitioner;
import org.apache.tez.tests.MiniTezClusterWithTimeline;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

public class TestHistoryParser {

  private static MiniDFSCluster miniDFSCluster;
  private static MiniTezClusterWithTimeline miniTezCluster;

  //location within miniHDFS cluster's hdfs
  private static Path inputLoc = new Path("/tmp/sample.txt");

  private final static String INPUT = "Input";
  private final static String OUTPUT = "Output";
  private final static String TOKENIZER = "Tokenizer";
  private final static String SUMMATION = "Summation";
  private final static String SIMPLE_HISTORY_DIR = "/tmp/simplehistory/";
  private final static String HISTORY_TXT = "history.txt";

  private static Configuration conf = new Configuration();
  private static FileSystem fs;
  private static String TEST_ROOT_DIR =
      "target" + Path.SEPARATOR + TestHistoryParser.class.getName() + "-tmpDir";
  private static String TEZ_BASE_DIR =
      "target" + Path.SEPARATOR + TestHistoryParser.class.getName() + "-tez";
  private static String DOWNLOAD_DIR = TEST_ROOT_DIR + Path.SEPARATOR + "download";

  @BeforeClass
  public static void setupCluster() throws Exception {
    conf = new Configuration();
    conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH, false);
    EditLogFileOutputStream.setShouldSkipFsyncForTesting(true);
    conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
    miniDFSCluster =
        new MiniDFSCluster.Builder(conf).numDataNodes(1).format(true).build();
    fs = miniDFSCluster.getFileSystem();
    conf.set("fs.defaultFS", fs.getUri().toString());

    setupTezCluster();
  }

  @AfterClass
  public static void shutdownCluster() {
    try {
      if (miniDFSCluster != null) {
        miniDFSCluster.shutdown();
      }
      if (miniTezCluster != null) {
        miniTezCluster.stop();
      }
    } finally {
      try {
        FileUtils.deleteDirectory(new File(TEST_ROOT_DIR));
        FileUtils.deleteDirectory(new File(TEZ_BASE_DIR));
      } catch (IOException e) {
        //safe to ignore
      }
    }
  }

  // @Before
  public static void setupTezCluster() throws Exception {
    conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT, 3 * 1000);
    conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT, 3 * 1000);
    conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT, 2);

    //Enable per edge counters
    conf.setBoolean(TezConfiguration.TEZ_TASK_GENERATE_COUNTERS_PER_IO, true);
    conf.setBoolean(TezConfiguration.TEZ_AM_ALLOW_DISABLED_TIMELINE_DOMAINS, true);
    conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
    conf.set(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS, ATSHistoryLoggingService
        .class.getName());

    conf.set(TezConfiguration.TEZ_SIMPLE_HISTORY_LOGGING_DIR, SIMPLE_HISTORY_DIR);

    miniTezCluster =
        new MiniTezClusterWithTimeline(TEZ_BASE_DIR, 1, 1, 1, true);

    miniTezCluster.init(conf);
    miniTezCluster.start();

    createSampleFile(inputLoc);

    TezConfiguration tezConf = new TezConfiguration(miniTezCluster.getConfig());
    tezConf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
    tezConf.set(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS, "0.0.0.0:8188");
    tezConf.setBoolean(TezConfiguration.TEZ_AM_ALLOW_DISABLED_TIMELINE_DOMAINS, true);
    tezConf.set(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS,
        ATSHistoryLoggingService.class.getName());

  }

  /**
   * Run a word count example in mini cluster and check if it is possible to download
   * data from ATS and parse it. Also, run with SimpleHistoryLogging option and verify
   * if it matches with ATS data.
   *
   * @throws Exception
   */
  @Test
  public void testParserWithSuccessfulJob() throws Exception {
    //Run basic word count example.
    String dagId = runWordCount(WordCount.TokenProcessor.class.getName(),
        WordCount.SumProcessor.class.getName(), "WordCount", true);

    //Export the data from ATS
    String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR };

    int result = ATSImportTool.process(args);
    assertTrue(result == 0);

    //Parse ATS data and verify results
    DagInfo dagInfoFromATS = getDagInfo(dagId);
    verifyDagInfo(dagInfoFromATS, true);
    verifyJobSpecificInfo(dagInfoFromATS);

    //Now run with SimpleHistoryLogging
    dagId = runWordCount(WordCount.TokenProcessor.class.getName(),
        WordCount.SumProcessor.class.getName(), "WordCount", false);
    Thread.sleep(10000); //For all flushes to happen and to avoid half-cooked download.

    DagInfo shDagInfo = getDagInfoFromSimpleHistory(dagId);
    verifyDagInfo(shDagInfo, false);
    verifyJobSpecificInfo(shDagInfo);

    //Compare dagInfo by parsing ATS data with DagInfo obtained by parsing SimpleHistoryLog
    isDAGEqual(dagInfoFromATS, shDagInfo);
  }

  private DagInfo getDagInfoFromSimpleHistory(String dagId) throws TezException, IOException {
    TezDAGID tezDAGID = TezDAGID.fromString(dagId);
    ApplicationAttemptId applicationAttemptId = ApplicationAttemptId.newInstance(tezDAGID
        .getApplicationId(), 1);
    Path historyPath = new Path(conf.get("fs.defaultFS")
        + SIMPLE_HISTORY_DIR + HISTORY_TXT + "."
        + applicationAttemptId);
    FileSystem fs = historyPath.getFileSystem(conf);

    Path localPath = new Path(DOWNLOAD_DIR, HISTORY_TXT);
    fs.copyToLocalFile(historyPath, localPath);
    File localFile = new File(DOWNLOAD_DIR, HISTORY_TXT);

    //Now parse via SimpleHistory
    SimpleHistoryParser parser = new SimpleHistoryParser(localFile);
    DagInfo dagInfo = parser.getDAGData(dagId);
    assertTrue(dagInfo.getDagId().equals(dagId));
    return dagInfo;
  }

  private void verifyJobSpecificInfo(DagInfo dagInfo) {
    //Job specific
    assertTrue(dagInfo.getNumVertices() == 2);
    assertTrue(dagInfo.getName().equals("WordCount"));
    assertTrue(dagInfo.getVertex(TOKENIZER).getProcessorClassName().equals(
        WordCount.TokenProcessor.class.getName()));
    assertTrue(dagInfo.getVertex(SUMMATION).getProcessorClassName()
        .equals(WordCount.SumProcessor.class.getName()));
    assertTrue(dagInfo.getEdges().size() == 1);
    EdgeInfo edgeInfo = dagInfo.getEdges().iterator().next();
    assertTrue(edgeInfo.getDataMovementType().
        equals(EdgeProperty.DataMovementType.SCATTER_GATHER.toString()));
    assertTrue(edgeInfo.getSourceVertex().getVertexName().equals(TOKENIZER));
    assertTrue(edgeInfo.getDestinationVertex().getVertexName().equals(SUMMATION));
    assertTrue(edgeInfo.getInputVertexName().equals(TOKENIZER));
    assertTrue(edgeInfo.getOutputVertexName().equals(SUMMATION));
    assertTrue(edgeInfo.getEdgeSourceClass().equals(OrderedPartitionedKVOutput.class.getName()));
    assertTrue(edgeInfo.getEdgeDestinationClass().equals(OrderedGroupedKVInput.class.getName()));
    assertTrue(dagInfo.getVertices().size() == 2);
    String lastSourceTA = null;
    String lastDataEventSourceTA = null;
    for (VertexInfo vertexInfo : dagInfo.getVertices()) {
      assertTrue(vertexInfo.getKilledTasksCount() == 0);
      assertTrue(vertexInfo.getInitRequestedTime() > 0);
      assertTrue(vertexInfo.getInitTime() > 0);
      assertTrue(vertexInfo.getStartRequestedTime() > 0);
      assertTrue(vertexInfo.getStartTime() > 0);
      assertTrue(vertexInfo.getFinishTime() > 0);
      long finishTime = 0;
      for (TaskInfo taskInfo : vertexInfo.getTasks()) {
        assertTrue(taskInfo.getNumberOfTaskAttempts() == 1);
        assertTrue(taskInfo.getMaxTaskAttemptDuration() >= 0);
        assertTrue(taskInfo.getMinTaskAttemptDuration() >= 0);
        assertTrue(taskInfo.getAvgTaskAttemptDuration() >= 0);
        assertTrue(taskInfo.getLastTaskAttemptToFinish() != null);
        assertTrue(taskInfo.getContainersMapping().size() > 0);
        assertTrue(taskInfo.getSuccessfulTaskAttempts().size() > 0);
        assertTrue(taskInfo.getFailedTaskAttempts().size() == 0);
        assertTrue(taskInfo.getKilledTaskAttempts().size() == 0);
        List<TaskAttemptInfo> attempts = taskInfo.getTaskAttempts();
        if (vertexInfo.getVertexName().equals(TOKENIZER)) {
          // get the last task to finish and track its successful attempt
          if (finishTime < taskInfo.getFinishTime()) {
            finishTime = taskInfo.getFinishTime();
            lastSourceTA = taskInfo.getSuccessfulAttemptId();
          }
        } else {
          for (TaskAttemptInfo attempt : attempts) {
            DataDependencyEvent item = attempt.getLastDataEvents().get(0);
            assertTrue(item.getTimestamp() > 0);

            if (lastDataEventSourceTA == null) {
              lastDataEventSourceTA = item.getTaskAttemptId();
            } else {
              // all attempts should have the same last data event source TA
              assertTrue(lastDataEventSourceTA.equals(item.getTaskAttemptId()));
            }
          }
        }
        for (TaskAttemptInfo attemptInfo : taskInfo.getTaskAttempts()) {
          assertTrue(attemptInfo.getCreationTime() > 0);
          assertTrue(attemptInfo.getAllocationTime() > 0);
          assertTrue(attemptInfo.getStartTime() > 0);
        }
      }
      assertTrue(vertexInfo.getLastTaskToFinish() != null);
      if (vertexInfo.getVertexName().equals(TOKENIZER)) {
        assertTrue(vertexInfo.getInputEdges().size() == 0);
        assertTrue(vertexInfo.getOutputEdges().size() == 1);
        assertTrue(vertexInfo.getOutputVertices().size() == 1);
        assertTrue(vertexInfo.getInputVertices().size() == 0);
      } else {
        assertTrue(vertexInfo.getInputEdges().size() == 1);
        assertTrue(vertexInfo.getOutputEdges().size() == 0);
        assertTrue(vertexInfo.getOutputVertices().size() == 0);
        assertTrue(vertexInfo.getInputVertices().size() == 1);
      }
    }
    assertTrue(lastSourceTA.equals(lastDataEventSourceTA));
  }

  /**
   * Run a word count example in mini cluster.
   * Provide invalid URL for ATS.
   *
   * @throws Exception
   */
  @Test
  public void testParserWithSuccessfulJob_InvalidATS() throws Exception {
    //Run basic word count example.
    String dagId =  runWordCount(WordCount.TokenProcessor.class.getName(),
        WordCount.SumProcessor.class.getName(), "WordCount-With-WrongATS-URL", true);

    //Export the data from ATS
    String atsAddress = "--atsAddress=http://atsHost:8188";
    String[] args = { "--dagId=" + dagId,
        "--downloadDir=" + DOWNLOAD_DIR,
        atsAddress
      };

    int result = ATSImportTool.process(args);
    assertTrue(result == -1);
  }

  /**
   * Run a failed job and parse the data from ATS
   */
  @Test
  public void testParserWithFailedJob() throws Exception {
    //Run a job which would fail
    String dagId = runWordCount(WordCount.TokenProcessor.class.getName(), FailProcessor.class
        .getName(), "WordCount-With-Exception", true);

    //Export the data from ATS
    String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR };

    int result = ATSImportTool.process(args);
    assertTrue(result == 0);

    //Parse ATS data
    DagInfo dagInfo = getDagInfo(dagId);

    //Verify DAGInfo. Verifies vertex, task, taskAttempts in recursive manner
    verifyDagInfo(dagInfo, true);

    //Dag specific
    VertexInfo summationVertex = dagInfo.getVertex(SUMMATION);
    assertTrue(summationVertex.getFailedTasks().size() == 1); //1 task, 4 attempts failed
    assertTrue(summationVertex.getFailedTasks().get(0).getFailedTaskAttempts().size() == 4);
    assertTrue(summationVertex.getStatus().equals(VertexState.FAILED.toString()));

    assertTrue(dagInfo.getFailedVertices().size() == 1);
    assertTrue(dagInfo.getFailedVertices().get(0).getVertexName().equals(SUMMATION));
    assertTrue(dagInfo.getSuccessfullVertices().size() == 1);
    assertTrue(dagInfo.getSuccessfullVertices().get(0).getVertexName().equals(TOKENIZER));

    assertTrue(dagInfo.getStatus().equals(DAGState.FAILED.toString()));

    verifyCounter(dagInfo.getCounter(DAGCounter.NUM_FAILED_TASKS.toString()), null, 4);
    verifyCounter(dagInfo.getCounter(DAGCounter.NUM_SUCCEEDED_TASKS.toString()), null, 1);
    verifyCounter(dagInfo.getCounter(DAGCounter.TOTAL_LAUNCHED_TASKS.toString()), null, 5);

    verifyCounter(dagInfo.getCounter(TaskCounter.INPUT_RECORDS_PROCESSED.toString()),
        "TaskCounter_Tokenizer_INPUT_Input", 10);
    verifyCounter(dagInfo.getCounter(TaskCounter.ADDITIONAL_SPILLS_BYTES_READ.toString()),
        "TaskCounter_Tokenizer_OUTPUT_Summation", 0);
    verifyCounter(dagInfo.getCounter(TaskCounter.OUTPUT_RECORDS.toString()),
        "TaskCounter_Tokenizer_OUTPUT_Summation",
        20); //Every line has 2 words. 10 lines x 2 words = 20
    verifyCounter(dagInfo.getCounter(TaskCounter.SPILLED_RECORDS.toString()),
        "TaskCounter_Tokenizer_OUTPUT_Summation", 20); //Same as above

    for (TaskInfo taskInfo : summationVertex.getTasks()) {
      TaskAttemptInfo lastAttempt = null;
      for (TaskAttemptInfo attemptInfo : taskInfo.getTaskAttempts()) {
        if (lastAttempt != null) {
          // failed attempt should be causal TA of next attempt
          assertTrue(lastAttempt.getTaskAttemptId().equals(attemptInfo.getCreationCausalTA()));
          assertTrue(lastAttempt.getTerminationCause() != null);
        }
        lastAttempt = attemptInfo;
      }
    }

    //TODO: Need to check for SUMMATION vertex counters. Since all attempts are failed, counters are not getting populated.
    //TaskCounter.REDUCE_INPUT_RECORDS

    //Verify if the processor exception is given in diagnostics
    assertTrue(dagInfo.getDiagnostics().contains("Failing this processor for some reason"));

  }

  /**
   * Adding explicit equals here instead of in DAG/Vertex/Edge where hashCode also needs to
   * change. Also, some custom comparisons are done here for unit testing.
   */
  private void isDAGEqual(DagInfo dagInfo1, DagInfo dagInfo2) {
    assertNotNull(dagInfo1);
    assertNotNull(dagInfo2);
    assertEquals(dagInfo1.getStatus(), dagInfo2.getStatus());
    isEdgeEqual(dagInfo1.getEdges(), dagInfo2.getEdges());
    isVertexEqual(dagInfo1.getVertices(), dagInfo2.getVertices());
  }

  private void isVertexEqual(VertexInfo vertexInfo1, VertexInfo vertexInfo2) {
    assertTrue(vertexInfo1 != null);
    assertTrue(vertexInfo2 != null);
    assertTrue(vertexInfo1.getVertexName().equals(vertexInfo2.getVertexName()));
    assertTrue(vertexInfo1.getProcessorClassName().equals(vertexInfo2.getProcessorClassName()));
    assertTrue(vertexInfo1.getNumTasks() == vertexInfo2.getNumTasks());
    assertTrue(vertexInfo1.getCompletedTasksCount() == vertexInfo2.getCompletedTasksCount());
    assertTrue(vertexInfo1.getStatus().equals(vertexInfo2.getStatus()));

    isEdgeEqual(vertexInfo1.getInputEdges(), vertexInfo2.getInputEdges());
    isEdgeEqual(vertexInfo1.getOutputEdges(), vertexInfo2.getOutputEdges());

    assertTrue(vertexInfo1.getInputVertices().size() == vertexInfo2.getInputVertices().size());
    assertTrue(vertexInfo1.getOutputVertices().size() == vertexInfo2.getOutputVertices().size());

    assertTrue(vertexInfo1.getNumTasks() == vertexInfo2.getNumTasks());
    isTaskEqual(vertexInfo1.getTasks(), vertexInfo2.getTasks());
  }

  private void isVertexEqual(List<VertexInfo> vertexList1, List<VertexInfo> vertexList2) {
    assertTrue("Vertices sizes should be the same", vertexList1.size() == vertexList2.size());
    Iterator<VertexInfo> it1 = vertexList1.iterator();
    Iterator<VertexInfo> it2 = vertexList2.iterator();
    while (it1.hasNext()) {
      assertTrue(it2.hasNext());
      VertexInfo info1 = it1.next();
      VertexInfo info2 = it2.next();
      isVertexEqual(info1, info2);
    }
  }

  private void isEdgeEqual(EdgeInfo edgeInfo1, EdgeInfo edgeInfo2) {
    assertTrue(edgeInfo1 != null);
    assertTrue(edgeInfo2 != null);
    String info1 = edgeInfo1.toString();
    String info2 = edgeInfo1.toString();
    assertTrue(info1.equals(info2));
  }

  private void isEdgeEqual(Collection<EdgeInfo> info1, Collection<EdgeInfo> info2) {
    assertTrue("sizes should be the same", info1.size() == info1.size());
    Iterator<EdgeInfo> it1 = info1.iterator();
    Iterator<EdgeInfo> it2 = info2.iterator();
    while (it1.hasNext()) {
      assertTrue(it2.hasNext());
      isEdgeEqual(it1.next(), it2.next());
    }
  }

  private void isTaskEqual(Collection<TaskInfo> info1, Collection<TaskInfo> info2) {
    assertTrue("sizes should be the same", info1.size() == info1.size());
    Iterator<TaskInfo> it1 = info1.iterator();
    Iterator<TaskInfo> it2 = info2.iterator();
    while (it1.hasNext()) {
      assertTrue(it2.hasNext());
      isTaskEqual(it1.next(), it2.next());
    }
  }

  private void isTaskEqual(TaskInfo taskInfo1, TaskInfo taskInfo2) {
    assertTrue(taskInfo1 != null);
    assertTrue(taskInfo2 != null);
    assertTrue(taskInfo1.getVertexInfo() != null);
    assertTrue(taskInfo2.getVertexInfo() != null);
    assertTrue(taskInfo1.getStatus().equals(taskInfo2.getStatus()));
    assertTrue(
        taskInfo1.getVertexInfo().getVertexName()
            .equals(taskInfo2.getVertexInfo().getVertexName()));
    isTaskAttemptEqual(taskInfo1.getTaskAttempts(), taskInfo2.getTaskAttempts());

    //Verify counters
    isCountersSame(taskInfo1, taskInfo2);
  }

  private void isCountersSame(BaseInfo info1, BaseInfo info2) {
    isCounterSame(info1.getCounter(TaskCounter.ADDITIONAL_SPILL_COUNT.name()),
        info2.getCounter(TaskCounter.ADDITIONAL_SPILL_COUNT.name()));

    isCounterSame(info1.getCounter(TaskCounter.SPILLED_RECORDS.name()),
        info2.getCounter(TaskCounter.SPILLED_RECORDS.name()));

    isCounterSame(info1.getCounter(TaskCounter.OUTPUT_RECORDS.name()),
        info2.getCounter(TaskCounter.OUTPUT_RECORDS.name()));

    isCounterSame(info1.getCounter(TaskCounter.OUTPUT_BYTES.name()),
        info2.getCounter(TaskCounter.OUTPUT_BYTES.name()));

    isCounterSame(info1.getCounter(TaskCounter.OUTPUT_RECORDS.name()),
        info2.getCounter(TaskCounter.OUTPUT_RECORDS.name()));

    isCounterSame(info1.getCounter(TaskCounter.REDUCE_INPUT_GROUPS.name()),
        info2.getCounter(TaskCounter.REDUCE_INPUT_GROUPS.name()));

    isCounterSame(info1.getCounter(TaskCounter.REDUCE_INPUT_RECORDS.name()),
        info2.getCounter(TaskCounter.REDUCE_INPUT_RECORDS.name()));
  }

  private void isCounterSame(Map<String, TezCounter> counter1, Map<String, TezCounter> counter2) {
    for (Map.Entry<String, TezCounter> entry : counter1.entrySet()) {
      String source = entry.getKey();
      long val = entry.getValue().getValue();

      //check if other counter has the same value
      assertTrue(counter2.containsKey(entry.getKey()));
      assertTrue(counter2.get(entry.getKey()).getValue() == val);
    }
  }

  private void isTaskAttemptEqual(Collection<TaskAttemptInfo> info1,
      Collection<TaskAttemptInfo> info2) {
    assertTrue("sizes should be the same", info1.size() == info1.size());
    Iterator<TaskAttemptInfo> it1 = info1.iterator();
    Iterator<TaskAttemptInfo> it2 = info2.iterator();
    while (it1.hasNext()) {
      assertTrue(it2.hasNext());
      isTaskAttemptEqual(it1.next(), it2.next());
    }
  }

  private void isTaskAttemptEqual(TaskAttemptInfo info1, TaskAttemptInfo info2) {
    assertTrue(info1 != null);
    assertTrue(info2 != null);
    assertTrue(info1.getTaskInfo() != null);
    assertTrue(info2.getTaskInfo() != null);
    assertTrue(info1.getStatus().equals(info2.getStatus()));
    assertTrue(info1.getTaskInfo().getVertexInfo().getVertexName().equals(info2.getTaskInfo()
        .getVertexInfo().getVertexName()));

    //Verify counters
    isCountersSame(info1, info2);
  }

  /**
   * Create sample file for wordcount program
   *
   * @param inputLoc
   * @throws IOException
   */
  private static void createSampleFile(Path inputLoc) throws IOException {
    fs.deleteOnExit(inputLoc);
    FSDataOutputStream out = fs.create(inputLoc);
    BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
    for (int i = 0; i < 10; i++) {
      writer.write("Sample " + RandomStringUtils.randomAlphanumeric(5));
      writer.newLine();
    }
    writer.close();
  }

  private DagInfo getDagInfo(String dagId) throws TezException {
    //Parse downloaded contents
    File downloadedFile = new File(DOWNLOAD_DIR
        + Path.SEPARATOR + dagId + ".zip");
    ATSFileParser parser = new ATSFileParser(downloadedFile);
    DagInfo dagInfo = parser.getDAGData(dagId);
    assertTrue(dagInfo.getDagId().equals(dagId));
    return dagInfo;
  }

  private void verifyCounter(Map<String, TezCounter> counterMap,
      String counterGroupName, long expectedVal) {
    //Iterate through group-->tezCounter
    for (Map.Entry<String, TezCounter> entry : counterMap.entrySet()) {
      if (counterGroupName != null) {
        if (entry.getKey().equals(counterGroupName)) {
          assertTrue(entry.getValue().getValue() == expectedVal);
        }
      } else {
        assertTrue(entry.getValue().getValue() == expectedVal);
      }
    }
  }

  TezClient getTezClient(boolean withTimeline) throws Exception {
    TezConfiguration tezConf = new TezConfiguration(miniTezCluster.getConfig());
    if (withTimeline) {
      tezConf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, withTimeline);
      tezConf.set(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS, "0.0.0.0:8188");
      tezConf.set(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS,
          ATSHistoryLoggingService.class.getName());
    } else {
      tezConf.set(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS,
          SimpleHistoryLoggingService.class.getName());
    }
    tezConf.setBoolean(TezConfiguration.TEZ_AM_ALLOW_DISABLED_TIMELINE_DOMAINS, true);

    TezClient tezClient = TezClient.create("WordCount", tezConf, false);
    tezClient.start();
    tezClient.waitTillReady();
    return tezClient;
  }

  private String runWordCount(String tokenizerProcessor, String summationProcessor,
      String dagName, boolean withTimeline)
      throws Exception {
    //HDFS path
    Path outputLoc = new Path("/tmp/outPath_" + System.currentTimeMillis());

    DataSourceDescriptor dataSource = MRInput.createConfigBuilder(conf,
        TextInputFormat.class, inputLoc.toString()).build();

    DataSinkDescriptor dataSink =
        MROutput.createConfigBuilder(conf, TextOutputFormat.class, outputLoc.toString()).build();

    Vertex tokenizerVertex = Vertex.create(TOKENIZER, ProcessorDescriptor.create(
        tokenizerProcessor)).addDataSource(INPUT, dataSource);

    OrderedPartitionedKVEdgeConfig edgeConf = OrderedPartitionedKVEdgeConfig
        .newBuilder(Text.class.getName(), IntWritable.class.getName(),
            HashPartitioner.class.getName()).build();

    Vertex summationVertex = Vertex.create(SUMMATION,
        ProcessorDescriptor.create(summationProcessor), 1).addDataSink(OUTPUT, dataSink);

    // Create DAG and add the vertices. Connect the producer and consumer vertices via the edge
    DAG dag = DAG.create(dagName);
    dag.addVertex(tokenizerVertex).addVertex(summationVertex).addEdge(
        Edge.create(tokenizerVertex, summationVertex, edgeConf.createDefaultEdgeProperty()));

    TezClient tezClient = getTezClient(withTimeline);

    // Update Caller Context
    CallerContext callerContext = CallerContext.create("TezExamples", "Tez WordCount Example Job");
    ApplicationId appId = tezClient.getAppMasterApplicationId();
    if (appId == null) {
      appId = ApplicationId.newInstance(1001l, 1);
    }
    callerContext.setCallerIdAndType(appId.toString(), "TezApplication");
    dag.setCallerContext(callerContext);

    DAGClient client = tezClient.submitDAG(dag);
    client.waitForCompletionWithStatusUpdates(Sets.newHashSet(StatusGetOpts.GET_COUNTERS));
    TezDAGID tezDAGID = TezDAGID.getInstance(tezClient.getAppMasterApplicationId(), 1);

    if (tezClient != null) {
      tezClient.stop();
    }
    return tezDAGID.toString();
  }

  /**
   * Processor which would just throw exception.
   */
  public static class FailProcessor extends SimpleMRProcessor {
    public FailProcessor(ProcessorContext context) {
      super(context);
    }

    @Override
    public void run() throws Exception {
      throw new Exception("Failing this processor for some reason");
    }
  }

  private void verifyDagInfo(DagInfo dagInfo, boolean ats) {
    if (ats) {
      VersionInfo versionInfo = dagInfo.getVersionInfo();
      assertTrue(versionInfo != null); //should be present post 0.5.4
      assertTrue(versionInfo.getVersion() != null);
      assertTrue(versionInfo.getRevision() != null);
      assertTrue(versionInfo.getBuildTime() != null);
    }

    assertTrue(dagInfo.getStartTime() > 0);
    assertTrue(dagInfo.getFinishTimeInterval() > 0);
    assertTrue(dagInfo.getStartTimeInterval() == 0);
    assertTrue(dagInfo.getStartTime() > 0);
    if (dagInfo.getStatus().equalsIgnoreCase(DAGState.SUCCEEDED.toString())) {
      assertTrue(dagInfo.getFinishTime() >= dagInfo.getStartTime());
    }
    assertTrue(dagInfo.getFinishTimeInterval() > dagInfo.getStartTimeInterval());

    assertTrue(dagInfo.getStartTime() > dagInfo.getSubmitTime());
    assertTrue(dagInfo.getTimeTaken() > 0);

    assertNotNull(dagInfo.getCallerContext());
    assertEquals("TezExamples", dagInfo.getCallerContext().getContext());
    assertEquals("Tez WordCount Example Job", dagInfo.getCallerContext().getBlob());
    assertNotNull(dagInfo.getCallerContext().getCallerId());
    assertEquals("TezApplication", dagInfo.getCallerContext().getCallerType());

    //Verify all vertices
    for (VertexInfo vertexInfo : dagInfo.getVertices()) {
      verifyVertex(vertexInfo, vertexInfo.getFailedTasksCount() > 0);
    }

    VertexInfo fastestVertex = dagInfo.getFastestVertex();
    assertTrue(fastestVertex != null);

    if (dagInfo.getStatus().equals(DAGState.SUCCEEDED)) {
      assertTrue(dagInfo.getSlowestVertex() != null);
    }
  }

  private void verifyVertex(VertexInfo vertexInfo, boolean hasFailedTasks) {
    assertTrue(vertexInfo != null);
    if (hasFailedTasks) {
      assertTrue(vertexInfo.getFailedTasksCount() > 0);
    }
    assertTrue(vertexInfo.getStartTimeInterval() > 0);
    assertTrue(vertexInfo.getStartTime() > 0);
    assertTrue(vertexInfo.getFinishTimeInterval() > 0);
    assertTrue(vertexInfo.getStartTimeInterval() < vertexInfo.getFinishTimeInterval());
    assertTrue(vertexInfo.getVertexName() != null);
    if (!hasFailedTasks) {
      assertTrue(vertexInfo.getFinishTime() > 0);
      assertTrue(vertexInfo.getFailedTasks().size() == 0);
      assertTrue(vertexInfo.getSucceededTasksCount() == vertexInfo.getSuccessfulTasks().size());
      assertTrue(vertexInfo.getFailedTasksCount() == 0);
      assertTrue(vertexInfo.getAvgTaskDuration() > 0);
      assertTrue(vertexInfo.getMaxTaskDuration() > 0);
      assertTrue(vertexInfo.getMinTaskDuration() > 0);
      assertTrue(vertexInfo.getTimeTaken() > 0);
      assertTrue(vertexInfo.getStatus().equalsIgnoreCase(VertexState.SUCCEEDED.toString()));
      assertTrue(vertexInfo.getCompletedTasksCount() > 0);
      assertTrue(vertexInfo.getFirstTaskToStart() != null);
      assertTrue(vertexInfo.getSucceededTasksCount() > 0);
      assertTrue(vertexInfo.getTasks().size() > 0);
    }

    for (TaskInfo taskInfo : vertexInfo.getTasks()) {
      if (taskInfo.getStatus().equals(TaskState.SUCCEEDED.toString())) {
        verifyTask(taskInfo, false);
      }
    }

    for (TaskInfo taskInfo : vertexInfo.getFailedTasks()) {
      verifyTask(taskInfo, true);
    }

    assertTrue(vertexInfo.getProcessorClassName() != null);
    assertTrue(vertexInfo.getStatus() != null);
    assertTrue(vertexInfo.getDagInfo() != null);
    assertTrue(vertexInfo.getInitTimeInterval() > 0);
    assertTrue(vertexInfo.getNumTasks() > 0);
  }

  private void verifyTask(TaskInfo taskInfo, boolean hasFailedAttempts) {
    assertTrue(taskInfo != null);
    assertTrue(taskInfo.getStatus() != null);
    assertTrue(taskInfo.getStartTimeInterval() > 0);

    //Not testing for killed attempts. So if there are no failures, it should succeed
    if (!hasFailedAttempts) {
      assertTrue(taskInfo.getStatus().equals(TaskState.SUCCEEDED.toString()));
      assertTrue(taskInfo.getFinishTimeInterval() > 0 && taskInfo.getFinishTime() > taskInfo
          .getFinishTimeInterval());
      assertTrue(
          taskInfo.getStartTimeInterval() > 0 && taskInfo.getStartTime() > taskInfo.getStartTimeInterval());
      assertTrue(taskInfo.getSuccessfulAttemptId() != null);
      assertTrue(taskInfo.getSuccessfulTaskAttempt() != null);
    }
    assertTrue(taskInfo.getTaskId() != null);

    for (TaskAttemptInfo attemptInfo : taskInfo.getTaskAttempts()) {
      verifyTaskAttemptInfo(attemptInfo);
    }
  }

  private void verifyTaskAttemptInfo(TaskAttemptInfo attemptInfo) {
    if (attemptInfo.getStatus() != null && attemptInfo.getStatus()
        .equals(TaskAttemptState.SUCCEEDED)) {
      assertTrue(attemptInfo.getStartTimeInterval() > 0);
      assertTrue(attemptInfo.getFinishTimeInterval() > 0);
      assertTrue(attemptInfo.getCreationTime() > 0);
      assertTrue(attemptInfo.getAllocationTime() > 0);
      assertTrue(attemptInfo.getStartTime() > 0);
      assertTrue(attemptInfo.getFinishTime() > 0);
      assertTrue(attemptInfo.getFinishTime() > attemptInfo.getStartTime());
      assertTrue(attemptInfo.getFinishTime() > attemptInfo.getFinishTimeInterval());
      assertTrue(attemptInfo.getStartTime() > attemptInfo.getStartTimeInterval());
      assertTrue(attemptInfo.getNodeId() != null);
      assertTrue(attemptInfo.getTimeTaken() != -1);
      assertTrue(attemptInfo.getEvents() != null);
      assertTrue(attemptInfo.getTezCounters() != null);
      assertTrue(attemptInfo.getContainer() != null);
    }
    assertTrue(attemptInfo.getTaskInfo() != null);
  }
}


text:
package net.minecraft.network.play.server;

import java.io.IOException;
import net.minecraft.network.Packet;
import net.minecraft.network.PacketBuffer;
import net.minecraft.network.play.INetHandlerPlayClient;
import net.minecraft.util.IChatComponent;

public class S40PacketDisconnect implements Packet<INetHandlerPlayClient>
{
    private IChatComponent reason;

    public S40PacketDisconnect()
    {
    }

    public S40PacketDisconnect(IChatComponent reasonIn)
    {
        this.reason = reasonIn;
    }

    /**
     * Reads the raw packet data from the data stream.
     */
    public void readPacketData(PacketBuffer buf) throws IOException
    {
        this.reason = buf.readChatComponent();
    }

    /**
     * Writes the raw packet data to the data stream.
     */
    public void writePacketData(PacketBuffer buf) throws IOException
    {
        buf.writeChatComponent(this.reason);
    }

    /**
     * Passes this Packet on to the NetHandler for processing.
     */
    public void processPacket(INetHandlerPlayClient handler)
    {
        handler.handleDisconnect(this);
    }
}

redpajama_set_name: RedPajamaGithub


text:
[See html formatted version](https://huasofoundries.github.io/google-maps-documentation/Duration.html)

Duration interface
------------------

google.maps.Duration interface

A representation of duration as a numeric value and a display string.

Properties

[text](#Duration.text)

**Type:**  string

A string representation of the duration value.

[value](#Duration.value)

**Type:**  number

The duration in seconds.


text:
import string
import openravepy as orpy
import logging
import numpy
from orsampler import RobotCSpaceSampler, GraspApproachConstraintsManager
from sampler import FreeSpaceProximitySampler
from rrt import DynamicPGoalProvider, RRT
from utils import OpenRAVEDrawer, ObjectFileIO
from grasp_goal_sampler import GraspGoalSampler
from core import PlanningSceneInterface
from hierarchy_visualization import FreeSpaceProximitySamplerVisualizer

class IntegratedHFTSPlanner(object):
    """ Implements a simple to use interface to the integrated HFTS planner. """

    def __init__(self, env_file, hand_file, hand_cache_file, robot_name, manipulator_name,
                 data_root_path, dof_weights=None, max_num_hierarchy_descends=4,
                 min_iterations=20, max_iterations=70, p_goal_tree=0.8, vel_factor=0.2,
                 b_visualize_system=False, b_visualize_grasps=False, b_visualize_hfts=False,
                 b_show_traj=False, b_show_search_tree=False, free_space_weight=0.1, connected_space_weight=4.0,
                 use_approximates=True, compute_velocities=True, time_limit=60.0):
        """ Creates a new instance of an HFTS planner
            NOTE: It is only possible to display one scene in OpenRAVE at a time. Hence, if the parameters
            b_visualize_system and b_visualize_grasps are both true, only the motion planning scene is shown.
         @param env_file String containing a path to an OpenRAVE environment
         @param hand_file String containing a path to an OpenRAVE hand model (a robot consisting of just the hand)
         @param robot_name String containing the name of the robot to use
         @param manipulator_name String containing the name of the robot's manipulator to use
         @param dof_weights (optional) List of floats, containing a weight for each DOF. These weights are used in
            the distance function on C-Space.
         @param max_num_hierarchy_descends Number of sampling steps for sampling the HFTS.
            (should be greater than HFTS depth)
         @param min_iterations Minimum number of iterations for an optimization step on each HFTS layer
         @param max_iterations Maximum number of iterations for an optimization step on each HFTS layer
         @param p_goal_tree Float denoting the probability of selecting a goal tree rather than an approximate
            goal tree for extensions.
         @param b_visualize_system Boolean, if True, show OpenRAVE viewer displaying the motion planning scene
         @param b_visualize_grasps Boolean, if True, show OpenRAVE viewer displaying the grasp planning scene
         @param b_visualize_hfts Boolean, if True, show a window with a graph visualization of the explored HFTS space
         @param b_show_traj Boolean, if True, simulates the trajectory execution in OpenRAVE before returning
         @param b_show_search_tree Boolean, if True, visualizes a projection of the search trees in OpenRAVE
         @param free_space_weight Weight (float) for HFTS rating function t(n) for the distance to
            free space configurations
         @param connected_space_weight Weight (float) for HFTS rating function t(n) for the distance to
            connected configurations
         @param use_approximates Boolean, if True, the grasp sampler returns approximate grasps if it hasn't
            found a valid grasp yet
         @param use_velocities Boolean, if True, compute a whole trajectory (with velocities), else just a path
         @param time_limit Runtime limit for the algorithm in seconds (float)
         """
        self._env = orpy.Environment()
        self._env.Load(env_file)
        if b_visualize_system:
            self._env.SetViewer('qtcoin')
            b_visualize_grasps = False
        if len(self._env.GetRobots()) == 0:
            raise ValueError('The provided environment does not contain a robot!')
        self._robot = self._env.GetRobot(robot_name)
        self._robot.SetActiveManipulator(manipulator_name)
        # TODO this is a hack to fix a super weird behaviour regarding the
        # TODO the transformation of the kmr_robotiq robot. this should NOT be necessary
        self._robot_transformation_hack = self._robot.GetTransform()
        self._or_motion_planner = orpy.interfaces.BaseManipulation(self._robot)
        if dof_weights is None:
            dof_weights = self._robot.GetDOF() * [1.0]
        self._cSampler = RobotCSpaceSampler(self._env, self._robot, scaling_factors=dof_weights)
        # TODO read these robot-specific specs from a file
        planning_scene_interface = PlanningSceneInterface(self._env, self._robot.GetName())
        self._object_io_interface = ObjectFileIO(data_path=data_root_path)
        self._grasp_planner = GraspGoalSampler(object_io_interface=self._object_io_interface,
                                               hand_path=hand_file, hand_cache_file=hand_cache_file,
                                               planning_scene_interface=planning_scene_interface,
                                               visualize=b_visualize_grasps)
        hierarchy_visualizer = None
        if b_visualize_hfts:
            hierarchy_visualizer = FreeSpaceProximitySamplerVisualizer(self._robot)
        if max_num_hierarchy_descends <= 0:
            max_num_hierarchy_descends = self._grasp_planner.get_max_depth() + 1
        self._hierarchy_sampler = FreeSpaceProximitySampler(self._grasp_planner, self._cSampler,
                                                            k=max_num_hierarchy_descends,
                                                            num_iterations=max_iterations,
                                                            min_num_iterations=min_iterations,
                                                            b_return_approximates=use_approximates,
                                                            connected_weight=connected_space_weight,
                                                            free_space_weight=free_space_weight,
                                                            debug_drawer=hierarchy_visualizer)
        # TODO the open hand configuration should be given from a configuration file
        self._constraints_manager = GraspApproachConstraintsManager(self._env, self._robot,
                                                                    self._cSampler, numpy.array([0.0, 0.0495]))
        p_goal_provider = DynamicPGoalProvider()
        self._debug_tree_drawer = None
        if b_show_search_tree:
            self._debug_tree_drawer = OpenRAVEDrawer(self._env, self._robot, True)
        self._rrt_planner = RRT(p_goal_provider, self._cSampler, self._hierarchy_sampler, logging.getLogger(),
                                pgoal_tree=p_goal_tree, constraints_manager=self._constraints_manager)
        self._time_limit = time_limit
        self._vel_factor = vel_factor
        self._last_path = None
        self._last_traj = None
        self._last_obj = None
        self._compute_velocities = compute_velocities
        self._b_show_trajectory = b_show_traj

    def load_target_object(self, obj_id, model_id=None):
        self._last_obj = obj_id
        self._constraints_manager.set_object_name(obj_id)
        self._grasp_planner.set_object(obj_id=obj_id, model_id=model_id)

    def get_robot(self):
        return self._robot

    def get_object_frame(self, frame_id):
        body = self._env.GetKinBody(frame_id)
        if body is None:
            return None
        return body.GetTransform()

    def create_or_trajectory(self, path, vel_factor=0.2):
        if path is None:
            return None
        configurations_path = map(lambda x: x.get_configuration(), path)
        # The path ends in a pre-grasp configuration.
        # The final grasp configuration is stored as additional data in the last waypoint,
        # so we need to construct the final configuration here.
        grasp_hand_config = path[-1].get_data()
        last_config = numpy.array(configurations_path[-1])
        hand_idxs = self._robot.GetActiveManipulator().GetGripperIndices()
        assert len(hand_idxs) == len(grasp_hand_config)
        j = 0
        for i in hand_idxs:
            last_config[i] = grasp_hand_config[j]
            j += 1
        configurations_path.append(last_config)

        vel_limits = self._robot.GetDOFVelocityLimits()
        self._robot.SetDOFVelocityLimits(vel_factor * vel_limits)
        traj = orpy.RaveCreateTrajectory(self._env, '')
        cs = traj.GetConfigurationSpecification()
        dof_string = string.join([' ' + str(x) for x in range(self._robot.GetDOF())])
        cs.AddGroup('joint_values ' + self._robot.GetName() + dof_string, self._robot.GetDOF(), 'linear')
        # cs.AddDerivativeGroups(1, True)
        traj.Init(cs)
        for idx in range(len(configurations_path)):
            traj.Insert(idx, configurations_path[idx])
        orpy.planningutils.RetimeTrajectory(traj, hastimestamps=False)
        self._robot.SetDOFVelocityLimits(vel_limits)
        return traj

    def plan_arm_motion(self, target_pose, start_configuration, grasped_object=None):
        self._robot.SetDOFValues(start_configuration)
        if self._debug_tree_drawer is not None:
            self._debug_tree_drawer.clear()
            self._debug_tree_drawer.draw_pose(target_pose)
        vel_limits = self._robot.GetDOFVelocityLimits()
        self._robot.SetDOFVelocityLimits(self._vel_factor * vel_limits)
        if grasped_object is not None:
            body = self._env.GetKinBody(grasped_object)
            if body is not None:
                self._robot.Grab(body)
        try:
            traj = self._or_motion_planner.MoveToHandPosition(matrices=[target_pose], outputtraj=True,
                                                              outputtrajobj=True, execute=self._b_show_trajectory)
        except orpy.planning_error as pe:
            logging.info('[IntegratedHFTSPlanner::plan_arm_motion] Planning arm motion to pose failed.')
            traj = None
        if grasped_object is not None:
            body = self._env.GetKinBody(grasped_object)
            if body is not None:
                self._robot.Release(body)
        self._robot.SetDOFVelocityLimits(vel_limits)
        return traj

    def plan(self, start_configuration):
        if self._debug_tree_drawer is not None:
            self._debug_tree_drawer.clear()
            debug_function = self._debug_tree_drawer.draw_trees
        else:
            def debug_function(forward_tree, backward_trees):
                pass
        self._robot.SetDOFValues(start_configuration)
        self._last_path = self._rrt_planner.proximity_birrt(start_configuration, time_limit=self._time_limit,
                                                            debug_function=debug_function)
        grasp_pose = None
        if self._last_path is not None:
            last_config = self._last_path[-1].get_configuration()
            self._robot.SetDOFValues(last_config)
            grasp_pose = self._robot.GetActiveManipulator().GetEndEffectorTransform()
            object_pose = self._env.GetKinBody(self._last_obj).GetTransform()
            grasp_pose = numpy.dot(numpy.linalg.inv(object_pose), grasp_pose)
        if self._compute_velocities:
            self._last_traj = self.create_or_trajectory(self._last_path, self._vel_factor)
            if self._b_show_trajectory and self._last_traj is not None:
                controller = self._robot.GetController()
                controller.SetPath(self._last_traj)
                self._robot.WaitForController(self._last_traj.GetDuration())
                controller.Reset()
            return self._last_traj, grasp_pose
        return self._last_path, grasp_pose

    def add_planning_scene_object(self, object_name, object_class_name, pose):
        # TODO this is a hack, remove this again:
        if object_name == 'ground_plane':
            return True
        body = self._env.GetKinBody(object_name)
        if body is None:
            or_file_name = self._object_io_interface.get_openrave_file_name(object_class_name)
            body_loaded = self._env.Load(or_file_name)
            if not body_loaded:
                logging.error('Failed to load object from file ' + str(or_file_name))
                return False
            body = self._env.GetBodies()[-1]  # last body is most recently added
            body.SetName(object_name)
        if body == self._robot:
            pose = numpy.dot(pose, self._robot_transformation_hack)
        body.SetTransform(pose)
        return True

    def remove_planning_scene_object(self, object_name):
        if object_name == 'all':
            robot = self._env.GetRobots()[0]
            kinbodies = self._env.GetBodies()
            for body in kinbodies:
                if body != robot:
                    self._env.Remove(body)
            return True
        else:
            body = self._env.GetKinBody(object_name)
            if body is not None:
                self._env.Remove(body)
                return True
            return False

    def set_parameters(self, min_iterations=None, max_iterations=None,
                       free_space_weight=None, connected_space_weight=None,
                       use_approximates=None, compute_velocities=None,
                       time_limit=None, com_center_weight=None,
                       reachability_weight=None,
                       hfts_generation_params=None, max_num_hierarchy_descends=None,
                       b_force_new_hfts=None, vel_factor=None):
        # TODO some of these parameters are robot hand specific
        if time_limit is not None:
            self._time_limit = time_limit
        if compute_velocities is not None:
            self._compute_velocities = compute_velocities
        self._grasp_planner.set_parameters(com_center_weight=com_center_weight,
                                           reachability_weight=reachability_weight,
                                           b_force_new_hfts=b_force_new_hfts,
                                           hfts_generation_params=hfts_generation_params)
        if max_num_hierarchy_descends == 0:
            max_num_hierarchy_descends = self._grasp_planner.get_max_depth() + 1
        self._hierarchy_sampler.set_parameters(min_iterations=min_iterations,
                                               max_iterations=max_iterations,
                                               free_space_weight=free_space_weight,
                                               connected_space_weight=connected_space_weight,
                                               k=max_num_hierarchy_descends,
                                               use_approximates=use_approximates)
        if vel_factor is not None:
            self._vel_factor = vel_factor
        # TODO implement the rest


text:
package io.joynr.dispatching.subscription;

import static org.hamcrest.Matchers.contains;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.argThat;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;

import java.io.File;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.Lists;
import io.joynr.dispatching.DispatcherImpl;
import io.joynr.dispatching.ProviderDirectory;
import io.joynr.dispatching.RequestCaller;
import io.joynr.dispatching.RequestCallerFactory;
import io.joynr.messaging.MessagingQos;
import io.joynr.provider.AbstractSubscriptionPublisher;
import io.joynr.provider.Deferred;
import io.joynr.provider.Promise;
import io.joynr.provider.ProviderContainer;
import io.joynr.pubsub.SubscriptionQos;
import io.joynr.pubsub.publication.BroadcastFilter;
import io.joynr.runtime.ShutdownNotifier;
import joynr.BroadcastFilterParameters;
import joynr.BroadcastSubscriptionRequest;
import joynr.MulticastPublication;
import joynr.OnChangeSubscriptionQos;
import joynr.PeriodicSubscriptionQos;
import joynr.SubscriptionPublication;
import joynr.SubscriptionReply;
import joynr.SubscriptionRequest;
import joynr.tests.testBroadcastInterface;
import joynr.tests.testLocationUpdateSelectiveBroadcastFilter;
import joynr.tests.testLocationUpdateWithSpeedSelectiveBroadcastFilter;
import joynr.tests.testProvider;
import joynr.types.Localisation.GpsFixEnum;
import joynr.types.Localisation.GpsLocation;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;

@RunWith(MockitoJUnitRunner.class)
public class PublicationManagerTest {

    private static final String PROVIDER_PARTICIPANT_ID = "providerParticipantId";
    private static final String PROXY_PARTICIPANT_ID = "proxyParticipantId";

    private static final String SUBSCRIPTION_ID = "PublicationTest_id";

    ScheduledExecutorService cleanupScheduler;
    PublicationManagerImpl publicationManager;

    @Captor
    ArgumentCaptor<String> sentProviderParticipantId;
    @Captor
    ArgumentCaptor<String> sentProxyParticipantId;
    @Captor
    ArgumentCaptor<SubscriptionPublication> sentPublication;
    @Captor
    ArgumentCaptor<MessagingQos> sentMessagingQos;

    @Mock
    AttributePollInterpreter attributePollInterpreter;
    @Mock
    private ProviderDirectory providerDirectory;
    @Mock
    private DispatcherImpl dispatcher;
    @Mock
    private testProvider provider;

    @Mock
    private AbstractSubscriptionPublisher subscriptionPublisher;

    private RequestCaller requestCaller;

    @Mock
    private ProviderContainer providerContainer;

    @Mock
    private ShutdownNotifier shutdownNotifier;

    String valueToPublish = "valuePublished";

    @Before
    public void setUp() {
        Deferred<String> valueToPublishDeferred = new Deferred<String>();
        valueToPublishDeferred.resolve(valueToPublish);
        Promise<Deferred<String>> valueToPublishPromise = new Promise<Deferred<String>>(valueToPublishDeferred);

        cleanupScheduler = new ScheduledThreadPoolExecutor(1);
        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        providerDirectory,
                                                        cleanupScheduler,
                                                        Mockito.mock(SubscriptionRequestStorage.class),
                                                        shutdownNotifier);

        requestCaller = new RequestCallerFactory().create(provider);
        when(providerContainer.getProviderProxy()).thenReturn(requestCaller.getProxy());
        when(providerContainer.getSubscriptionPublisher()).thenReturn(subscriptionPublisher);

        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(false);

        doReturn(valueToPublishPromise).when(attributePollInterpreter).execute(any(ProviderContainer.class),
                                                                               any(Method.class));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 4000)
    public void doNotDelayBroadcastPublicationBurstsForOnChangeSubscriptionsWithoutMinInterval() throws Exception {
        int subscriptionLength = 500;

        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos();
        qos.setMinIntervalMs(0);
        qos.setValidityMs(subscriptionLength);
        qos.setPublicationTtlMs(400);

        String subscriptionId = "subscriptionId";
        String proxyId = "proxyId";
        String providerId = "providerId";
        String broadcastName = "location";

        ProviderDirectory requestCallerDirectory = mock(ProviderDirectory.class);
        SubscriptionRequest subscriptionRequest = new BroadcastSubscriptionRequest(subscriptionId,
                                                                                broadcastName,
                                                                                new BroadcastFilterParameters(),
                                                                                qos);
        PublicationManager publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                                           dispatcher,
                                                                           requestCallerDirectory,
                                                                           cleanupScheduler,
                                                                           Mockito.mock(SubscriptionRequestStorage.class),
                                                                           shutdownNotifier);

        when(requestCallerDirectory.get(eq(providerId))).thenReturn(providerContainer);
        when(requestCallerDirectory.contains(eq(providerId))).thenReturn(true);

        publicationManager.addSubscriptionRequest(proxyId, providerId, subscriptionRequest);
        List<BroadcastFilter> noFilters = Lists.newArrayList();

        int nrBroadcasts = 100;
        for (int i = 0; i < nrBroadcasts; i++) {
            publicationManager.broadcastOccurred(subscriptionId, noFilters, 2 * i + 1);
        }

        Thread.sleep(subscriptionLength);

        verify(dispatcher, times(nrBroadcasts)).sendSubscriptionPublication(eq(providerId),
                                                                            (Set<String>) argThat(contains(proxyId)),
                                                                            any(SubscriptionPublication.class),
                                                                            any(MessagingQos.class));

        Thread.sleep(subscriptionLength);

        verify(dispatcher).sendSubscriptionReply(eq(providerId),
                                                 eq(proxyId),
                                                 any(SubscriptionReply.class),
                                                 any(MessagingQos.class));
        //(eq(subscriptionRequest), eq(proxyId), eq(providerId));
        verifyNoMoreInteractions(dispatcher);
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 4000)
    public void delayBroadcastPublicationBurstsForOnChangeSubscriptions() throws Exception {
        int subscriptionLength = 500;
        int minIntervalMs = 100;

        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos();
        qos.setMinIntervalMs(minIntervalMs);
        qos.setValidityMs(subscriptionLength);
        qos.setPublicationTtlMs(400);

        String subscriptionId = "subscriptionId";
        String proxyId = "proxyId";
        String providerId = "providerId";
        String broadcastName = "location";

        ProviderDirectory providerDirectory = mock(ProviderDirectory.class);
        SubscriptionRequest subscriptionRequest = new BroadcastSubscriptionRequest(subscriptionId,
                                                                                broadcastName,
                                                                                new BroadcastFilterParameters(),
                                                                                qos);
        PublicationManager publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                                           dispatcher,
                                                                           providerDirectory,
                                                                           cleanupScheduler,
                                                                           Mockito.mock(SubscriptionRequestStorage.class),
                                                                           shutdownNotifier);

        when(providerDirectory.get(eq(providerId))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(providerId))).thenReturn(true);

        publicationManager.addSubscriptionRequest(proxyId, providerId, subscriptionRequest);
        List<BroadcastFilter> noFilters = Lists.newArrayList();

        publicationManager.broadcastOccurred(subscriptionId, noFilters, 0);
        int nrIterations = 10;

        for (int i = 1; i <= nrIterations; i++) {
            publicationManager.broadcastOccurred(subscriptionId, noFilters, i);
        }

        Thread.sleep(minIntervalMs);

        publicationManager.broadcastOccurred(subscriptionId, noFilters, nrIterations + 1);
        verify(dispatcher, times(2)).sendSubscriptionPublication(eq(providerId),
                                                                 (Set<String>) argThat(contains(proxyId)),
                                                                 any(SubscriptionPublication.class),
                                                                 any(MessagingQos.class));
        verify(dispatcher).sendSubscriptionReply(anyString(),
                                                 anyString(),
                                                 any(SubscriptionReply.class),
                                                 any(MessagingQos.class));

        Thread.sleep(subscriptionLength);
        verifyNoMoreInteractions(dispatcher);

        reset(dispatcher);
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 4000)
    public void delayAttributePublicationBurstsForOnChangeSubscriptions() throws Exception {
        int subscriptionLength = 500;
        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos();
        qos.setMinIntervalMs(400);
        qos.setValidityMs(subscriptionLength);
        qos.setPublicationTtlMs(400);

        String subscriptionId = "subscriptionId";
        String proxyId = "proxyId";
        String providerId = "providerId";
        String attributeName = "location";

        ProviderDirectory providerDirectory = mock(ProviderDirectory.class);
        SubscriptionRequest subscriptionRequest = new SubscriptionRequest(subscriptionId, attributeName, qos);
        PublicationManager publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                                           dispatcher,
                                                                           providerDirectory,
                                                                           cleanupScheduler,
                                                                           Mockito.mock(SubscriptionRequestStorage.class),
                                                                           shutdownNotifier);

        when(providerDirectory.get(eq(providerId))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(providerId))).thenReturn(true);

        final Semaphore onReceiveSemaphore = new Semaphore(0);
        doAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock invocation) {
                onReceiveSemaphore.release();
                return null;
            }
        }).when(dispatcher).sendSubscriptionPublication(eq(providerId),
                                                        (Set<String>) argThat(contains(proxyId)),
                                                        any(SubscriptionPublication.class),
                                                        any(MessagingQos.class));

        publicationManager.addSubscriptionRequest(proxyId, providerId, subscriptionRequest);

        /* a burst of attribute changes only leads to one publication send out */
        for (int i = 0; i < 3; i++) {
            publicationManager.attributeValueChanged(subscriptionId, i);
        }

        assertTrue(onReceiveSemaphore.tryAcquire(2, subscriptionLength + 1000, TimeUnit.MILLISECONDS));

        assertFalse(onReceiveSemaphore.tryAcquire(1, Math.max(subscriptionLength, 200), TimeUnit.MILLISECONDS));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void addPublicationWithExpiryDate() throws Exception {
        long pubicationActiveForMs = 300;

        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos();
        qos.setMinIntervalMs(0);
        qos.setValidityMs(pubicationActiveForMs);
        qos.setPublicationTtlMs(1000);

        SubscriptionRequest subscriptionRequest = new SubscriptionRequest(SUBSCRIPTION_ID, "location", qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        publicationManager.attributeValueChanged(SUBSCRIPTION_ID, valueToPublish);

        // sending initial value plus the attributeValueChanged
        verify(dispatcher, times(2)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                 (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                 any(SubscriptionPublication.class),
                                                                 any(MessagingQos.class));

        Thread.sleep(pubicationActiveForMs);
        reset(dispatcher);

        publicationManager.attributeValueChanged(SUBSCRIPTION_ID, valueToPublish);

        verify(dispatcher, timeout(300).times(0)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                              (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                              any(SubscriptionPublication.class),
                                                                              any(MessagingQos.class));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void addPublicationWithoutExpiryDate() throws Exception {
        int period = 200;
        PeriodicSubscriptionQos qos = new PeriodicSubscriptionQos();
        qos.setPeriodMs(period).setValidityMs(SubscriptionQos.IGNORE_VALUE);
        qos.setAlertAfterIntervalMs(500).setPublicationTtlMs(1000);

        SubscriptionRequest subscriptionRequest = new SubscriptionRequest(SUBSCRIPTION_ID, "location", qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        verify(dispatcher, timeout(period * 5).times(6)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                                (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));

        reset(dispatcher);
        publicationManager.stopPublication(SUBSCRIPTION_ID);

        verify(dispatcher, timeout(300).times(0)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                              (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                              any(SubscriptionPublication.class),
                                                                              any(MessagingQos.class));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void startAndStopPeriodicPublication() throws Exception {
        int period = 200;
        int testLengthMax = 3000;
        PeriodicSubscriptionQos qos = new PeriodicSubscriptionQos();
        qos.setPeriodMs(200).setValidityMs(testLengthMax).setPublicationTtlMs(testLengthMax);
        SubscriptionRequest subscriptionRequest = new SubscriptionRequest(SUBSCRIPTION_ID, "location", qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        verify(dispatcher, timeout(period * 5).times(6)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                                (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));

        reset(dispatcher);
        publicationManager.stopPublication(SUBSCRIPTION_ID);

        verify(dispatcher, timeout(testLengthMax).times(0)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                                (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void stopAllPublicationsFromProvider() throws Exception {
        String subscriptionId1 = "subscriptionid1";
        String subscriptionId2 = "subscriptionid2";
        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos();
        qos.setMinIntervalMs(0).setValidityMs(3000).setPublicationTtlMs(1000);
        SubscriptionRequest subscriptionRequest1 = new SubscriptionRequest(subscriptionId1, "location", qos);
        SubscriptionRequest subscriptionRequest2 = new SubscriptionRequest(subscriptionId2, "location", qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest1);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest2);

        publicationManager.attributeValueChanged(subscriptionId1, valueToPublish);
        publicationManager.attributeValueChanged(subscriptionId2, valueToPublish);

        // sending initial values for 2 subscriptions, plus the 2 attributeValueChanged
        verify(dispatcher, times(4)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                 (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                 any(SubscriptionPublication.class),
                                                                 any(MessagingQos.class));

        reset(dispatcher);
        publicationManager.entryRemoved(PROVIDER_PARTICIPANT_ID);

        publicationManager.attributeValueChanged(subscriptionId1, valueToPublish);
        publicationManager.attributeValueChanged(subscriptionId2, valueToPublish);

        verify(dispatcher, timeout(300).times(0)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                              (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                              any(SubscriptionPublication.class),
                                                                              any(MessagingQos.class));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void restorePublications() throws Exception {
        int period = 200;
        String subscriptionId1 = "subscriptionid1";
        String subscriptionId2 = "subscriptionid2";
        long validityMs = 3000;
        SubscriptionQos qosNoExpiry = new PeriodicSubscriptionQos().setPeriodMs(100)
                                                                   .setExpiryDateMs(SubscriptionQos.NO_EXPIRY_DATE)
                                                                   .setAlertAfterIntervalMs(500)
                                                                   .setPublicationTtlMs(1000);
        SubscriptionQos qosExpires = new PeriodicSubscriptionQos().setPeriodMs(100)
                                                                  .setValidityMs(validityMs)
                                                                  .setAlertAfterIntervalMs(500)
                                                                  .setPublicationTtlMs(1000);
        SubscriptionRequest subscriptionRequest1 = new SubscriptionRequest(subscriptionId1, "location", qosNoExpiry);
        SubscriptionRequest subscriptionRequest2 = new SubscriptionRequest(subscriptionId2, "location", qosExpires);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest1);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest2);

        Thread.sleep(period);
        verify(dispatcher, times(0)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                 (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                 any(SubscriptionPublication.class),
                                                                 any(MessagingQos.class));

        publicationManager.entryAdded(PROVIDER_PARTICIPANT_ID, providerContainer);

        verify(dispatcher, timeout(period * 5).times(12)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                                (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void removeQueuedSubscriptionsProperly() throws Exception {
        int period = 200;
        String subscriptionId1 = "subscriptionid_removeQueuedSubscriptionsProperly";
        SubscriptionQos qosNoExpiry = new PeriodicSubscriptionQos().setPeriodMs(100)
                                                                   .setExpiryDateMs(SubscriptionQos.NO_EXPIRY_DATE)
                                                                   .setAlertAfterIntervalMs(500)
                                                                   .setPublicationTtlMs(1000);
        SubscriptionRequest subscriptionRequest1 = new SubscriptionRequest(subscriptionId1, "location", qosNoExpiry);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest1);

        publicationManager.stopPublication(subscriptionId1);

        publicationManager.entryAdded(PROVIDER_PARTICIPANT_ID, providerContainer);
        Thread.sleep(period);
        verify(dispatcher, times(0)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                 (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                 any(SubscriptionPublication.class),
                                                                 any(MessagingQos.class));
    }

    @SuppressWarnings("unchecked")
    @Test
    public void broadcastPublicationIsSent() throws Exception {

        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        providerDirectory,
                                                        cleanupScheduler,
                                                        Mockito.mock(SubscriptionRequestStorage.class),
                                                        shutdownNotifier);

        long minInterval_ms = 0;
        long ttl = 1000;
        testBroadcastInterface.LocationUpdateWithSpeedSelectiveBroadcastFilterParameters filterParameters = new testBroadcastInterface.LocationUpdateWithSpeedSelectiveBroadcastFilterParameters();
        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos().setMinIntervalMs(minInterval_ms)
                                                                   .setExpiryDateMs(SubscriptionQos.NO_EXPIRY_DATE)
                                                                   .setPublicationTtlMs(ttl);
        ;

        SubscriptionRequest subscriptionRequest = new BroadcastSubscriptionRequest(SUBSCRIPTION_ID,
                                                                                "subscribedToName",
                                                                                filterParameters,
                                                                                qos);
        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        GpsLocation location = new GpsLocation(1.0, 2.0, 3.0, GpsFixEnum.MODE2D, 4.0, 5.0, 6.0, 7.0, 9l, 10l, 11);
        double speed = 100;

        publicationManager.broadcastOccurred(subscriptionRequest.getSubscriptionId(),
                                             new ArrayList<BroadcastFilter>(),
                                             location,
                                             speed);

        ArgumentCaptor<SubscriptionPublication> publicationCaptured = ArgumentCaptor.forClass(SubscriptionPublication.class);
        ArgumentCaptor<MessagingQos> qosCaptured = ArgumentCaptor.forClass(MessagingQos.class);

        verify(dispatcher).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                       (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                       publicationCaptured.capture(),
                                                       qosCaptured.capture());

        List<?> response = (List<?>) publicationCaptured.getValue().getResponse();
        assertEquals(location, response.get(0));
        assertEquals(speed, response.get(1));
        assertEquals(ttl, qosCaptured.getValue().getRoundTripTtl_ms());

    }

    @SuppressWarnings("unchecked")
    @Test
    public void broadcastPublicationCallsAllFiltersWithFilterParametersAndValues() throws Exception {

        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        providerDirectory,
                                                        cleanupScheduler,
                                                        Mockito.mock(SubscriptionRequestStorage.class),
                                                        shutdownNotifier);

        long minInterval_ms = 0;
        long ttl = 1000;

        testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters filterParameters = new testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters();
        filterParameters.setCountry("Germany");
        filterParameters.setStartTime("4:00");

        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos().setMinIntervalMs(minInterval_ms)
                                                                   .setExpiryDateMs(SubscriptionQos.NO_EXPIRY_DATE)
                                                                   .setPublicationTtlMs(ttl);

        SubscriptionRequest subscriptionRequest = new BroadcastSubscriptionRequest(SUBSCRIPTION_ID,
                                                                                "subscribedToName",
                                                                                filterParameters,
                                                                                qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        GpsLocation eventValue = new GpsLocation();

        ArrayList<BroadcastFilter> filters = new ArrayList<BroadcastFilter>();
        testLocationUpdateSelectiveBroadcastFilter filter1 = mock(testLocationUpdateSelectiveBroadcastFilter.class);
        when(filter1.filter(any(GpsLocation.class),
                            any(testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters.class))).thenReturn(true);
        filters.add(filter1);

        testLocationUpdateSelectiveBroadcastFilter filter2 = mock(testLocationUpdateSelectiveBroadcastFilter.class);
        when(filter2.filter(any(GpsLocation.class),
                            any(testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters.class))).thenReturn(true);
        filters.add(filter2);

        publicationManager.broadcastOccurred(subscriptionRequest.getSubscriptionId(), filters, eventValue);

        ArgumentCaptor<SubscriptionPublication> publicationCaptured = ArgumentCaptor.forClass(SubscriptionPublication.class);
        ArgumentCaptor<MessagingQos> qosCaptured = ArgumentCaptor.forClass(MessagingQos.class);

        verify(dispatcher).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                       (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                       publicationCaptured.capture(),
                                                       qosCaptured.capture());

        verify(filter1).filter(eventValue, filterParameters);
        verify(filter2).filter(eventValue, filterParameters);

    }

    @SuppressWarnings("unchecked")
    @Test
    public void broadcastPublicationIsSentWhenFiltersPass() throws Exception {

        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        providerDirectory,
                                                        cleanupScheduler,
                                                        Mockito.mock(SubscriptionRequestStorage.class),
                                                        shutdownNotifier);

        long minInterval_ms = 0;
        long ttl = 1000;
        testBroadcastInterface.LocationUpdateWithSpeedSelectiveBroadcastFilterParameters filterParameters = new testBroadcastInterface.LocationUpdateWithSpeedSelectiveBroadcastFilterParameters();
        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos().setMinIntervalMs(minInterval_ms)
                                                                   .setExpiryDateMs(SubscriptionQos.NO_EXPIRY_DATE)
                                                                   .setPublicationTtlMs(ttl);

        SubscriptionRequest subscriptionRequest = new BroadcastSubscriptionRequest(SUBSCRIPTION_ID,
                                                                                "subscribedToName",
                                                                                filterParameters,
                                                                                qos);
        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        GpsLocation location = new GpsLocation(1.0, 2.0, 3.0, GpsFixEnum.MODE2D, 4.0, 5.0, 6.0, 7.0, 9l, 10l, 11);
        float speed = 100;

        ArrayList<BroadcastFilter> filters = new ArrayList<BroadcastFilter>();
        testLocationUpdateWithSpeedSelectiveBroadcastFilter filterTrue = mock(testLocationUpdateWithSpeedSelectiveBroadcastFilter.class);
        when(filterTrue.filter(any(GpsLocation.class),
                               any(Float.class),
                               any(testBroadcastInterface.LocationUpdateWithSpeedSelectiveBroadcastFilterParameters.class))).thenReturn(true);
        filters.add(filterTrue);

        publicationManager.broadcastOccurred(subscriptionRequest.getSubscriptionId(), filters, location, speed);

        ArgumentCaptor<SubscriptionPublication> publicationCaptured = ArgumentCaptor.forClass(SubscriptionPublication.class);
        ArgumentCaptor<MessagingQos> qosCaptured = ArgumentCaptor.forClass(MessagingQos.class);

        verify(dispatcher).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                       (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                       publicationCaptured.capture(),
                                                       qosCaptured.capture());

        List<?> response = (List<?>) publicationCaptured.getValue().getResponse();
        assertEquals(location, response.get(0));
        assertEquals(speed, response.get(1));

    }

    @SuppressWarnings("unchecked")
    @Test
    public void broadcastPublicationNotSentWhenFiltersFail() throws Exception {

        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        providerDirectory,
                                                        cleanupScheduler,
                                                        Mockito.mock(SubscriptionRequestStorage.class),
                                                        shutdownNotifier);

        long minInterval_ms = 0;
        long ttl = 1000;
        testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters filterParameters = new testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters();
        OnChangeSubscriptionQos qos = new OnChangeSubscriptionQos().setMinIntervalMs(minInterval_ms)
                                                                   .setExpiryDateMs(SubscriptionQos.NO_EXPIRY_DATE)
                                                                   .setPublicationTtlMs(ttl);

        SubscriptionRequest subscriptionRequest = new BroadcastSubscriptionRequest(SUBSCRIPTION_ID,
                                                                                "subscribedToName",
                                                                                filterParameters,
                                                                                qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        GpsLocation eventValue = new GpsLocation();

        ArrayList<BroadcastFilter> filters = new ArrayList<BroadcastFilter>();
        testLocationUpdateSelectiveBroadcastFilter filterTrue = mock(testLocationUpdateSelectiveBroadcastFilter.class);
        when(filterTrue.filter(any(GpsLocation.class),
                               any(testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters.class))).thenReturn(true);
        filters.add(filterTrue);

        testLocationUpdateSelectiveBroadcastFilter filterFalse = mock(testLocationUpdateSelectiveBroadcastFilter.class);
        when(filterFalse.filter(any(GpsLocation.class),
                                any(testBroadcastInterface.LocationUpdateSelectiveBroadcastFilterParameters.class))).thenReturn(false);
        filters.add(filterFalse);

        publicationManager.broadcastOccurred(subscriptionRequest.getSubscriptionId(), filters, eventValue);

        verify(dispatcher, never()).sendSubscriptionPublication(any(String.class),
                                                                any(Set.class),
                                                                any(SubscriptionPublication.class),
                                                                any(MessagingQos.class));

    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000)
    public void modifySubscriptionTypeForExistingSubscription() throws Exception {
        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        providerDirectory,
                                                        cleanupScheduler,
                                                        Mockito.mock(SubscriptionRequestStorage.class),
                                                        shutdownNotifier);
        int period = 200;
        int testLengthMax = 3000;
        long validityMs = testLengthMax;
        long publicationTtl = testLengthMax;
        SubscriptionQos qos = new PeriodicSubscriptionQos().setPeriodMs(period)
                                                           .setValidityMs(validityMs)
                                                           .setPublicationTtlMs(publicationTtl);
        SubscriptionRequest subscriptionRequest = new SubscriptionRequest(SUBSCRIPTION_ID, "location", qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        verify(dispatcher, timeout(period * 5).times(6)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                                (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));

        qos = new OnChangeSubscriptionQos().setMinIntervalMs(0)
                                           .setValidityMs(validityMs)
                                           .setPublicationTtlMs(publicationTtl);
        subscriptionRequest = new SubscriptionRequest(SUBSCRIPTION_ID, "location", qos);

        when(providerDirectory.get(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(providerContainer);
        when(providerDirectory.contains(eq(PROVIDER_PARTICIPANT_ID))).thenReturn(true);

        publicationManager.addSubscriptionRequest(PROXY_PARTICIPANT_ID, PROVIDER_PARTICIPANT_ID, subscriptionRequest);

        reset(dispatcher);
        publicationManager.attributeValueChanged(SUBSCRIPTION_ID, valueToPublish);

        verify(dispatcher, timeout(testLengthMax).times(1)).sendSubscriptionPublication(eq(PROVIDER_PARTICIPANT_ID),
                                                                                (Set<String>) argThat(contains(PROXY_PARTICIPANT_ID)),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));
    }

    @Test
    public void testMulticastOccurred() {
        String providerParticipantId = "providerParticipantId";
        String multicastName = "multicastName";
        String[] partitions = new String[]{ "first", "second" };

        publicationManager.multicastOccurred(providerParticipantId, multicastName, partitions, "one", 1);

        verify(dispatcher).sendMulticast(eq(providerParticipantId),
                                         Mockito.<MulticastPublication> any(),
                                         Mockito.<MessagingQos> any());
    }

    @SuppressWarnings("unchecked")
    @Test(timeout = 3000000)
    public void persistedSubscriptionRequestsAreQueued() throws Exception {
        String persistenceFileName = "target/" + PublicationManagerTest.class.getCanonicalName()
                + ".test_persistenceSubscriptionRequests";

        String providerParticipantId = "providerParticipantId";
        String proxyParticipantId = "proxyParticipantId";
        int period = 200;
        int times = 5;
        int validityMs = (period * times) + period;
        long publicationTtl = validityMs;
        SubscriptionQos qos = new PeriodicSubscriptionQos().setPeriodMs(period)
                                                           .setValidityMs(validityMs)
                                                           .setPublicationTtlMs(publicationTtl);
        SubscriptionRequest subscriptionRequest = new SubscriptionRequest(SUBSCRIPTION_ID, "location", qos);

        new File(persistenceFileName).delete();

        // pre-fill the persistence file
        FileSubscriptionRequestStorage fileSubscriptionRequestStorage = new FileSubscriptionRequestStorage(persistenceFileName);
        assertEquals(0, fileSubscriptionRequestStorage.getSavedSubscriptionRequests().size());

        // no providers are currently registered
        ProviderDirectory myProviderDirectory = new ProviderDirectory();
        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        myProviderDirectory,
                                                        cleanupScheduler,
                                                        fileSubscriptionRequestStorage,
                                                        shutdownNotifier);

        publicationManager.addSubscriptionRequest(proxyParticipantId, providerParticipantId, subscriptionRequest);
        assertEquals(1, fileSubscriptionRequestStorage.getSavedSubscriptionRequests().size());

        publicationManager.shutdown();

        // open the persistence file that should now contain one element
        fileSubscriptionRequestStorage = new FileSubscriptionRequestStorage(persistenceFileName);
        assertEquals(1, fileSubscriptionRequestStorage.getSavedSubscriptionRequests().size());
        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        myProviderDirectory,
                                                        cleanupScheduler,
                                                        fileSubscriptionRequestStorage,
                                                        shutdownNotifier);

        // when the provider is registered, persisted subscription requests should be activated
        myProviderDirectory.add(providerParticipantId, providerContainer);

        verify(dispatcher, timeout(validityMs).atLeast(times)).sendSubscriptionPublication(any(String.class),
                                                                                any(Set.class),
                                                                                any(SubscriptionPublication.class),
                                                                                any(MessagingQos.class));
        ;

        Thread.sleep(validityMs + 1000);
        publicationManager.shutdown();

        // Start again with the same file, that should now be empty as all persisted subscriptions were already queued
        reset(dispatcher);
        fileSubscriptionRequestStorage = new FileSubscriptionRequestStorage(persistenceFileName);
        publicationManager = new PublicationManagerImpl(attributePollInterpreter,
                                                        dispatcher,
                                                        myProviderDirectory,
                                                        cleanupScheduler,
                                                        fileSubscriptionRequestStorage,
                                                        shutdownNotifier);
        verifyNoMoreInteractions(dispatcher);
        fileSubscriptionRequestStorage = new FileSubscriptionRequestStorage(persistenceFileName);
        assertEquals(0, fileSubscriptionRequestStorage.getSavedSubscriptionRequests().size());
    }
}