| /******************************************************************************* |
| * Copyright (c) 2004, 2007 Boeing. |
| * 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: |
| * Boeing - initial API and implementation |
| *******************************************************************************/ |
| package org.eclipse.osee.ote.messaging.dds.entity; |
| |
| import java.util.Collection; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import org.eclipse.osee.ote.messaging.dds.DataStoreItem; |
| import org.eclipse.osee.ote.messaging.dds.NotImplementedException; |
| import org.eclipse.osee.ote.messaging.dds.ReturnCode; |
| import org.eclipse.osee.ote.messaging.dds.StatusKind; |
| import org.eclipse.osee.ote.messaging.dds.listener.DataReaderListener; |
| import org.eclipse.osee.ote.messaging.dds.listener.Listener; |
| import org.eclipse.osee.ote.messaging.dds.listener.SubscriberListener; |
| import org.eclipse.osee.ote.messaging.dds.service.TopicDescription; |
| import org.eclipse.osee.ote.messaging.dds.status.SampleLostStatus; |
| |
| /** |
| * Provides functionality for applications to create <code>DataReader</code>'s to receive the appropriate published |
| * data. Maintains the data which is made available to those <code>DataReader</code> 's. |
| * |
| * @author Robert A. Fisher |
| * @author David Diepenbrock |
| */ |
| public class Subscriber extends DomainEntity implements EntityFactory { |
| |
| private final ConcurrentHashMap<TopicDescription, CopyOnWriteArrayList<DataReader>> topicMap = |
| new ConcurrentHashMap<TopicDescription, CopyOnWriteArrayList<DataReader>>(512); |
| private final DomainParticipant participant; |
| private final ExecutorService threadService; |
| |
| /** |
| * Constructor for <code>Subscriber</code> with the provided listener attached, and enabled status set as passed. |
| * This constructor is only visible to the DDS system, applications should use |
| * {@link DomainParticipant#createSubscriber(SubscriberListener)}to create a <code>Subscriber</code>. |
| * |
| * @param participant The participant creating this <code>Subscriber</code>. This is also used as the parentEntity |
| * for enabling purposes. |
| * @param enabled If <b>true </b>, the created <code>Subscriber</code> will be enabled iff the parentEntity is |
| * enabled. |
| * @param listener The listener to attach to the created subscriber. |
| */ |
| Subscriber(DomainParticipant participant, boolean enabled, Listener listener) { |
| super(enabled, listener, participant); |
| this.participant = participant; |
| threadService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1); |
| } |
| |
| /** |
| * Gets the attached listener. |
| * |
| * @return The <code>SubscriberListener</code> attached to this. |
| * @see Entity#getBaseListener() |
| */ |
| public SubscriberListener getListener() { |
| return (SubscriberListener) super.getBaseListener(); |
| } |
| |
| /** |
| * Sets the <code>SubscriberListener</code> attached to this. If a listener is already set, this will replace it. |
| * |
| * @see Entity#setBaseListener(Listener, StatusKind) |
| */ |
| public ReturnCode setListener(SubscriberListener listener, StatusKind mask) { |
| return super.setBaseListener(listener, mask); |
| } |
| |
| /** |
| * Creates a <code>DataReader</code> for the given <code>Topic</code> and adds it to this <code>Subscriber</code>. |
| * |
| * @param topicDescription The <code>Topic</code> associated with the <code>DataReader</code> to be created. |
| * @param listener The <code>DataReaderListener</code> to be attached to the created <code>DataReader</code>. |
| * @return The <code>DataReader</code> created, or null if it could not be created. |
| */ |
| public DataReader createDataReader(TopicDescription topicDescription, DataReaderListener listener) { |
| DataReader dataReader = null; |
| |
| if (true != topicDescription instanceof Topic) { |
| return null; |
| } |
| |
| //TUNE - get the classloader & constructor during Topic creation instead of here |
| try { |
| // Class readerClass = topic.getTypeSignature().getClassLoader().loadClass(topic.getTypeSignature().getReaderName()); |
| // |
| // Constructor constructor = readerClass.getConstructor(new Class[] {TopicDescription.class, Subscriber.class, Boolean.class, DataReaderListener.class, |
| // EntityFactory.class}); |
| // |
| // dataReader = (DataReader) constructor.newInstance(new Object[] {topicDescription, this, new Boolean(this.isEnabled()), listener, this}); |
| |
| dataReader = new DataReader(topicDescription, this, new Boolean(this.isEnabled()), listener, this); |
| } |
| // catch (InstantiationException ex) { |
| // ex.printStackTrace(); |
| // } |
| // catch (IllegalAccessException ex) { |
| // ex.printStackTrace(); |
| // } |
| // catch (ClassNotFoundException ex) { |
| // ex.printStackTrace(); |
| // } |
| catch (IllegalArgumentException ex) { |
| ex.printStackTrace(); |
| } catch (SecurityException ex) { |
| ex.printStackTrace(); |
| } |
| // catch (InvocationTargetException ex) { |
| // ex.printStackTrace(); |
| // } |
| // catch (NoSuchMethodException ex) { |
| // ex.printStackTrace(); |
| // } |
| catch (OutOfMemoryError er) { |
| er.printStackTrace(); |
| } |
| |
| // Only keep this writer if it was successfully created. |
| if (dataReader != null) { |
| CopyOnWriteArrayList<DataReader> readers = topicMap.get(topicDescription); |
| if (readers == null) { |
| readers = new CopyOnWriteArrayList<>(); |
| topicMap.put(topicDescription, readers); |
| } |
| readers.add(dataReader); |
| } |
| return dataReader; |
| } |
| |
| /** |
| * Removes the <code>DataReader</code> from this <code>Subscriber</code>. If the reader was already deleted, or was |
| * not created by this <code>Subscriber</code>, an error is returned. |
| * |
| * @param dataReader The reader to delete. |
| * @return {@link ReturnCode#OK}if the reader was successfully deleted, otherwise {@link ReturnCode#NOT_ENABLED}if |
| * this <code>Subscriber</code> is not enabled, or {@link ReturnCode#ALREADY_DELETED}if the reader has already been |
| * deleted, or {@link ReturnCode#PRECONDITION_NOT_MET}if dataReader was not created by this <code>Subscriber</code>. |
| */ |
| public ReturnCode deleteDataReader(DataReader dataReader) { |
| |
| // Check that the Entity is enabled before proceeding (See description of enable on the Entity object in the DDS spec) |
| if (!isEnabled()) { |
| return ReturnCode.NOT_ENABLED; |
| } |
| |
| // Check that the writer is not already marked as deleted (in case others kept a reference to it |
| if (dataReader.isDeleted()) { |
| return ReturnCode.ALREADY_DELETED; |
| } |
| |
| // boolean found = dataReaders.remove(dataReader); |
| |
| if (topicMap.remove(dataReader.getTopicDescription()) != null) { |
| dataReader.markAsDeleted(); |
| return ReturnCode.OK; |
| } else { |
| return ReturnCode.PRECONDITION_NOT_MET; |
| } |
| |
| } |
| |
| /** |
| * This method is here for future functionality that is described in the DDS specification but has not been |
| * implemented or used. |
| */ |
| public DataReader lookupDataReader(String topicName) { |
| // UNSURE This method has not been implemented, but is called out in the spec |
| if (true) { |
| throw new NotImplementedException(); |
| } |
| return null; |
| } |
| |
| /** |
| * This method is here for future functionality that is described in the DDS specification but has not been |
| * implemented or used. |
| */ |
| public ReturnCode beginAccess() { |
| // DONT_NEED This method has not been implemented, but is called out in the spec |
| if (true) { |
| throw new NotImplementedException(); |
| } |
| |
| // Check that the Entity is enabled before proceeding (See description of enable on the Entity object in the DDS spec) |
| if (!isEnabled()) { |
| return ReturnCode.NOT_ENABLED; |
| } |
| |
| return ReturnCode.ERROR; |
| } |
| |
| /** |
| * This method is here for future functionality that is described in the DDS specification but has not been |
| * implemented or used. |
| */ |
| public ReturnCode endAccess() { |
| // DONT_NEED This method has not been implemented, but is called out in the spec |
| if (true) { |
| throw new NotImplementedException(); |
| } |
| |
| // Check that the Entity is enabled before proceeding (See description of enable on the Entity object in the DDS spec) |
| if (!isEnabled()) { |
| return ReturnCode.NOT_ENABLED; |
| } |
| |
| return ReturnCode.ERROR; |
| } |
| |
| /** |
| * This method is here for future functionality that is described in the DDS specification but has not been |
| * implemented or used. |
| */ |
| public ReturnCode getDataReaders(Collection<DataReader> dataReaders, Collection<?> sampleStateKind, Collection<?> viewStateKind, Collection<?> instanceStateKind) { |
| // UNSURE This method has not been implemented, but is called out in the spec |
| if (true) { |
| throw new NotImplementedException(); |
| } |
| |
| // Check that the Entity is enabled before proceeding (See description of enable on the Entity object in the DDS spec) |
| if (!isEnabled()) { |
| return ReturnCode.NOT_ENABLED; |
| } |
| |
| return ReturnCode.ERROR; |
| } |
| |
| /** |
| * This method is here for future functionality that is described in the DDS specification but has not been |
| * implemented or used. |
| */ |
| public void notifyDataReaders() { |
| // UNSURE This method has not been implemented, but is called out in the spec |
| if (true) { |
| throw new NotImplementedException(); |
| } |
| } |
| |
| /** |
| * This method is here for future functionality that is described in the DDS specification but has not been |
| * implemented or used. |
| */ |
| public SampleLostStatus getSampleLostStatus() { |
| // UNSURE This method has not been implemented, but is called out in the spec |
| if (true) { |
| throw new NotImplementedException(); |
| } |
| return null; |
| } |
| |
| /** |
| * @return the <code>DomainParticipant</code> to which this belongs. |
| */ |
| public DomainParticipant getParticipant() { |
| // NOTE this method is part of the DDS spec. |
| return participant; |
| } |
| |
| /** |
| * Deletes all of the <code>DataReader</code> objects currently in this <code>Subscriber</code>. |
| * |
| * @return {@link ReturnCode#OK}if the readers were successfully deleted, otherwise {@link ReturnCode#NOT_ENABLED}if |
| * this <code>Subscriber</code> is not enabled. |
| */ |
| public ReturnCode deleteContainedEntities() { |
| /* |
| * PARTIAL per the DDS spec, this should also delete the ReadCondition and QueryCondition objects stored in the |
| * <code> DataReader </code> 's, however because we haven't implemented the conditions they are not deleted here. |
| */ |
| |
| // Check that the Entity is enabled before proceeding (See description of enable on the Entity object in the DDS spec) |
| if (!isEnabled()) { |
| return ReturnCode.NOT_ENABLED; |
| } |
| |
| for (CopyOnWriteArrayList<DataReader> readers : topicMap.values()) { |
| for (DataReader reader : readers) { |
| reader.markAsDeleted(); |
| } |
| readers.clear(); |
| } |
| topicMap.clear(); |
| return ReturnCode.OK; |
| } |
| |
| /** |
| * @return <b>true </b> iff this <code>Subscriber</code> currently has any data readers. |
| */ |
| public boolean hasDataReaders() { |
| return !topicMap.isEmpty(); |
| } |
| |
| /** |
| * Processes new data received. If this Subscriber has at least one DataReader who is interested in this data, and |
| * the <code>Subscriber</code> has an attached listener, notify it via |
| * {@link SubscriberListener#onDataOnReaders(CopyOfSubscriber)}. Otherwise, if there is no |
| * <code>SubscriberListener</code> then notify all of the interested <code>DataReader</code>'s via |
| * {@link DataReaderListener#onDataAvailable(DataReader)}. The data is stored so that it is available for a reader to |
| * read or take. |
| * |
| * @param dataStoreItem The newly published data. |
| */ |
| void processNewData(final DataStoreItem dataStoreItem) { // This has package scope since it is a system-level type call |
| final Collection<DataReader> readers = topicMap.get(dataStoreItem.getTheTopicDescription()); |
| if (readers != null && !readers.isEmpty()) { |
| |
| // SPEC NOTE: The following listener logic is based on paragraph "Listener access to Read Communication Status" |
| // in the DDS specification. |
| |
| // Invoke the SubscriberListener if available |
| SubscriberListener listener = getListener(); |
| if (listener != null) { |
| listener.onDataOnReaders(this); |
| |
| // Otherwise, invoke the DataReaderListeners |
| } else { |
| // Check all DataReader objects for listeners |
| for (final DataReader reader : readers) { |
| reader.processNewData(dataStoreItem); |
| /* |
| * threadService.submit(new Runnable() { public void run() { reader.processNewData(dataStoreItem); } }); |
| */ |
| |
| } |
| } |
| } |
| } |
| } |
