mirror of
https://github.com/Alfresco/alfresco-community-repo.git
synced 2025-07-31 17:39:05 +00:00
Bugfix/repo 5610 events are not actually sent to activemq (#360)
* Add events tests * Polished put test: connects to JMS via TCP and validate that the event sent is also received back * Now the tests provides a simple main() that listens on the topic, useful for quick debug sessions * Now the user name is collected in the calling thread, so that the sendEvent does not silently fails * Apply changes following review * Now using queue system to guarantee events order * Add license * Updated logs and corrected comments * Remove empty methods * Now catering for spurious events at startup when database is bootstrapped * Now preserving the txn-id in all events * Moved up definitions in events2.xml after PR feedback Co-authored-by: Bruno Bossola <bruno@meterian.com>
This commit is contained in:
Nana Insaidoo
GitHub
@@ -30,6 +30,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import javax.jms.ConnectionFactory;
|
||||
@@ -77,17 +78,19 @@ import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
public abstract class AbstractContextAwareRepoEvent extends BaseSpringTest
|
||||
{
|
||||
protected static final boolean DEBUG = false;
|
||||
|
||||
protected static final String TEST_NAMESPACE = "http://www.alfresco.org/test/ContextAwareRepoEvent";
|
||||
protected static final RepoEventContainer EVENT_CONTAINER = new RepoEventContainer();
|
||||
|
||||
private static final String BROKER_URL = "tcp://localhost:61616";
|
||||
private static final String TOPIC_NAME = "alfresco.repo.event2";
|
||||
private static final String CAMEL_ROUTE = "jms:topic:" + TOPIC_NAME;
|
||||
private static final RepoEventContainer EVENT_CONTAINER = new RepoEventContainer();
|
||||
private static final CamelContext CAMEL_CONTEXT = new DefaultCamelContext();
|
||||
|
||||
private static boolean isCamelConfigured;
|
||||
private static DataFormat dataFormat;
|
||||
|
||||
|
||||
@Autowired
|
||||
protected RetryingTransactionHelper retryingTransactionHelper;
|
||||
|
||||
@@ -104,6 +107,13 @@ public abstract class AbstractContextAwareRepoEvent extends BaseSpringTest
|
||||
@Autowired
|
||||
protected ObjectMapper event2ObjectMapper;
|
||||
|
||||
@Autowired @Qualifier("eventGeneratorV2")
|
||||
protected EventGenerator eventGenerator;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("eventGeneratorQueue")
|
||||
protected EventGeneratorQueue eventQueue;
|
||||
|
||||
protected NodeRef rootNodeRef;
|
||||
|
||||
@BeforeClass
|
||||
@@ -141,8 +151,35 @@ public abstract class AbstractContextAwareRepoEvent extends BaseSpringTest
|
||||
}
|
||||
return nodeService.getRootNode(storeRef);
|
||||
});
|
||||
|
||||
flushSpuriousEvents();
|
||||
}
|
||||
|
||||
/*
|
||||
* When running with an empty database some events related to the creation may
|
||||
* creep up here making the test fails. After attempting several other
|
||||
* strategies, a smart sleep seems to do the work.
|
||||
*/
|
||||
protected void flushSpuriousEvents() throws InterruptedException
|
||||
{
|
||||
int maxloops = 5;
|
||||
|
||||
int count = maxloops;
|
||||
do
|
||||
{
|
||||
Thread.sleep(165l);
|
||||
if (EVENT_CONTAINER.isEmpty())
|
||||
{
|
||||
count--;
|
||||
} else
|
||||
{
|
||||
EVENT_CONTAINER.reset();
|
||||
count = maxloops;
|
||||
}
|
||||
|
||||
} while (count > 0);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown()
|
||||
{
|
||||
@@ -179,6 +216,16 @@ public abstract class AbstractContextAwareRepoEvent extends BaseSpringTest
|
||||
propertyMap).getChildRef());
|
||||
}
|
||||
|
||||
protected NodeRef updateNodeName(NodeRef nodeRef, String newName)
|
||||
{
|
||||
PropertyMap propertyMap = new PropertyMap();
|
||||
propertyMap.put(ContentModel.PROP_NAME, newName);
|
||||
return retryingTransactionHelper.doInTransaction(() -> {
|
||||
nodeService.addProperties(nodeRef, propertyMap);
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
protected void deleteNode(NodeRef nodeRef)
|
||||
{
|
||||
retryingTransactionHelper.doInTransaction(() -> {
|
||||
@@ -376,13 +423,18 @@ public abstract class AbstractContextAwareRepoEvent extends BaseSpringTest
|
||||
|
||||
public static class RepoEventContainer implements Processor
|
||||
{
|
||||
private final List<RepoEvent<?>> events = new ArrayList<>();
|
||||
private final List<RepoEvent<?>> events = Collections.synchronizedList(new ArrayList<>());
|
||||
|
||||
@Override
|
||||
public void process(Exchange exchange)
|
||||
{
|
||||
Object object = exchange.getIn().getBody();
|
||||
events.add((RepoEvent<?>) object);
|
||||
|
||||
if (DEBUG)
|
||||
{
|
||||
System.err.println("XX: "+object);
|
||||
}
|
||||
}
|
||||
|
||||
public List<RepoEvent<?>> getEvents()
|
||||
@@ -404,6 +456,12 @@ public abstract class AbstractContextAwareRepoEvent extends BaseSpringTest
|
||||
{
|
||||
events.clear();
|
||||
}
|
||||
|
||||
public boolean isEmpty()
|
||||
{
|
||||
return events.isEmpty();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
|
||||
@@ -0,0 +1,290 @@
|
||||
/*
|
||||
* #%L
|
||||
* Alfresco Repository
|
||||
* %%
|
||||
* Copyright (C) 2005 - 2021 Alfresco Software Limited
|
||||
* %%
|
||||
* This file is part of the Alfresco software.
|
||||
* If the software was purchased under a paid Alfresco license, the terms of
|
||||
* the paid license agreement will prevail. Otherwise, the software is
|
||||
* provided under the following open source license terms:
|
||||
*
|
||||
* Alfresco is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Lesser General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* Alfresco is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License
|
||||
* along with Alfresco. If not, see <http://www.gnu.org/licenses/>.
|
||||
* #L%
|
||||
*/
|
||||
package org.alfresco.repo.event2;
|
||||
|
||||
import static java.lang.Thread.sleep;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.SynchronousQueue;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.alfresco.repo.event.v1.model.RepoEvent;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import org.mockito.invocation.InvocationOnMock;
|
||||
import org.mockito.stubbing.Answer;
|
||||
|
||||
public class EventGeneratorQueueUnitTest
|
||||
{
|
||||
private EventGeneratorQueue queue;
|
||||
|
||||
private Event2MessageProducer bus;
|
||||
private ExecutorService enqueuePool;
|
||||
private ExecutorService dequeuePool;
|
||||
private List<RepoEvent<?>> recordedEvents;
|
||||
private Map<String, RepoEvent<?>> events;
|
||||
|
||||
@Before
|
||||
public void setup()
|
||||
{
|
||||
queue = new EventGeneratorQueue();
|
||||
|
||||
enqueuePool = newThreadPool();
|
||||
queue.setEnqueueThreadPoolExecutor(enqueuePool);
|
||||
dequeuePool = newThreadPool();
|
||||
queue.setDequeueThreadPoolExecutor(dequeuePool);
|
||||
|
||||
bus = mock(Event2MessageProducer.class);
|
||||
queue.setEvent2MessageProducer(bus);
|
||||
|
||||
events = new HashMap<>();
|
||||
|
||||
setupEventsRecorder();
|
||||
}
|
||||
|
||||
@After
|
||||
public void teardown()
|
||||
{
|
||||
enqueuePool.shutdown();
|
||||
}
|
||||
|
||||
private void setupEventsRecorder()
|
||||
{
|
||||
recordedEvents = new CopyOnWriteArrayList<>();
|
||||
|
||||
Mockito.doAnswer(new Answer<Void>()
|
||||
{
|
||||
@Override
|
||||
public Void answer(InvocationOnMock invocation) throws Throwable
|
||||
{
|
||||
RepoEvent<?> event = invocation.getArgument(0, RepoEvent.class);
|
||||
recordedEvents.add(event);
|
||||
return null;
|
||||
}
|
||||
}).when(bus).send(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveSingleQuickMessage() throws Exception
|
||||
{
|
||||
queue.accept(messageWithDelay("A", 55l));
|
||||
|
||||
sleep(150l);
|
||||
|
||||
assertEquals(1, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldNotReceiveEventsWhenMessageIsNull() throws Exception
|
||||
{
|
||||
queue.accept(() -> { return null; });
|
||||
|
||||
sleep(150l);
|
||||
|
||||
assertEquals(0, recordedEvents.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveMultipleMessagesPreservingOrderScenarioOne() throws Exception {
|
||||
queue.accept(messageWithDelay("A", 0l));
|
||||
queue.accept(messageWithDelay("B", 100l));
|
||||
queue.accept(messageWithDelay("C", 200l));
|
||||
|
||||
sleep(450l);
|
||||
|
||||
assertEquals(3, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
assertEquals("B", recordedEvents.get(1).getId());
|
||||
assertEquals("C", recordedEvents.get(2).getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveMultipleMessagesPreservingOrderScenarioTwo() throws Exception
|
||||
{
|
||||
queue.accept(messageWithDelay("A", 300l));
|
||||
queue.accept(messageWithDelay("B", 150l));
|
||||
queue.accept(messageWithDelay("C", 0l));
|
||||
|
||||
sleep(950l);
|
||||
|
||||
assertEquals(3, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
assertEquals("B", recordedEvents.get(1).getId());
|
||||
assertEquals("C", recordedEvents.get(2).getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveMultipleMessagesPreservingOrderEvenWhenMakerPoisoned() throws Exception
|
||||
{
|
||||
queue.accept(messageWithDelay("A", 300l));
|
||||
queue.accept(() -> {throw new RuntimeException("Boom! (not to worry, this is a test)");});
|
||||
queue.accept(messageWithDelay("B", 55l));
|
||||
queue.accept(messageWithDelay("C", 0l));
|
||||
|
||||
sleep(950l);
|
||||
|
||||
assertEquals(3, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
assertEquals("B", recordedEvents.get(1).getId());
|
||||
assertEquals("C", recordedEvents.get(2).getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveMultipleMessagesPreservingOrderEvenWhenSenderPoisoned() throws Exception
|
||||
{
|
||||
Callable<RepoEvent<?>> makerB = messageWithDelay("B", 55l);
|
||||
RepoEvent<?> messageB = makerB.call();
|
||||
doThrow(new RuntimeException("Boom! (not to worry, this is a test)")).when(bus).send(messageB);
|
||||
queue.accept(messageWithDelay("A", 300l));
|
||||
queue.accept(makerB);
|
||||
queue.accept(messageWithDelay("C", 0l));
|
||||
|
||||
sleep(950l);
|
||||
|
||||
assertEquals(2, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
assertEquals("C", recordedEvents.get(1).getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveMultipleMessagesPreservingOrderEvenWhenMakerPoisonedWithError() throws Exception
|
||||
{
|
||||
queue.accept(messageWithDelay("A", 300l));
|
||||
queue.accept(() -> {throw new OutOfMemoryError("Boom! (not to worry, this is a test)");});
|
||||
queue.accept(messageWithDelay("B", 55l));
|
||||
queue.accept(messageWithDelay("C", 0l));
|
||||
|
||||
sleep(950l);
|
||||
|
||||
assertEquals(3, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
assertEquals("B", recordedEvents.get(1).getId());
|
||||
assertEquals("C", recordedEvents.get(2).getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveMultipleMessagesPreservingOrderEvenWhenSenderPoisonedWithError() throws Exception
|
||||
{
|
||||
Callable<RepoEvent<?>> makerB = messageWithDelay("B", 55l);
|
||||
RepoEvent<?> messageB = makerB.call();
|
||||
doThrow(new OutOfMemoryError("Boom! (not to worry, this is a test)")).when(bus).send(messageB);
|
||||
queue.accept(messageWithDelay("A", 300l));
|
||||
queue.accept(makerB);
|
||||
queue.accept(messageWithDelay("C", 0l));
|
||||
|
||||
sleep(950l);
|
||||
|
||||
assertEquals(2, recordedEvents.size());
|
||||
assertEquals("A", recordedEvents.get(0).getId());
|
||||
assertEquals("C", recordedEvents.get(1).getId());
|
||||
}
|
||||
|
||||
private Callable<RepoEvent<?>> messageWithDelay(String id, long delay)
|
||||
{
|
||||
Callable<RepoEvent<?>> res = new Callable<RepoEvent<?>>() {
|
||||
|
||||
@Override
|
||||
public RepoEvent<?> call() throws Exception
|
||||
{
|
||||
if(delay != 0)
|
||||
{
|
||||
sleep(delay);
|
||||
}
|
||||
return newRepoEvent(id);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString()
|
||||
{
|
||||
return id;
|
||||
}
|
||||
};
|
||||
return res;
|
||||
}
|
||||
|
||||
private RepoEvent<?> newRepoEvent(String id)
|
||||
{
|
||||
RepoEvent<?> ev = events.get(id);
|
||||
if (ev!=null)
|
||||
return ev;
|
||||
|
||||
ev = mock(RepoEvent.class);
|
||||
when(ev.getId()).thenReturn(id);
|
||||
when(ev.toString()).thenReturn(id);
|
||||
events.put(id, ev);
|
||||
|
||||
return ev;
|
||||
}
|
||||
|
||||
public static ExecutorService newThreadPool()
|
||||
{
|
||||
return new ThreadPoolExecutor(2, Integer.MAX_VALUE,
|
||||
60L, TimeUnit.SECONDS,
|
||||
new SynchronousQueue<Runnable>());
|
||||
}
|
||||
|
||||
public static final Executor SYNC_EXECUTOR_SAME_THREAD = new Executor()
|
||||
{
|
||||
@Override
|
||||
public void execute(Runnable command)
|
||||
{
|
||||
command.run();
|
||||
}
|
||||
};
|
||||
|
||||
public static final Executor SYNC_EXECUTOR_NEW_THREAD = new Executor()
|
||||
{
|
||||
@Override
|
||||
public void execute(Runnable command)
|
||||
{
|
||||
Thread t = new Thread(command);
|
||||
t.start();
|
||||
try
|
||||
{
|
||||
t.join();
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,249 @@
|
||||
/*
|
||||
* #%L
|
||||
* Alfresco Repository
|
||||
* %%
|
||||
* Copyright (C) 2005 - 2020 Alfresco Software Limited
|
||||
* %%
|
||||
* This file is part of the Alfresco software.
|
||||
* If the software was purchased under a paid Alfresco license, the terms of
|
||||
* the paid license agreement will prevail. Otherwise, the software is
|
||||
* provided under the following open source license terms:
|
||||
*
|
||||
* Alfresco is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Lesser General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* Alfresco is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License
|
||||
* along with Alfresco. If not, see <http://www.gnu.org/licenses/>.
|
||||
* #L%
|
||||
*/
|
||||
package org.alfresco.repo.event2;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import javax.jms.Destination;
|
||||
import javax.jms.JMSException;
|
||||
import javax.jms.Message;
|
||||
import javax.jms.MessageConsumer;
|
||||
import javax.jms.MessageListener;
|
||||
import javax.jms.Session;
|
||||
|
||||
import org.alfresco.model.ContentModel;
|
||||
import org.alfresco.repo.event.databind.ObjectMapperFactory;
|
||||
import org.alfresco.repo.event.v1.model.RepoEvent;
|
||||
import org.alfresco.service.cmr.repository.NodeRef;
|
||||
import org.apache.activemq.ActiveMQConnection;
|
||||
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||
import org.apache.activemq.advisory.DestinationSource;
|
||||
import org.apache.activemq.command.ActiveMQQueue;
|
||||
import org.apache.activemq.command.ActiveMQTextMessage;
|
||||
import org.apache.activemq.command.ActiveMQTopic;
|
||||
import org.awaitility.Awaitility;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
public class EventGeneratorTest extends AbstractContextAwareRepoEvent
|
||||
{
|
||||
private static final String EVENT2_TOPIC_NAME = "alfresco.repo.event2";
|
||||
|
||||
private static final long DUMP_BROKER_TIMEOUT = 50000000l;
|
||||
|
||||
@Autowired @Qualifier("event2ObjectMapper")
|
||||
private ObjectMapper objectMapper;
|
||||
|
||||
private ActiveMQConnection connection;
|
||||
protected List<RepoEvent<?>> receivedEvents;
|
||||
|
||||
@Before
|
||||
public void startupTopicListener() throws Exception
|
||||
{
|
||||
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
|
||||
connection = (ActiveMQConnection) connectionFactory.createConnection();
|
||||
connection.start();
|
||||
|
||||
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
Destination destination = session.createTopic(EVENT2_TOPIC_NAME);
|
||||
MessageConsumer consumer = session.createConsumer(destination);
|
||||
|
||||
receivedEvents = Collections.synchronizedList(new LinkedList<>());
|
||||
consumer.setMessageListener(new MessageListener()
|
||||
{
|
||||
@Override
|
||||
public void onMessage(Message message)
|
||||
{
|
||||
String text = getText(message);
|
||||
RepoEvent<?> event = toRepoEvent(text);
|
||||
|
||||
if (DEBUG)
|
||||
{
|
||||
System.err.println("RX: " + event);
|
||||
}
|
||||
|
||||
receivedEvents.add(event);
|
||||
}
|
||||
|
||||
private RepoEvent<?> toRepoEvent(String json)
|
||||
{
|
||||
try
|
||||
{
|
||||
return objectMapper.readValue(json, RepoEvent.class);
|
||||
} catch (Exception e)
|
||||
{
|
||||
e.printStackTrace();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
if (DEBUG)
|
||||
{
|
||||
System.err.println("Now actively listening on topic " + EVENT2_TOPIC_NAME);
|
||||
}
|
||||
}
|
||||
|
||||
protected ObjectMapper createObjectMapper()
|
||||
{
|
||||
return ObjectMapperFactory.createInstance();
|
||||
}
|
||||
|
||||
@After
|
||||
public void shutdownTopicListener() throws Exception
|
||||
{
|
||||
connection.close();
|
||||
connection = null;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveEvent2EventsOnNodeCreation() throws Exception
|
||||
{
|
||||
createNode(ContentModel.TYPE_CONTENT);
|
||||
|
||||
Awaitility.await().atMost(6, TimeUnit.SECONDS).until(() -> receivedEvents.size() == 1);
|
||||
|
||||
RepoEvent<?> sent = getRepoEvent(1);
|
||||
RepoEvent<?> received = receivedEvents.get(0);
|
||||
assertEventsEquals("Events are different!", sent, received);
|
||||
}
|
||||
|
||||
private void assertEventsEquals(String message, RepoEvent<?> expected, RepoEvent<?> current)
|
||||
{
|
||||
if (DEBUG)
|
||||
{
|
||||
System.err.println("XP: " + expected);
|
||||
System.err.println("CU: " + current);
|
||||
}
|
||||
|
||||
assertEquals(message, expected, current);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReceiveEvent2EventsInOrder() throws Exception
|
||||
{
|
||||
NodeRef nodeRef = createNode(ContentModel.TYPE_CONTENT);
|
||||
updateNodeName(nodeRef, "TestFile-" + System.currentTimeMillis() + ".txt");
|
||||
deleteNode(nodeRef);
|
||||
|
||||
Awaitility.await().atMost(6, TimeUnit.SECONDS).until(() -> receivedEvents.size() == 3);
|
||||
|
||||
RepoEvent<?> sentCreation = getRepoEvent(1);
|
||||
RepoEvent<?> sentUpdate = getRepoEvent(2);
|
||||
RepoEvent<?> sentDeletion = getRepoEvent(3);
|
||||
assertEquals("Expected create event!", sentCreation, (RepoEvent<?>) receivedEvents.get(0));
|
||||
assertEquals("Expected update event!", sentUpdate, (RepoEvent<?>) receivedEvents.get(1));
|
||||
assertEquals("Expected delete event!", sentDeletion, (RepoEvent<?>) receivedEvents.get(2));
|
||||
}
|
||||
|
||||
private static String getText(Message message)
|
||||
{
|
||||
try
|
||||
{
|
||||
ActiveMQTextMessage am = (ActiveMQTextMessage) message;
|
||||
return am.getText();
|
||||
} catch (JMSException e)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// a simple main to investigate the contents of the local broker
|
||||
public static void main(String[] args) throws Exception
|
||||
{
|
||||
dumpBroker("tcp://localhost:61616", DUMP_BROKER_TIMEOUT);
|
||||
System.exit(0);
|
||||
}
|
||||
|
||||
private static void dumpBroker(String url, long timeout) throws Exception
|
||||
{
|
||||
System.out.println("Broker at url: '" + url + "'");
|
||||
|
||||
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
|
||||
ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
|
||||
try
|
||||
{
|
||||
connection.start();
|
||||
|
||||
DestinationSource ds = connection.getDestinationSource();
|
||||
|
||||
Set<ActiveMQQueue> queues = ds.getQueues();
|
||||
System.out.println("\nFound " + queues.size() + " queues:");
|
||||
for (ActiveMQQueue queue : queues)
|
||||
{
|
||||
try
|
||||
{
|
||||
System.out.println("- " + queue.getQueueName());
|
||||
} catch (JMSException e)
|
||||
{
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
Set<ActiveMQTopic> topics = ds.getTopics();
|
||||
System.out.println("\nFound " + topics.size() + " topics:");
|
||||
for (ActiveMQTopic topic : topics)
|
||||
{
|
||||
try
|
||||
{
|
||||
System.out.println("- " + topic.getTopicName());
|
||||
} catch (JMSException e)
|
||||
{
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
Destination destination = session.createTopic(EVENT2_TOPIC_NAME);
|
||||
MessageConsumer consumer = session.createConsumer(destination);
|
||||
|
||||
System.out.println("\nListening to topic " + EVENT2_TOPIC_NAME + "...");
|
||||
consumer.setMessageListener(new MessageListener()
|
||||
{
|
||||
@Override
|
||||
public void onMessage(Message message)
|
||||
{
|
||||
String text = getText(message);
|
||||
System.out.println("Received message " + message + "\n" + text + "\n");
|
||||
}
|
||||
});
|
||||
|
||||
Thread.sleep(timeout);
|
||||
} finally
|
||||
{
|
||||
connection.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -34,7 +34,8 @@ import org.junit.runners.Suite.SuiteClasses;
|
||||
UpdateRepoEventIT.class,
|
||||
DeleteRepoEventIT.class,
|
||||
ChildAssociationRepoEventIT.class,
|
||||
PeerAssociationRepoEventIT.class
|
||||
PeerAssociationRepoEventIT.class,
|
||||
EventGeneratorTest.class
|
||||
})
|
||||
public class RepoEvent2ITSuite
|
||||
{
|
||||
|
||||
@@ -33,7 +33,8 @@ import org.junit.runners.Suite.SuiteClasses;
|
||||
@RunWith(Suite.class)
|
||||
@SuiteClasses({ EventFilterUnitTest.class,
|
||||
EventConsolidatorUnitTest.class,
|
||||
EventJSONSchemaUnitTest.class
|
||||
EventJSONSchemaUnitTest.class,
|
||||
EventGeneratorQueueUnitTest.class
|
||||
})
|
||||
public class RepoEvent2UnitSuite
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user