Source code

001/*
002 * Copyright 2015 DuraSpace, Inc.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016package org.fcrepo.integration.jms.observer;
017
018import static com.google.common.base.Throwables.propagate;
019import static com.jayway.awaitility.Awaitility.await;
020import static com.jayway.awaitility.Duration.ONE_SECOND;
021import static javax.jms.Session.AUTO_ACKNOWLEDGE;
022import static org.fcrepo.jms.headers.DefaultMessageFactory.BASE_URL_HEADER_NAME;
023import static org.fcrepo.jms.headers.DefaultMessageFactory.EVENT_TYPE_HEADER_NAME;
024import static org.fcrepo.jms.headers.DefaultMessageFactory.IDENTIFIER_HEADER_NAME;
025import static org.fcrepo.jms.headers.DefaultMessageFactory.PROPERTIES_HEADER_NAME;
026import static org.fcrepo.jms.headers.DefaultMessageFactory.TIMESTAMP_HEADER_NAME;
027import static org.fcrepo.kernel.api.RdfLexicon.HAS_SIZE;
028import static org.fcrepo.kernel.api.RdfLexicon.REPOSITORY_NAMESPACE;
029import static org.fcrepo.kernel.api.RequiredRdfContext.PROPERTIES;
030import static org.jgroups.util.UUID.randomUUID;
031import static org.slf4j.LoggerFactory.getLogger;
032
033import java.io.ByteArrayInputStream;
034import java.util.HashSet;
035import java.util.Set;
036import javax.inject.Inject;
037import javax.jcr.Repository;
038import javax.jcr.RepositoryException;
039import javax.jcr.Session;
040import javax.jms.Connection;
041import javax.jms.JMSException;
042import javax.jms.Message;
043import javax.jms.MessageConsumer;
044import javax.jms.MessageListener;
045
046import org.apache.activemq.ActiveMQConnectionFactory;
047
048import org.fcrepo.kernel.api.exception.InvalidChecksumException;
049import org.fcrepo.kernel.api.models.FedoraResource;
050import org.fcrepo.kernel.api.observer.EventType;
051import org.fcrepo.kernel.api.models.Container;
052import org.fcrepo.kernel.api.services.BinaryService;
053import org.fcrepo.kernel.api.services.ContainerService;
054import org.fcrepo.kernel.modeshape.rdf.impl.DefaultIdentifierTranslator;
055
056import org.junit.After;
057import org.junit.Before;
058import org.junit.Test;
059import org.junit.runner.RunWith;
060import org.slf4j.Logger;
061import org.springframework.test.annotation.DirtiesContext;
062import org.springframework.test.context.ContextConfiguration;
063import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
064
065
066/**
067 * <p>
068 * HeadersJMSIT class.
069 * </p>
070 *
071 * @author ajs6f
072 */
073@RunWith(SpringJUnit4ClassRunner.class)
074@ContextConfiguration({ "/spring-test/headers-jms.xml", "/spring-test/repo.xml",
075    "/spring-test/eventing.xml" })
076@DirtiesContext
077public class HeadersJMSIT implements MessageListener {
078
079    /**
080     * Time to wait for a set of test messages, in milliseconds.
081     */
082    private static final long TIMEOUT = 20000;
083
084    private static final String testIngested = "/testMessageFromIngestion-" + randomUUID();
085
086    private static final String testRemoved = "/testMessageFromRemoval-" + randomUUID();
087
088    private static final String testFile = "/testMessageFromFile-" + randomUUID() + "/file1";
089
090    private static final String testMeta = "/testMessageFromMetadata-" + randomUUID();
091
092    private static final String NODE_ADDED_EVENT_TYPE
093            = REPOSITORY_NAMESPACE + EventType.NODE_ADDED;
094    private static final String NODE_REMOVED_EVENT_TYPE
095            = REPOSITORY_NAMESPACE + EventType.NODE_REMOVED;
096    private static final String PROP_ADDED_EVENT_TYPE
097            = REPOSITORY_NAMESPACE + EventType.PROPERTY_ADDED;
098    private static final String PROP_CHANGED_EVENT_TYPE
099            = REPOSITORY_NAMESPACE + EventType.PROPERTY_CHANGED;
100    private static final String PROP_REMOVED_EVENT_TYPE
101            = REPOSITORY_NAMESPACE + EventType.PROPERTY_REMOVED;
102
103    @Inject
104    private Repository repository;
105
106    @Inject
107    private BinaryService binaryService;
108
109    @Inject
110    private ContainerService containerService;
111
112    @Inject
113    private ActiveMQConnectionFactory connectionFactory;
114
115    private Connection connection;
116
117    private javax.jms.Session session;
118
119    private MessageConsumer consumer;
120
121    private volatile Set<Message> messages = new HashSet<>();
122
123    private static final Logger LOGGER = getLogger(HeadersJMSIT.class);
124
125    @Test(timeout = TIMEOUT)
126    public void testIngestion() throws RepositoryException {
127
128        LOGGER.debug("Expecting a {} event", NODE_ADDED_EVENT_TYPE);
129
130        final Session session = repository.login();
131        try {
132            containerService.findOrCreate(session, testIngested);
133            session.save();
134            awaitMessageOrFail(testIngested, NODE_ADDED_EVENT_TYPE, null);
135        } finally {
136            session.logout();
137        }
138    }
139
140    @Test(timeout = TIMEOUT)
141    public void testFileEvents() throws InvalidChecksumException, RepositoryException {
142
143        final Session session = repository.login();
144
145        try {
146            binaryService.findOrCreate(session, testFile)
147                .setContent(new ByteArrayInputStream("foo".getBytes()), "text/plain", null, null, null);
148            session.save();
149            awaitMessageOrFail(testFile, NODE_ADDED_EVENT_TYPE, HAS_SIZE.toString());
150
151            binaryService.find(session, testFile)
152                .setContent(new ByteArrayInputStream("bar".getBytes()), "text/plain", null, null, null);
153            session.save();
154            awaitMessageOrFail(testFile, PROP_CHANGED_EVENT_TYPE, HAS_SIZE.toString());
155
156            binaryService.find(session, testFile).delete();
157            session.save();
158            awaitMessageOrFail(testFile, NODE_REMOVED_EVENT_TYPE, null);
159        } finally {
160            session.logout();
161        }
162    }
163
164    @Test(timeout = TIMEOUT)
165    public void testMetadataEvents() throws RepositoryException {
166
167        final Session session = repository.login();
168        final DefaultIdentifierTranslator subjects = new DefaultIdentifierTranslator(session);
169
170        try {
171            final FedoraResource resource1 = containerService.findOrCreate(session, testMeta);
172            final String sparql1 = "insert data { <> <http://foo.com/prop> \"foo\" . }";
173            resource1.updateProperties(subjects, sparql1, resource1.getTriples(subjects, PROPERTIES));
174            session.save();
175            awaitMessageOrFail(testMeta, PROP_ADDED_EVENT_TYPE, "http://foo.com/prop");
176
177            final FedoraResource resource2 = containerService.findOrCreate(session, testMeta);
178            final String sparql2 = " delete { <> <http://foo.com/prop> \"foo\" . } "
179                + "insert { <> <http://foo.com/prop> \"bar\" . } where {}";
180            resource2.updateProperties(subjects, sparql2, resource2.getTriples(subjects, PROPERTIES));
181            session.save();
182            awaitMessageOrFail(testMeta, PROP_CHANGED_EVENT_TYPE, "http://foo.com/prop");
183        } finally {
184            session.logout();
185        }
186    }
187
188    private void awaitMessageOrFail(final String id, final String eventType, final String property) {
189        await().pollInterval(ONE_SECOND).until(() -> messages.stream().anyMatch(msg -> {
190            try {
191                return getPath(msg).equals(id) && getEventTypes(msg).contains(eventType)
192                        && (property == null || getProperties(msg).contains(property));
193            } catch (final JMSException e) {
194                throw propagate(e);
195            }
196        }));
197    }
198
199    @Test(timeout = TIMEOUT)
200    public void testRemoval() throws RepositoryException {
201
202        LOGGER.debug("Expecting a {} event", NODE_REMOVED_EVENT_TYPE);
203        final Session session = repository.login();
204        try {
205            final Container resource = containerService.findOrCreate(session, testRemoved);
206            session.save();
207            resource.delete();
208            session.save();
209            awaitMessageOrFail(testRemoved, NODE_REMOVED_EVENT_TYPE, null);
210        } finally {
211            session.logout();
212        }
213    }
214
215    @Override
216    public void onMessage(final Message message) {
217        try {
218            LOGGER.debug(
219                    "Received JMS message: {} with path: {}, timestamp: {}, event type: {}, properties: {},"
220                            + " and baseURL: {}", message.getJMSMessageID(), getPath(message), getTimestamp(message),
221                            getEventTypes(message), getProperties(message), getBaseURL(message));
222        } catch (final JMSException e) {
223            propagate(e);
224        }
225        messages.add(message);
226    }
227
228    @Before
229    public void acquireConnection() throws JMSException {
230        LOGGER.debug(this.getClass().getName() + " acquiring JMS connection.");
231        connection = connectionFactory.createConnection();
232        connection.start();
233        session = connection.createSession(false, AUTO_ACKNOWLEDGE);
234        consumer = session.createConsumer(session.createTopic("fedora"));
235        messages.clear();
236        consumer.setMessageListener(this);
237    }
238
239    @After
240    public void releaseConnection() throws JMSException {
241        // ignore any remaining or queued messages
242        consumer.setMessageListener(msg -> { });
243        // and shut the listening machinery down
244        LOGGER.debug(this.getClass().getName() + " releasing JMS connection.");
245        consumer.close();
246        session.close();
247        connection.close();
248    }
249
250    private static String getPath(final Message msg) throws JMSException {
251        final String id = msg.getStringProperty(IDENTIFIER_HEADER_NAME);
252        LOGGER.debug("Processing an event with identifier: {}", id);
253        return id;
254    }
255
256    private static String getEventTypes(final Message msg) throws JMSException {
257        final String type = msg.getStringProperty(EVENT_TYPE_HEADER_NAME);
258        LOGGER.debug("Processing an event with type: {}", type);
259        return type;
260    }
261
262    private static Long getTimestamp(final Message msg) throws JMSException {
263        return msg.getLongProperty(TIMESTAMP_HEADER_NAME);
264    }
265
266    private static String getBaseURL(final Message msg) throws JMSException {
267        return msg.getStringProperty(BASE_URL_HEADER_NAME);
268    }
269
270    private static String getProperties(final Message msg) throws JMSException {
271        return msg.getStringProperty(PROPERTIES_HEADER_NAME);
272    }
273
274}