Merge branch 'master' of git@github.com:d3scomp/JDEECo.git

jdeeco-core/src/cz/cuni/mff/d3s/deeco/annotations/PartitionedBy.java

@@ -0,0 +1,37 @@
+/*******************************************************************************
+ * Copyright 2012 Charles University in Prague
+ * 
+ * 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 cz.cuni.mff.d3s.deeco.annotations;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Used to define knowledge value the ensemble is partitioned by. 
+ * <p>
+ * The attribute <code>value</code> is dot-separated absolute path, describing
+ * the path pointing to an object in the component knowledge.
+ * </p>
+ * 
+ * @author Vladimír Matěna
+ * 
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface PartitionedBy {
+	String value();
+}

jdeeco-core/src/cz/cuni/mff/d3s/deeco/model/runtime/api/EnsembleDefinition.java

@@ -21,6 +21,7 @@
  *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getKnowledgeExchange <em>Knowledge Exchange</em>}</li>
  *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getTriggers <em>Triggers</em>}</li>
  *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getCommunicationBoundary <em>Communication Boundary</em>}</li>
+ *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getPartitionedBy <em>Partitioned By</em>}</li>
  * </ul>
  * </p>
  *
@@ -149,4 +150,30 @@ public interface EnsembleDefinition extends EObject {
 	 */
 	void setCommunicationBoundary(CommunicationBoundaryPredicate value);
 
+	/**
+	 * Returns the value of the '<em><b>Partitioned By</b></em>' attribute.
+	 * <!-- begin-user-doc -->
+	 * <p>
+	 * If the meaning of the '<em>Partitioned By</em>' attribute isn't clear,
+	 * there really should be more of a description here...
+	 * </p>
+	 * <!-- end-user-doc -->
+	 * @return the value of the '<em>Partitioned By</em>' attribute.
+	 * @see #setPartitionedBy(String)
+	 * @see cz.cuni.mff.d3s.deeco.model.runtime.meta.RuntimeMetadataPackage#getEnsembleDefinition_PartitionedBy()
+	 * @model
+	 * @generated
+	 */
+	String getPartitionedBy();
+
+	/**
+	 * Sets the value of the '{@link cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getPartitionedBy <em>Partitioned By</em>}' attribute.
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @param value the new value of the '<em>Partitioned By</em>' attribute.
+	 * @see #getPartitionedBy()
+	 * @generated
+	 */
+	void setPartitionedBy(String value);
+
 } // EnsembleDefinition

jdeeco-core/src/cz/cuni/mff/d3s/deeco/model/runtime/impl/EnsembleDefinitionImpl.java

@@ -39,6 +39,7 @@
  *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.impl.EnsembleDefinitionImpl#getKnowledgeExchange <em>Knowledge Exchange</em>}</li>
  *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.impl.EnsembleDefinitionImpl#getTriggers <em>Triggers</em>}</li>
  *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.impl.EnsembleDefinitionImpl#getCommunicationBoundary <em>Communication Boundary</em>}</li>
