DedupCache { public List<AlertStreamEvent> dedup(AlertStreamEvent event, EventUniq eventEniq, String dedupStateField, String stateFieldValue, String stateCloseValue) { DedupValue[] dedupValues = this.addOrUpdate(eventEniq, event, stateFieldValue, stateCloseValue); if (dedupValues != null) { if (dedupValues.length == 2) { return Arrays.asList( this.mergeEventWithDedupValue(event, dedupValues[0], dedupStateField), this.mergeEventWithDedupValue(event, dedupValues[1], dedupStateField)); } else if (dedupValues.length == 1) { return Arrays.asList(this.mergeEventWithDedupValue(event, dedupValues[0], dedupStateField)); } } return null; } DedupCache(Config config, String publishName); Map<EventUniq, ConcurrentLinkedDeque<DedupValue>> getEvents(); boolean contains(EventUniq eventEniq); void removeEvent(EventUniq eventEniq); List<AlertStreamEvent> dedup(AlertStreamEvent event, EventUniq eventEniq,
                                        String dedupStateField, String stateFieldValue,
                                        String stateCloseValue); synchronized DedupValue[] addOrUpdate(EventUniq eventEniq, AlertStreamEvent event, String stateFieldValue, String stateCloseValue); static final String DEDUP_COUNT; static final String DOC_ID; static final String DEDUP_FIRST_OCCURRENCE; }
@Test public void testNormal() throws Exception { Config config = ConfigFactory.load(); DedupCache dedupCache = new DedupCache(config, "testPublishment"); StreamDefinition stream = createStream(); PolicyDefinition policy = createPolicy(stream.getStreamId(), "testPolicy"); String[] states = new String[] {"OPEN", "WARN", "CLOSE"}; Random random = new Random(); for (int i = 0; i < 20; i++) { AlertStreamEvent event = createEvent(stream, policy, new Object[] { System.currentTimeMillis(), "host1", "testPolicy-host1-01", states[random.nextInt(3)], 0, 0 }); HashMap<String, String> dedupFieldValues = new HashMap<String, String>(); dedupFieldValues.put("alertKey", (String) event.getData()[event.getSchema().getColumnIndex("alertKey")]); List<AlertStreamEvent> result = dedupCache.dedup(event, new EventUniq(event.getStreamId(), event.getPolicyId(), event.getCreatedTime(), dedupFieldValues), "state", (String) event.getData()[event.getSchema().getColumnIndex("state")], "closed"); System.out.println((i + 1) + " >>>> " + ToStringBuilder.reflectionToString(result)); } Assert.assertTrue(true); }