InvertIndexArray extends BaseKeyValueOperator<K,V> { @Override public void endWindow() { for (Map.Entry<V, ArrayList<K>> e: map.entrySet()) { HashMap<V, ArrayList<K>> tuple = new HashMap<V, ArrayList<K>>(1); tuple.put(e.getKey(), e.getValue()); index.emit(tuple); } map.clear(); }  @Override void endWindow(); final transient DefaultInputPort<HashMap<K, ArrayList<V>>> data; final transient DefaultOutputPort<HashMap<V, ArrayList<K>>> index; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws Exception { InvertIndexArray<String,String> oper = new InvertIndexArray<String,String>(); CollectorTestSink indexSink = new CollectorTestSink(); Sink inSink = oper.data.getSink(); oper.index.setSink(indexSink); oper.beginWindow(0); HashMap<String, ArrayList> input = new HashMap<String, ArrayList>(); ArrayList<String> alist = new ArrayList<String>(); alist.add("str"); alist.add("str1"); input.put("a", alist); input.put("b", alist); inSink.put(input); alist = new ArrayList<String>(); input = new HashMap<String, ArrayList>(); alist.add("blah"); alist.add("str1"); input.put("c", alist); inSink.put(input); oper.endWindow(); Assert.assertEquals("number emitted tuples", 3, indexSink.collectedTuples.size()); for (Object o: indexSink.collectedTuples) { log.debug(o.toString()); HashMap<String, ArrayList<String>> output = (HashMap<String, ArrayList<String>>)o; for (Map.Entry<String, ArrayList<String>> e: output.entrySet()) { String key = e.getKey(); alist = e.getValue(); if (key.equals("str1")) { Assert.assertEquals("Index for \"str1\" contains \"a\"", true, alist.contains("a")); Assert.assertEquals("Index for \"str1\" contains \"b\"", true, alist.contains("b")); Assert.assertEquals("Index for \"str1\" contains \"c\"", true, alist.contains("c")); } else if (key.equals("str")) { Assert.assertEquals("Index for \"str1\" contains \"a\"", true, alist.contains("a")); Assert.assertEquals("Index for \"str1\" contains \"b\"", true, alist.contains("b")); Assert.assertEquals("Index for \"str1\" contains \"c\"", false, alist.contains("c")); } else if (key.equals("blah")) { Assert.assertEquals("Index for \"str1\" contains \"a\"", false, alist.contains("a")); Assert.assertEquals("Index for \"str1\" contains \"b\"", false, alist.contains("b")); Assert.assertEquals("Index for \"str1\" contains \"c\"", true, alist.contains("c")); } } } }