EsEventPersistence implements EventPersistence { @Override public String persist(final Event event) { String evStr = null; try { evStr = jsonMapper.writeValueAsString(event); final IndexRequest indexRequest = Requests .indexRequest(indexNamingStrategy.buildActiveName(event.getSnifferId())) .type(getType(event.getSnifferId())).source(evStr); final String eventId = clientTpl.executeWithClient(new ClientCallback<IndexResponse>() { @Override public IndexResponse execute(final Client client) { return client.index(indexRequest).actionGet(); } }).getId(); logger.debug("Persisted event with id: {}", eventId); return eventId; } catch (final Exception e) { throw new DataAccessException("Failed to persiste event: " + evStr, e); } }  static String getType(final long snifferId); @Override String persist(final Event event); @Override void delete(final long snifferId, final String[] eventIds); @Override void deleteAll(final long snifferId); @Override EventQueryBuilder getEventsQueryBuilder(final long snifferId, final long offset, final int limit); @Override NativeQueryBuilder getEventsNativeQueryBuilder(final long snifferId, final long offset, final int limit); @Override Event getEvent(final long snifferId, final String eventId); @Override AspectProvider<AspectSniffer, Integer> getEventsCounter(); @EventListener void handleOrderCreatedEvent(final SnifferChangedEvent event); static final String EVENTS_COUNT; }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testPersist() throws Exception { final Event e = new Event(); e.setLogPath("log"); e.setSnifferId(sniffer1.getId()); e.setLogSourceId(source1.getId()); final LogEntry entry1 = new LogEntry(); entry1.setRawContent("1"); entry1.setStartOffset(new DefaultPointer(0, 1)); entry1.setEndOffset(new DefaultPointer(1, 1)); entry1.put("f1", new Date(0)); final LogEntry entry2 = new LogEntry(); entry2.setStartOffset(new DefaultPointer(1, 2)); entry2.setEndOffset(new DefaultPointer(2, 2)); entry2.setRawContent("2"); final ArrayList<LogEntry> entries = new ArrayList<LogEntry>(); entries.add(entry1); entries.add(entry2); e.setEntries(entries); e.setPublished(new Date(1000 * 100)); e.put("my", "value"); final String eventId = persister.persist(e); logger.info("Serialized event as: {}", objectMapper.writeValueAsString(e)); clientTpl.executeWithClient(new ClientCallback<Object>() { @Override public Object execute(final Client client) { client.admin().indices().prepareRefresh().get(); return null; } }); Assert.assertEquals(1, persister.getEventsQueryBuilder(sniffer1.getId(), 0, 10).list().getItems().size()); Assert.assertEquals(1, persister.getEventsQueryBuilder(sniffer1.getId(), 0, 10).list().getTotalCount()); final Event checkEvent = persister.getEvent(sniffer1.getId(), eventId); Assert.assertEquals(sniffer1.getId(), checkEvent.getSnifferId()); Assert.assertEquals(source1.getId(), checkEvent.getLogSourceId()); Assert.assertEquals("log", checkEvent.getLogPath()); Assert.assertEquals(1000 * 100, checkEvent.getPublished().getTime()); Assert.assertEquals(2, checkEvent.getEntries().size()); Assert.assertEquals("1", checkEvent.getEntries().get(0).getRawContent()); Assert.assertEquals(new Date(0), checkEvent.getEntries().get(0).get("f1")); Assert.assertEquals(entry1.getStartOffset().getJson(), checkEvent.getEntries().get(0).getStartOffset().getJson()); Assert.assertEquals("2", checkEvent.getEntries().get(1).getRawContent()); Assert.assertEquals("value", checkEvent.get("my")); Assert.assertEquals(0, persister.getEventsQueryBuilder(sniffer1.getId(), 1, 10).list().getItems().size()); Assert.assertEquals(1, persister.getEventsQueryBuilder(sniffer1.getId(), 1, 10).list().getTotalCount()); Assert.assertNotNull(persister.getEvent(sniffer1.getId(), eventId)); persister.delete(sniffer1.getId(), new String[] { eventId }); Assert.assertNull(persister.getEvent(sniffer1.getId(), eventId)); Mockito.when(snifferPersistence.getSniffer(sniffer1.getId())).thenReturn(sniffer1); Mockito.when(sourceProvider.getSourceById(sniffer1.getLogSourceId())).thenReturn((LogSource) source1); persister.deleteAll(sniffer1.getId()); persister.deleteAll(sniffer1.getId()); }