@Test public void testNoMatch() { Map<String, Object> values = new HashMap<String, Object>(); values.put("id", SupportConst.I_FIVE); values.put("description", "item:"); values.put("price", (double)SupportConst.I_TWENTY); newDataPerStream = new IEvent[SupportConst.I_TWO][]; newDataPerStream[0] = new IEvent[1]; newDataPerStream[1] = new IEvent[1]; newDataPerStream[0][0] = new TupleEvent("item", item, values); result = crossJoin.join(newDataPerStream, null); Set<MultiKey> expSet = new LinkedHashSet<MultiKey>(); Object[] obj = new Object[SupportConst.I_TWO]; obj[0] = newDataPerStream[0][0]; Object[] att = new Object[sell.getSize()]; obj[1] = new TupleEvent("sell", sell, att); expSet.add(new MultiKey(obj)); SupportAssertUtil.assertEuqalsArrayValue(expSet.toArray(new MultiKey[expSet.size()]), result.getFirst() .toArray(new MultiKey[result.getFirst().size()])); }
@Override public Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream) { if (null == newDataPerStream && null == oldDataPerStream) { throw new RuntimeException("Mission impossible."); } if ((null != newDataPerStream && newDataPerStream.length != STREAM_NUM) || (null != oldDataPerStream && oldDataPerStream.length != STREAM_NUM)) { LOG.error("This updated streams numbers is not for Cross Bi Stream Join."); throw new RuntimeException("This updated streams numbers is not for Bi Stream Join."); } newResults.clear(); oldResults.clear(); if (joinRStream) { if (null != oldDataPerStream) { for (int streamIndex = 0; streamIndex < oldDataPerStream.length; streamIndex++) { compose(oldDataPerStream[streamIndex], streamIndex, oldResults); } } } if (null != newDataPerStream) { for (int streamIndex = 0; streamIndex < newDataPerStream.length; streamIndex++) { compose(newDataPerStream[streamIndex], streamIndex, newResults); } } return new Pair<Set<MultiKey>, Set<MultiKey>>(newResults, oldResults); }
CrossBiJoinComposer implements IJoinComposer { @Override public Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream) { if (null == newDataPerStream && null == oldDataPerStream) { throw new RuntimeException("Mission impossible."); } if ((null != newDataPerStream && newDataPerStream.length != STREAM_NUM) || (null != oldDataPerStream && oldDataPerStream.length != STREAM_NUM)) { LOG.error("This updated streams numbers is not for Cross Bi Stream Join."); throw new RuntimeException("This updated streams numbers is not for Bi Stream Join."); } newResults.clear(); oldResults.clear(); if (joinRStream) { if (null != oldDataPerStream) { for (int streamIndex = 0; streamIndex < oldDataPerStream.length; streamIndex++) { compose(oldDataPerStream[streamIndex], streamIndex, oldResults); } } } if (null != newDataPerStream) { for (int streamIndex = 0; streamIndex < newDataPerStream.length; streamIndex++) { compose(newDataPerStream[streamIndex], streamIndex, newResults); } } return new Pair<Set<MultiKey>, Set<MultiKey>>(newResults, oldResults); } }
CrossBiJoinComposer implements IJoinComposer { @Override public Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream) { if (null == newDataPerStream && null == oldDataPerStream) { throw new RuntimeException("Mission impossible."); } if ((null != newDataPerStream && newDataPerStream.length != STREAM_NUM) || (null != oldDataPerStream && oldDataPerStream.length != STREAM_NUM)) { LOG.error("This updated streams numbers is not for Cross Bi Stream Join."); throw new RuntimeException("This updated streams numbers is not for Bi Stream Join."); } newResults.clear(); oldResults.clear(); if (joinRStream) { if (null != oldDataPerStream) { for (int streamIndex = 0; streamIndex < oldDataPerStream.length; streamIndex++) { compose(oldDataPerStream[streamIndex], streamIndex, oldResults); } } } if (null != newDataPerStream) { for (int streamIndex = 0; streamIndex < newDataPerStream.length; streamIndex++) { compose(newDataPerStream[streamIndex], streamIndex, newResults); } } return new Pair<Set<MultiKey>, Set<MultiKey>>(newResults, oldResults); } CrossBiJoinComposer(SimpleEventCollection left, SimpleEventCollection right, boolean rStream); }
CrossBiJoinComposer implements IJoinComposer { @Override public Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream) { if (null == newDataPerStream && null == oldDataPerStream) { throw new RuntimeException("Mission impossible."); } if ((null != newDataPerStream && newDataPerStream.length != STREAM_NUM) || (null != oldDataPerStream && oldDataPerStream.length != STREAM_NUM)) { LOG.error("This updated streams numbers is not for Cross Bi Stream Join."); throw new RuntimeException("This updated streams numbers is not for Bi Stream Join."); } newResults.clear(); oldResults.clear(); if (joinRStream) { if (null != oldDataPerStream) { for (int streamIndex = 0; streamIndex < oldDataPerStream.length; streamIndex++) { compose(oldDataPerStream[streamIndex], streamIndex, oldResults); } } } if (null != newDataPerStream) { for (int streamIndex = 0; streamIndex < newDataPerStream.length; streamIndex++) { compose(newDataPerStream[streamIndex], streamIndex, newResults); } } return new Pair<Set<MultiKey>, Set<MultiKey>>(newResults, oldResults); } CrossBiJoinComposer(SimpleEventCollection left, SimpleEventCollection right, boolean rStream); @Override Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream); @Override void maintainData(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream); @Override int getStreamsSize(); SimpleEventCollection getLeftStream(); SimpleEventCollection getRightStream(); }
CrossBiJoinComposer implements IJoinComposer { @Override public Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream) { if (null == newDataPerStream && null == oldDataPerStream) { throw new RuntimeException("Mission impossible."); } if ((null != newDataPerStream && newDataPerStream.length != STREAM_NUM) || (null != oldDataPerStream && oldDataPerStream.length != STREAM_NUM)) { LOG.error("This updated streams numbers is not for Cross Bi Stream Join."); throw new RuntimeException("This updated streams numbers is not for Bi Stream Join."); } newResults.clear(); oldResults.clear(); if (joinRStream) { if (null != oldDataPerStream) { for (int streamIndex = 0; streamIndex < oldDataPerStream.length; streamIndex++) { compose(oldDataPerStream[streamIndex], streamIndex, oldResults); } } } if (null != newDataPerStream) { for (int streamIndex = 0; streamIndex < newDataPerStream.length; streamIndex++) { compose(newDataPerStream[streamIndex], streamIndex, newResults); } } return new Pair<Set<MultiKey>, Set<MultiKey>>(newResults, oldResults); } CrossBiJoinComposer(SimpleEventCollection left, SimpleEventCollection right, boolean rStream); @Override Pair<Set<MultiKey>, Set<MultiKey>> join(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream); @Override void maintainData(IEvent[][] newDataPerStream, IEvent[][] oldDataPerStream); @Override int getStreamsSize(); SimpleEventCollection getLeftStream(); SimpleEventCollection getRightStream();  }