bundles/addons/org.eclipse.apogy.addons.ros/src-gen-custom/org/eclipse/apogy/addons/ros/impl/ROSListenerCustomImpl.java

/*******************************************************************************
 * Copyright (c) 2018 Agence spatiale canadienne / Canadian Space Agency 
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *     Pierre Allard, 
 *     Regent L'Archeveque,
 *     Sebastien Gemme - initial API and implementation
 * SPDX-License-Identifier: EPL-1.0
 *     
 *******************************************************************************/
package org.eclipse.apogy.addons.ros.impl;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.eclipse.apogy.addons.ros.ROSListenerState;
import org.eclipse.apogy.addons.ros.ROSNode;
import org.eclipse.apogy.addons.ros.utilities.AsynchronousShutdowner;
import org.ros.internal.message.Message;
import org.ros.message.MessageListener;
import org.ros.node.ConnectedNode;
import org.ros.node.topic.Subscriber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ROSListenerCustomImpl<T extends Message> extends ROSListenerImpl<T> {
	private static final Logger Logger = LoggerFactory.getLogger(ROSListenerImpl.class);

	private static final int MAX_WAIT_CONNECTION = 1000; // millisecond
	private final ReentrantLock lock;
	private final Condition condition;

	protected ROSListenerCustomImpl() {
		super();
		this.lock = new ReentrantLock();
		this.condition = this.lock.newCondition();
	}

	@Override
	public boolean startWithLocks(ROSNode node) throws Exception {
		boolean timeElapsed = false;
		this.lock.lock();
		try {
			setNode(node);
			connectSubscriber(node.getConnectedNode());

			while (!timeElapsed && !isRunning()) {
				try {
					timeElapsed = !this.condition.await(MAX_WAIT_CONNECTION, TimeUnit.MILLISECONDS);
				} catch (InterruptedException e) {
					Logger.error(e.getMessage(), e);
				}
			}
			if (timeElapsed) {
				throw new Exception("Cannot connect to the topic " + getTopicName() + " (connection timeout)");
			}
		} finally {
			this.lock.unlock();
		}
		return isRunning();
	}

	@Override
	public void start(ROSNode node) throws Exception {
		setNode(node);
		connectSubscriber(node.getConnectedNode());
	}

	@Override
	public void connectSubscriber(ConnectedNode connectedNode) {
		Subscriber<T> rosSubscriber = connectedNode.newSubscriber(getTopicName(), getMessageType());
		rosSubscriber.addMessageListener(getMessageListener());
		rosSubscriber.addMessageListener(new ConnectionStateListener());
		setRosSubscriber(rosSubscriber);
	}

	@Override
	public void stop() {
		try {
			AsynchronousShutdowner.add(getRosSubscriber());
		} catch (Exception e) {
		}

		setListenerState(ROSListenerState.STOPPED);
		setRosSubscriber(null);
		setRunning(false);
	}

	private class ConnectionStateListener implements MessageListener<T> {

		@Override
		public void onNewMessage(T message) {
			if (!isRunning()) {
				ROSListenerCustomImpl.this.lock.lock();
				try {
					Logger.info("onNewMessage()Listener for the topic " + getTopicName() + " is running.");
					setRunning(true);
					ROSListenerCustomImpl.this.condition.signal();
				} finally {
					ROSListenerCustomImpl.this.lock.unlock();
				}
			}
		}
	}

} // ROSListenerImpl