Кажется, этот конфлятор делает то, что вы ищете:
https://github.com/GuillaumeArnaud/conflator
В зависимости от ваших требований, реализация может быть изменена, чтобы объединить ИЛИ заменить последнее событие существующим событием, если оно существует в очереди соединения.
Например. для следующего каждое событие реализовано как «Тик», который определяет поведение слияния.
public class Tick implements Message<Tick> {
private final String ticker;
public long getInitialQuantity() {
return initialQuantity;
}
private final long initialQuantity;
public long getCurrentQuantity() {
return currentQuantity;
}
private long currentQuantity;
private int numberOfMerges;
public String getTicker() {
return ticker;
}
public Tick(String ticker, long quantity) {
this.ticker = ticker;
this.initialQuantity = quantity;
this.currentQuantity = quantity;
}
@Override
public String key() {
return this.ticker;
}
@Override
public String body() {
return String.valueOf(currentQuantity);
}
@Override
public boolean isMerged() {
return this.initialQuantity != this.currentQuantity;
}
@Override
public int mergesCount() {
return numberOfMerges;
}
@Override
public boolean isValid() {
return false;
}
@Override
public boolean merge(Tick message) {
if (this.equals(message)) {
this.currentQuantity += message.currentQuantity;
numberOfMerges++;
return true;
}
return false;
}
@Override
public int hashCode() {
return ticker.hashCode();
}
@Override
public boolean equals(Object obj) {
if (obj != null && obj instanceof Tick) {
Tick other = (Tick) obj;
return this.ticker.equals(other.getTicker());
}
return false;
}
Контрольный пример:
public class TickMergeTest {
MultiValuedMapConflator conflator;
@Test
public void two_unmergeable_ticks_should_be_remain_unmergeable() {
Tick tick1 = new Tick("GOOG", 100L);
Tick tick2 = new Tick("AAPL", 120L);
List<Tick> messages = conflator.merge(Lists.newArrayList(tick1, tick2));
assertNotNull(messages);
assertEquals(messages.size(), 2);
assertEquals(Long.valueOf(messages.get(0).body()).longValue(), tick1.getCurrentQuantity());
assertEquals(Long.valueOf(messages.get(1).body()).longValue(), tick2.getCurrentQuantity());
}
@Test(timeout = 1000)
public void two_mergeable_ticks_should_be_merged() {
Tick tick1 = new Tick("GOOG", 100L);
Tick tick2 = new Tick("GOOG", 120L);
List<Tick> messages = conflator.merge(Lists.newArrayList(tick1, tick2));
assertNotNull(messages);
assertEquals(messages.size(), 1);
assertEquals(Long.valueOf(messages.get(0).body()).longValue(), tick1.getInitialQuantity() + tick2.getInitialQuantity());
}
@Test(timeout = 1000)
public void should_merge_messages_on_same_key() throws InterruptedException {
// given
conflator.put(new Tick("GOOG", 100L));
conflator.put(new Tick("GOOG", 120L));
// test
Thread.sleep(300); // waiting the conflation
Message message = conflator.take();
// check
assertNotNull(message);
assertEquals(Long.valueOf(message.body()).longValue(), 220L);
assertTrue(message.isMerged());
}
@Test(timeout = 1000)
public void should_not_merge_messages_on_diff_key() throws InterruptedException {
// given
conflator.put(new Tick("GOOG", 100L));
conflator.put(new Tick("AAPL", 120L));
// test
Thread.sleep(300); // waiting the conflation
Message message1 = conflator.take();
Message message2 = conflator.take();
// check
assertNotNull(message1);
assertNotNull(message2);
assertEquals(Long.valueOf(message1.body()).longValue(), 100L);
assertFalse(message1.isMerged());
assertEquals(Long.valueOf(message2.body()).longValue(), 120L);
assertFalse(message2.isMerged());
}
@Before
public void setUp() {
conflator = new MultiValuedMapConflator<Tick>(true);
}
@After
public void tearDown() {
conflator.stop();
}
}