RemoteEventPublisher.java

Go to the documentation of this file.

/*
 * Autopsy Forensic Browser
 *
 * Copyright 2011-2018 Basis Technology Corp.
 * Contact: carrier <at> sleuthkit <dot> org
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sleuthkit.autopsy.events;

import java.net.URISyntaxException;
import java.util.logging.Level;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.sleuthkit.autopsy.coreutils.Logger;

@ThreadSafe
final class RemoteEventPublisher {

    private static final Logger logger = Logger.getLogger(RemoteEventPublisher.class.getName());
    private static final String ALL_MESSAGE_SELECTOR = "All"; //NON-NLS
    private final LocalEventPublisher localPublisher; // LocalEventPublisher is thread-safe
    @GuardedBy("this")
    private final Connection connection;
    @GuardedBy("this")
    private final Session session;
    @GuardedBy("this")
    private final MessageProducer producer;
    @GuardedBy("this")
    private final MessageConsumer consumer;

    RemoteEventPublisher(String eventChannelName, LocalEventPublisher localPublisher, MessageServiceConnectionInfo info) throws URISyntaxException, JMSException {
        try {
            this.localPublisher = localPublisher;
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(info.getUserName(), info.getPassword(), info.getURI());
            connection = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(eventChannelName);
            producer = session.createProducer(topic);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            consumer = session.createConsumer(topic, "events = '" + ALL_MESSAGE_SELECTOR + "'", true); //NON-NLS
            MessageReceiver receiver = new MessageReceiver();
            consumer.setMessageListener(receiver);
        } catch (URISyntaxException | JMSException ex) {
            logger.log(Level.SEVERE, "Failed to connect to event channel", ex); //NON-NLS
            try {
                stop();
            } catch (JMSException ignored) {
            }
            throw ex;
        }
    }

    synchronized void stop() throws JMSException {
        if (null != producer) {
            producer.close();
        }
        if (null != consumer) {
            consumer.close();
        }
        if (null != session) {
            session.close();
        }
        if (null != connection) {
            connection.close();
        }
    }

    synchronized void publish(AutopsyEvent event) throws JMSException {
        ObjectMessage message = session.createObjectMessage();
        message.setStringProperty("events", ALL_MESSAGE_SELECTOR); //NON-NLS
        message.setObject(event);
        producer.send(message);
    }

    private final class MessageReceiver implements MessageListener {

        @Override
        public void onMessage(Message message) {
            try {
                if (message instanceof ObjectMessage) {
                    ObjectMessage objectMessage = (ObjectMessage) message;
                    Object object = objectMessage.getObject();
                    if (object instanceof AutopsyEvent) {
                        AutopsyEvent event = (AutopsyEvent) object;
                        event.setSourceType(AutopsyEvent.SourceType.REMOTE);
                        localPublisher.publish(event);
                    }
                }
            } catch (JMSException ex) {
                logger.log(Level.SEVERE, "Error receiving message", ex); //NON-NLS
            } catch (Throwable ex) {
                // Exception firewall.
                logger.log(Level.SEVERE, "Unexpected error receiving message", ex); //NON-NLS                
            }
        }
    }
}