+ *   <li>{@link cz.cuni.mff.d3s.deeco.model.runtime.impl.EnsembleDefinitionImpl#getPartitionedBy <em>Partitioned By</em>}</li>
  * </ul>
  * </p>
  *
@@ -115,6 +116,26 @@ public class EnsembleDefinitionImpl extends MinimalEObjectImpl.Container impleme
 	 */
 	protected CommunicationBoundaryPredicate communicationBoundary = COMMUNICATION_BOUNDARY_EDEFAULT;
 
+	/**
+	 * The default value of the '{@link #getPartitionedBy() <em>Partitioned By</em>}' attribute.
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @see #getPartitionedBy()
+	 * @generated
+	 * @ordered
+	 */
+	protected static final String PARTITIONED_BY_EDEFAULT = null;
+
+	/**
+	 * The cached value of the '{@link #getPartitionedBy() <em>Partitioned By</em>}' attribute.
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @see #getPartitionedBy()
+	 * @generated
+	 * @ordered
+	 */
+	protected String partitionedBy = PARTITIONED_BY_EDEFAULT;
+
 	/**
 	 * <!-- begin-user-doc -->
 	 * <!-- end-user-doc -->
@@ -274,6 +295,27 @@ public void setCommunicationBoundary(CommunicationBoundaryPredicate newCommunica
 			eNotify(new ENotificationImpl(this, Notification.SET, RuntimeMetadataPackage.ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY, oldCommunicationBoundary, communicationBoundary));
 	}
 
+	/**
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @generated
+	 */
+	public String getPartitionedBy() {
+		return partitionedBy;
+	}
+
+	/**
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @generated
+	 */
+	public void setPartitionedBy(String newPartitionedBy) {
+		String oldPartitionedBy = partitionedBy;
+		partitionedBy = newPartitionedBy;
+		if (eNotificationRequired())
+			eNotify(new ENotificationImpl(this, Notification.SET, RuntimeMetadataPackage.ENSEMBLE_DEFINITION__PARTITIONED_BY, oldPartitionedBy, partitionedBy));
+	}
+
 	/**
 	 * <!-- begin-user-doc -->
 	 * <!-- end-user-doc -->
@@ -310,6 +352,8 @@ public Object eGet(int featureID, boolean resolve, boolean coreType) {
 				return getTriggers();
 			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY:
 				return getCommunicationBoundary();
+			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__PARTITIONED_BY:
+				return getPartitionedBy();
 		}
 		return super.eGet(featureID, resolve, coreType);
 	}
@@ -339,6 +383,9 @@ public void eSet(int featureID, Object newValue) {
 			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY:
 				setCommunicationBoundary((CommunicationBoundaryPredicate)newValue);
 				return;
+			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__PARTITIONED_BY:
+				setPartitionedBy((String)newValue);
+				return;
 		}
 		super.eSet(featureID, newValue);
 	}
@@ -366,6 +413,9 @@ public void eUnset(int featureID) {
 			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY:
 				setCommunicationBoundary(COMMUNICATION_BOUNDARY_EDEFAULT);
 				return;
+			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__PARTITIONED_BY:
+				setPartitionedBy(PARTITIONED_BY_EDEFAULT);
+				return;
 		}
 		super.eUnset(featureID);
 	}
@@ -388,6 +438,8 @@ public boolean eIsSet(int featureID) {
 				return triggers != null && !triggers.isEmpty();
 			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY:
 				return COMMUNICATION_BOUNDARY_EDEFAULT == null ? communicationBoundary != null : !COMMUNICATION_BOUNDARY_EDEFAULT.equals(communicationBoundary);
+			case RuntimeMetadataPackage.ENSEMBLE_DEFINITION__PARTITIONED_BY:
+				return PARTITIONED_BY_EDEFAULT == null ? partitionedBy != null : !PARTITIONED_BY_EDEFAULT.equals(partitionedBy);
 		}
 		return super.eIsSet(featureID);
 	}
@@ -406,6 +458,8 @@ public String toString() {
 		result.append(name);
 		result.append(", communicationBoundary: ");
 		result.append(communicationBoundary);
+		result.append(", partitionedBy: ");
+		result.append(partitionedBy);
 		result.append(')');
 		return result.toString();
 	}

jdeeco-core/src/cz/cuni/mff/d3s/deeco/model/runtime/impl/RuntimeMetadataPackageImpl.java

@@ -575,6 +575,15 @@ public EAttribute getEnsembleDefinition_CommunicationBoundary() {
 		return (EAttribute)ensembleDefinitionEClass.getEStructuralFeatures().get(4);
 	}
 
+	/**
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @generated
+	 */
+	public EAttribute getEnsembleDefinition_PartitionedBy() {
+		return (EAttribute)ensembleDefinitionEClass.getEStructuralFeatures().get(5);
+	}
+
 	/**
 	 * <!-- begin-user-doc -->
 	 * <!-- end-user-doc -->
@@ -905,6 +914,7 @@ public void createPackageContents() {
 		createEReference(ensembleDefinitionEClass, ENSEMBLE_DEFINITION__KNOWLEDGE_EXCHANGE);
 		createEReference(ensembleDefinitionEClass, ENSEMBLE_DEFINITION__TRIGGERS);
 		createEAttribute(ensembleDefinitionEClass, ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY);
+		createEAttribute(ensembleDefinitionEClass, ENSEMBLE_DEFINITION__PARTITIONED_BY);
 
 		conditionEClass = createEClass(CONDITION);
 
@@ -1031,6 +1041,7 @@ public void initializePackageContents() {
 		initEReference(getEnsembleDefinition_KnowledgeExchange(), this.getExchange(), null, "knowledgeExchange", null, 1, 1, EnsembleDefinition.class, !IS_TRANSIENT, !IS_VOLATILE, IS_CHANGEABLE, IS_COMPOSITE, !IS_RESOLVE_PROXIES, !IS_UNSETTABLE, IS_UNIQUE, !IS_DERIVED, IS_ORDERED);
 		initEReference(getEnsembleDefinition_Triggers(), this.getTrigger(), null, "triggers", null, 0, -1, EnsembleDefinition.class, !IS_TRANSIENT, !IS_VOLATILE, IS_CHANGEABLE, IS_COMPOSITE, !IS_RESOLVE_PROXIES, !IS_UNSETTABLE, IS_UNIQUE, !IS_DERIVED, IS_ORDERED);
 		initEAttribute(getEnsembleDefinition_CommunicationBoundary(), this.getCommunicationBoundary(), "communicationBoundary", null, 0, 1, EnsembleDefinition.class, !IS_TRANSIENT, !IS_VOLATILE, IS_CHANGEABLE, !IS_UNSETTABLE, !IS_ID, IS_UNIQUE, !IS_DERIVED, IS_ORDERED);
+		initEAttribute(getEnsembleDefinition_PartitionedBy(), ecorePackage.getEString(), "partitionedBy", null, 0, 1, EnsembleDefinition.class, !IS_TRANSIENT, !IS_VOLATILE, IS_CHANGEABLE, !IS_UNSETTABLE, !IS_ID, IS_UNIQUE, !IS_DERIVED, IS_ORDERED);
 
 		initEClass(conditionEClass, Condition.class, "Condition", !IS_ABSTRACT, !IS_INTERFACE, IS_GENERATED_INSTANCE_CLASS);
 

jdeeco-core/src/cz/cuni/mff/d3s/deeco/model/runtime/meta/RuntimeMetadataPackage.java

@@ -518,14 +518,23 @@ public interface RuntimeMetadataPackage extends EPackage {
 	 */
 	int ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY = 4;
 
+	/**
+	 * The feature id for the '<em><b>Partitioned By</b></em>' attribute.
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @generated
+	 * @ordered
+	 */
+	int ENSEMBLE_DEFINITION__PARTITIONED_BY = 5;
+
 	/**
 	 * The number of structural features of the '<em>Ensemble Definition</em>' class.
 	 * <!-- begin-user-doc -->
 	 * <!-- end-user-doc -->
 	 * @generated
 	 * @ordered
 	 */
-	int ENSEMBLE_DEFINITION_FEATURE_COUNT = 5;
+	int ENSEMBLE_DEFINITION_FEATURE_COUNT = 6;
 
 	/**
 	 * The number of operations of the '<em>Ensemble Definition</em>' class.
@@ -1389,6 +1398,17 @@ public interface RuntimeMetadataPackage extends EPackage {
 	 */
 	EAttribute getEnsembleDefinition_CommunicationBoundary();
 
+	/**
+	 * Returns the meta object for the attribute '{@link cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getPartitionedBy <em>Partitioned By</em>}'.
+	 * <!-- begin-user-doc -->
+	 * <!-- end-user-doc -->
+	 * @return the meta object for the attribute '<em>Partitioned By</em>'.
+	 * @see cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition#getPartitionedBy()
+	 * @see #getEnsembleDefinition()
+	 * @generated
+	 */
+	EAttribute getEnsembleDefinition_PartitionedBy();
+
 	/**
 	 * Returns the meta object for class '{@link cz.cuni.mff.d3s.deeco.model.runtime.api.Condition <em>Condition</em>}'.
 	 * <!-- begin-user-doc -->
@@ -1998,6 +2018,14 @@ interface Literals {
 		 */
 		EAttribute ENSEMBLE_DEFINITION__COMMUNICATION_BOUNDARY = eINSTANCE.getEnsembleDefinition_CommunicationBoundary();
 
+		/**
+		 * The meta object literal for the '<em><b>Partitioned By</b></em>' attribute feature.
+		 * <!-- begin-user-doc -->
+		 * <!-- end-user-doc -->
+		 * @generated
+		 */
+		EAttribute ENSEMBLE_DEFINITION__PARTITIONED_BY = eINSTANCE.getEnsembleDefinition_PartitionedBy();
+
 		/**
 		 * The meta object literal for the '{@link cz.cuni.mff.d3s.deeco.model.runtime.impl.ConditionImpl <em>Condition</em>}' class.
 		 * <!-- begin-user-doc -->

jdeeco-core/src/cz/cuni/mff/d3s/deeco/network/IPGossipStrategy.java

@@ -0,0 +1,9 @@
+package cz.cuni.mff.d3s.deeco.network;
+
+import java.util.Collection;
+
+import cz.cuni.mff.d3s.deeco.knowledge.KnowledgeManager;
+
+public interface IPGossipStrategy {
+	public Collection<String> getRecipients(KnowledgeData data, KnowledgeManager sender);
+}

jdeeco-core/src/cz/cuni/mff/d3s/deeco/network/KnowledgeDataManager.java

@@ -19,7 +19,6 @@
 import cz.cuni.mff.d3s.deeco.logging.Log;
 import cz.cuni.mff.d3s.deeco.model.runtime.api.EnsembleDefinition;
 import cz.cuni.mff.d3s.deeco.model.runtime.api.KnowledgePath;
-import cz.cuni.mff.d3s.deeco.model.runtime.api.PathNodeField;
 import cz.cuni.mff.d3s.deeco.model.runtime.custom.RuntimeMetadataFactoryExt;
 import cz.cuni.mff.d3s.deeco.model.runtime.meta.RuntimeMetadataFactory;
 import cz.cuni.mff.d3s.deeco.scheduler.CurrentTimeProvider;
@@ -98,10 +97,7 @@ public class KnowledgeDataManager implements KnowledgeDataReceiver,
 	private final int maxRebroadcastDelay;	
 	private final int ipDelay;
 
-
-	//TODO This needs to be changed
-	private final Collection<DirectRecipientSelector> recipientSelectors;
-	private final DirectGossipStrategy directGossipStrategy;
+	private final IPGossipStrategy ipGossipStrategy;
 
 
 	/**
@@ -118,8 +114,7 @@ public KnowledgeDataManager(
 			List<EnsembleDefinition> ensembleDefinitions,
 			String host,
 			Scheduler scheduler,
-			Collection<DirectRecipientSelector> recipientSelectors,
-			DirectGossipStrategy directGossipStrategy) {
+			IPGossipStrategy ipGossipStrategy) {
 		this.host = host;		
 		this.scheduler = scheduler;
 		this.timeProvider = scheduler;
@@ -128,8 +123,7 @@ public KnowledgeDataManager(
 		this.ensembleDefinitions = ensembleDefinitions;
 		this.localVersion = 0;
 		this.replicaMetadata = new HashMap<>();
-		this.recipientSelectors = recipientSelectors;
-		this.directGossipStrategy = directGossipStrategy;
+		this.ipGossipStrategy = ipGossipStrategy;
 
 		dataToRebroadcastOverMANET = new HashMap<>();
 		dataToRebroadcastOverIP = new HashMap<>();
@@ -210,20 +204,16 @@ public void publish() {
 	}
 
 	private void sendDirect(List<KnowledgeData> data) {
-		if (recipientSelectors != null && !recipientSelectors.isEmpty()) {
-			//Publishing to IP
-			Collection<String> recipients;
-			//For IP part we are using individual publishing only
-			for (KnowledgeData kd : data) {
-				recipients = getRecipients(kd, getNodeKnowledge());
-				for (String recipient: recipients) {					
-					logPublish(data, recipient);
-					knowledgeDataSender.sendKnowledgeData(Arrays.asList(kd), recipient);					
-				}
+		// Publishing to IP
+		// For IP part we are using individual publishing only
+		for (KnowledgeData kd : data) {
+			Collection<String> recipients = ipGossipStrategy.getRecipients(kd, getNodeKnowledge());
+			for (String recipient: recipients) {					
+				logPublish(data, recipient);
+				knowledgeDataSender.sendKnowledgeData(Arrays.asList(kd), recipient);					
 			}
 		}
-	}
-
+	}	
 
 	@Override
 	public void rebroacast(KnowledgeMetaData metadata, NICType nicType) {
@@ -375,21 +365,6 @@ private boolean isInSomeBoundary(KnowledgeData data, KnowledgeManager sender) {
 		return isInSomeBoundary;
 	}
 
-	private Collection<String> getRecipients(KnowledgeData data, KnowledgeManager sender) {
-		List<String> result = new LinkedList<>();
-		for (DirectRecipientSelector selector: recipientSelectors) {
-			result.addAll(selector.getRecipients(data, sender));
-		}
-
-		// filter the sender and owner of the data
-		// remove all
-		while (result.remove(data.getMetaData().componentId));
-		// remove all
-		while(result.remove(sender.getId()));
-		return directGossipStrategy.filterRecipients(result);
-	}
-
-
 //	private boolean satisfiesGossipCondition(KnowledgeMetaData kmd) {
 //		// rssi < 0 means received from IP
 //		if (kmd.rssi < 0) {